-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathnative_core.py
More file actions
96 lines (81 loc) · 3.05 KB
/
Copy pathnative_core.py
File metadata and controls
96 lines (81 loc) · 3.05 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
"""Optional loader for the Windows x64 ThermalSim CPU core."""
import ctypes
import os
from pathlib import Path
import numpy as np
class NativeCore:
"""Thin stable C-ABI wrapper around ``thermalsim_core.dll``."""
def __init__(self, path):
self.path = str(path)
self.library = ctypes.CDLL(self.path)
double_pointer = ctypes.POINTER(ctypes.c_double)
self.library.thermalsim_core_version.argtypes = []
self.library.thermalsim_core_version.restype = ctypes.c_int
self.library.thermalsim_apply_structured.argtypes = [
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
double_pointer,
double_pointer,
double_pointer,
double_pointer,
double_pointer,
double_pointer,
]
self.library.thermalsim_apply_structured.restype = ctypes.c_int
self.version = int(self.library.thermalsim_core_version())
if self.version < 1:
raise RuntimeError("Unsupported ThermalSim native-core version.")
@staticmethod
def _pointer(array):
return array.ctypes.data_as(ctypes.POINTER(ctypes.c_double))
def apply_structured(self, layer_count, rows, cols, gx, gy, gz, boundary, vector):
"""Apply the structured operator and return a new result array."""
arrays = [
np.ascontiguousarray(item, dtype=np.float64)
for item in (gx, gy, gz, boundary, vector)
]
gx_array, gy_array, gz_array, boundary_array, vector_array = arrays
result = np.empty_like(vector_array)
status = self.library.thermalsim_apply_structured(
int(layer_count),
int(rows),
int(cols),
self._pointer(gx_array),
self._pointer(gy_array),
self._pointer(gz_array),
self._pointer(boundary_array),
self._pointer(vector_array),
self._pointer(result),
)
if status != 0:
raise RuntimeError(f"Native structured operator failed ({status}).")
return result
def _candidate_paths():
override = os.environ.get("THERMALSIM_NATIVE_CORE")
if override:
yield Path(override)
root = Path(__file__).resolve().parent
yield root / "thermalsim_core.dll"
yield root / "native" / "bin" / "thermalsim_core.dll"
def _load_native_core():
for path in _candidate_paths():
if not path.is_file():
continue
try:
return NativeCore(path)
except (OSError, RuntimeError, AttributeError):
continue
return None
NATIVE_CORE = _load_native_core()
HAS_NATIVE_CORE = NATIVE_CORE is not None
def apply_structured_native(layer_count, rows, cols, gx, gy, gz, boundary, vector):
"""Return a native operator result or ``None`` when unavailable."""
if NATIVE_CORE is None:
return None
try:
return NATIVE_CORE.apply_structured(
layer_count, rows, cols, gx, gy, gz, boundary, vector
)
except (OSError, RuntimeError, ValueError):
return None