get environment set in C/C++ side

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "PyCXpress"
-version = "0.0.7"
+version = "0.0.8"
 description = "PyCXpress is a high-performance hybrid framework that seamlessly integrates Python and C++ to harness the flexibility of Python and the speed of C++ for efficient and expressive computation, particularly in the realm of deep learning and numerical computing."
 readme = "README.md"
 authors = ["chaoqing <[email protected]>"]

src/PyCXpress/example/main.cpp

@@ -24,6 +24,8 @@ void show_test(pcx::Model &model) {
     memcpy(model.set_buffer("new_2d_shape", {2}).first, shape.data(),
            shape.size() * sizeof(uint8_t));
 
+    std::string status = "model ready to run";
+    setenv("PYCXPRESS_STATUS", status.c_str(), 1);
     model.run();
 
     // test retrieve output tensor
@@ -54,12 +56,14 @@ void show_test(pcx::Model &model) {
 }
 
 int main(int argc, char *argv[]) {
-    auto &python     = utils::Singleton<pcx::PythonInterpreter>::Instance();
+    auto &python = utils::Singleton<pcx::PythonInterpreter>::Instance();
+
+    std::string status = "model ready to initialized.";
+    setenv("PYCXPRESS_STATUS", status.c_str(), 1);
     auto &model0     = python.create_model("model.Model");
     auto &model1     = python.create_model("model.Model", "odd");
     int   loop_times = 3;
 
-
     while (loop_times--) {
         std::cout << "looping " << loop_times << std::endl;
         if (loop_times % 2 == 0) {

src/PyCXpress/example/model.py

@@ -19,6 +19,7 @@
     convert_to_spec_tuple,
     pycxpress_debugger,
 )
+from PyCXpress.utils import getenv
 
 
 def show(a: np.ndarray):
@@ -76,6 +77,7 @@ def __init__(self):
 
     def initialize(self):
         self.input, self.output = InputDataSet(), OutputDataSet()
+        print("current status: ", getenv("PYCXPRESS_STATUS", ""))
 
         return (
             self.input,
@@ -84,6 +86,7 @@ def initialize(self):
         )
 
     def run(self):
+        print("current status: ", getenv("PYCXPRESS_STATUS", ""))
         self.model(self.input, self.output)
 
     @staticmethod

src/PyCXpress/utils.py

@@ -1,13 +1,60 @@
 from typing import Any, Dict, Tuple
 
 import logging
+import os
 
 import numpy as np
 from numpy.typing import DTypeLike
 
 logger = logging.getLogger("PyCXpress")
 
 
+# mypy: disable-error-code="attr-defined"
+def _refresh_environ():
+    """https://discuss.python.org/t/method-to-refresh-os-environ/54774/11"""
+    import ctypes
+    import sys
+
+    if sys.platform == "linux":
+
+        def _get_env_array(*, lib=ctypes.CDLL(None)):
+            return ctypes.POINTER(ctypes.c_char_p).in_dll(lib, "environ")
+
+    elif sys.platform == "win32":
+
+        def _get_env_array(*, lib=ctypes.CDLL("ucrtbase")):
+            p = ctypes.CFUNCTYPE(ctypes.POINTER(ctypes.POINTER(ctypes.c_wchar_p)))
+            return p(("__p__wenviron", lib))()[0]
+
+    else:
+        raise ImportError
+
+    uppercase_names = sys.platform == "win32"
+    _env_array = _get_env_array()
+    if isinstance(_env_array[0], bytes):
+        equals = b"="
+    else:
+        equals = "="  # type: ignore
+    c_environ = {}
+    for entry in _env_array:
+        if entry is None:
+            break
+        name, value = entry.split(equals, 1)
+        if uppercase_names:
+            c_environ[name.upper()] = value
+        else:
+            c_environ[name] = value
+
+    os.environ._data.clear()
+    os.environ._data.update(c_environ)
+
+
+def getenv(key, default=None):
+    """Just like getenv but always do refresh to obtain the latest process environments"""
+    _refresh_environ()
+    return os.getenv(key, default)
+
+
 def get_c_type(t: DTypeLike) -> Tuple[str, int]:
     dtype = np.dtype(t)
     relation = {