Repository navigation
Expand file tree
/
Copy pathom_session_abc.py
More file actions
346 lines (277 loc) · 11.9 KB
/
Copy pathom_session_abc.py
File metadata and controls
346 lines (277 loc) · 11.9 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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
# -*- coding: utf-8 -*-
"""
Definition of a generic OM session.
"""
from __future__ import annotations
import abc
import logging
import os
import pathlib
import platform
import sys
from typing import Any, Optional
import uuid
# define logger using the current module name as ID
logger = logging.getLogger(__name__)
OMSESSION_TIMEOUT: float = 300.0
class OMSessionException(Exception):
"""
Exception which is raised by any OMC* class.
"""
# due to the compatibility layer to Python < 3.12, the OM(C)Path classes must be hidden behind the following if
# conditions. This is also the reason for OMPathABC, a simple base class to be used in ModelicaSystem* classes.
# Reason: before Python 3.12, pathlib.PurePosixPath can not be derived from; therefore, OMPathABC is not possible
if sys.version_info < (3, 12):
class _OMPathCompatibility(pathlib.Path):
"""
Compatibility class for OMPathABC in Python < 3.12. This allows to run all code which uses OMPathABC (mainly
ModelicaSystem) on these Python versions. There are remaining limitation as only local execution is possible.
"""
# modified copy of pathlib.Path.__new__() definition
def __new__(cls, *args, **kwargs):
"""Modified copy of ``pathlib.Path.__new__`` for Python < 3.12."""
logger.warning("Python < 3.12 - using a version of class OMCPath "
"based on pathlib.Path for local usage only.")
if cls is _OMPathCompatibility:
cls = _OMPathCompatibilityWindows if os.name == 'nt' else _OMPathCompatibilityPosix
self = cls._from_parts(args)
if not self._flavour.is_supported:
raise NotImplementedError(f"cannot instantiate {cls.__name__} on your system")
return self
def size(self) -> int:
"""
Needed compatibility function to have the same interface as OMCPathReal
"""
return self.stat().st_size
class _OMPathCompatibilityPosix(pathlib.PosixPath, _OMPathCompatibility):
"""
Compatibility class for OMCPath on Posix systems (Python < 3.12)
"""
class _OMPathCompatibilityWindows(pathlib.WindowsPath, _OMPathCompatibility):
"""
Compatibility class for OMCPath on Windows systems (Python < 3.12)
"""
OMPathABC = _OMPathCompatibility
else:
class OMPathABC(pathlib.PurePosixPath, metaclass=abc.ABCMeta):
"""
Implementation of a basic (PurePosix)Path object to be used within OMPython. The derived classes can use OMC as
backend and - thus - work on different configurations like docker or WSL. The connection to OMC is provided via
an instances of classes derived from BaseSession.
PurePosixPath is selected as it covers all but Windows systems (Linux, docker, WSL). However, the code is
written such that possible Windows system are taken into account. Nevertheless, the overall functionality is
limited compared to standard pathlib.Path objects.
"""
def __init__(self, *path, session: OMSessionABC) -> None:
"""Initialize the OMPath with the owning session.
Args:
*path: Path segments as in pathlib.PurePosixPath.
session: The session used for filesystem-related operations.
"""
super().__init__(*path)
self._session = session
def get_session(self) -> OMSessionABC:
"""
Get session definition used for this instance of OMPath.
"""
return self._session
def with_segments(self, *pathsegments) -> OMPathABC:
"""
Create a new OMCPath object with the given path segments.
The original definition of Path is overridden to ensure the session data is set.
"""
return type(self)(*pathsegments, session=self._session)
@abc.abstractmethod
def is_file(self, *, follow_symlinks=True) -> bool:
"""
Check if the path is a regular file.
"""
@abc.abstractmethod
def is_dir(self, *, follow_symlinks: bool = True) -> bool:
"""
Check if the path is a directory.
"""
@abc.abstractmethod
def is_absolute(self) -> bool:
"""
Check if the path is an absolute path.
"""
@abc.abstractmethod
def read_text(self, encoding=None, errors=None, newline=None) -> str:
"""
Read the content of the file represented by this path as text.
"""
@abc.abstractmethod
def write_text(self, data: str, encoding=None, errors=None, newline=None) -> int:
"""
Write text data to the file represented by this path.
"""
@abc.abstractmethod
def mkdir(self, mode=0o777, parents: bool = False, exist_ok: bool = False) -> None:
"""
Create a directory at the path represented by this class.
The argument parents with default value True exists to ensure compatibility with the fallback solution for
Python < 3.12. In this case, pathlib.Path is used directly and this option ensures, that missing parent
directories are also created.
"""
@abc.abstractmethod
def cwd(self) -> OMPathABC: # pylint: disable=W0221 # is @classmethod in the original; see pathlib.PathBase
"""
Returns the current working directory as an OMPathABC object.
"""
@abc.abstractmethod
def unlink(self, missing_ok: bool = False) -> None:
"""
Unlink (delete) the file or directory represented by this path.
"""
@abc.abstractmethod
def resolve(self, strict: bool = False) -> OMPathABC:
"""
Resolve the path to an absolute path.
"""
def absolute(self) -> OMPathABC:
"""
Resolve the path to an absolute path. Just a wrapper for resolve().
"""
return self.resolve()
def exists(self) -> bool:
"""
Semi replacement for pathlib.Path.exists().
"""
return self.is_file() or self.is_dir()
@abc.abstractmethod
def size(self) -> int:
"""
Get the size of the file in bytes - this is an extra function and the best we can do using OMC.
"""
class PostInitCaller(type):
"""
Metaclass definition to define a new function __post_init__() which is called after all __init__() functions where
executed. The workflow would read as follows:
On creating a class with the following inheritance Class2 => Class1 => Class0, where each class calls the __init__()
functions of its parent, i.e. super().__init__(), as well as __post_init__() the call schema would be:
myclass = Class2()
Class2.__init__()
Class1.__init__()
Class0.__init__()
Class2.__post_init__() <= this is done due to the metaclass
Class1.__post_init__()
Class0.__post_init__()
References:
* https://stackoverflow.com/questions/100003/what-are-metaclasses-in-python
* https://stackoverflow.com/questions/795190/how-to-perform-common-post-initialization-tasks-in-inherited-classes
"""
def __call__(cls, *args, **kwargs):
"""Call the class and trigger ``__post_init__`` of all bases.
Invokes ``type.__call__`` and then automatically runs every
``__post_init__`` up the MRO.
"""
obj = type.__call__(cls, *args, **kwargs)
obj.__post_init__()
return obj
class OMSessionMeta(abc.ABCMeta, PostInitCaller):
"""
Helper class to get a combined metaclass of ABCMeta and PostInitCaller.
References:
* https://stackoverflow.com/questions/11276037/resolving-metaclass-conflicts
"""
class OMSessionABC(metaclass=OMSessionMeta):
"""
This class implements the basic structure a OMPython session definition needs. It provides the structure for an
implementation using OMC as backend (via ZMQ) or a dummy implementation which just runs a model executable.
"""
def __init__(
self,
timeout: Optional[float] = None,
**kwargs,
) -> None:
"""
Initialisation for OMSessionBase
"""
# some helper data
self.model_execution_windows = platform.system() == "Windows"
self.model_execution_local = False
# store variables
self._timeout = OMSESSION_TIMEOUT
self.set_timeout(timeout=timeout)
# command prefix (to be used for docker or WSL)
self._cmd_prefix: list[str] = []
def __post_init__(self) -> None:
"""
Post initialisation method.
"""
def set_timeout(self, timeout: Optional[float] = None) -> float:
"""
Set the timeout to be used for OMC communication (OMCSession).
The defined value is set and the current value is returned. If None is provided as argument, nothing is changed.
"""
retval = self._timeout
if timeout is not None:
if timeout <= 0.0:
raise OMSessionException(f"Invalid timeout value: {timeout}s!")
logger.info(f"Update timeout for {self.__class__.__name__}: {retval}s => {timeout}s")
self._timeout = timeout
return retval
def get_cmd_prefix(self) -> list[str]:
"""
Get session definition used for this instance of OMPath.
"""
return self._cmd_prefix.copy()
@staticmethod
def escape_str(value: str) -> str:
"""
Escape a string such that it can be used as string within OMC expressions, i.e. escape all double quotes.
"""
return value.replace("\\", "\\\\").replace('"', '\\"')
@abc.abstractmethod
def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]:
"""
Helper function which returns a command prefix.
"""
@abc.abstractmethod
def get_version(self) -> str:
"""
Get the OM version.
"""
@abc.abstractmethod
def set_workdir(self, workdir: OMPathABC) -> None:
"""
Set the workdir for this session.
"""
@abc.abstractmethod
def omcpath(self, *path) -> OMPathABC:
"""
Create an OMPathABC object based on the given path segments and the current class.
"""
@abc.abstractmethod
def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC:
"""
Get a temporary directory based on the specific definition for this session.
"""
@staticmethod
def _tempdir(tempdir_base: OMPathABC) -> OMPathABC:
"""Create a new unique temporary directory below ``tempdir_base``.
Args:
tempdir_base: Base directory for the new temporary directory.
Returns:
An OMPathABC pointing to the created temporary directory.
"""
names = [str(uuid.uuid4()) for _ in range(100)]
tempdir: Optional[OMPathABC] = None
for name in names:
# create a unique temporary directory name
tempdir = tempdir_base / name
if tempdir.exists():
continue
tempdir.mkdir(parents=True, exist_ok=False)
break
if tempdir is None or not tempdir.is_dir():
raise FileNotFoundError(f"Cannot create a temporary directory in {tempdir_base}!")
return tempdir
@abc.abstractmethod
def sendExpression(self, expr: str, parsed: bool = True, raise_on_error: bool = True) -> Any:
"""
Function needed to send expressions to the OMC server via ZMQ.
If raise_on_error is False, 'error'-level OMC diagnostics are logged instead of raised as an
OMSessionException; use this only when the caller has its own, more precise way of verifying success.
"""