Skip to content

Commit d21e7df

Browse files
committed
[OMCSession] add *Runner related classes for OMPath and OMSession
1 parent 4256238 commit d21e7df

1 file changed

Lines changed: 153 additions & 0 deletions

File tree

OMPython/OMCSession.py

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1720,3 +1720,156 @@ def _omc_port_get(self) -> str:
17201720
f"pid={self._omc_process.pid if isinstance(self._omc_process, subprocess.Popen) else '?'}")
17211721

17221722
return port
1723+
1724+
1725+
class OMPathRunnerABC(OMPathABC, metaclass=abc.ABCMeta):
1726+
"""
1727+
Base function for OMPath definitions *without* OMC server
1728+
"""
1729+
1730+
def _path(self) -> pathlib.Path:
1731+
return pathlib.Path(self.as_posix())
1732+
1733+
1734+
class _OMPathRunnerLocal(OMPathRunnerABC):
1735+
"""
1736+
Implementation of OMPathBase which does not use the session data at all. Thus, this implementation can run locally
1737+
without any usage of OMC.
1738+
1739+
This class is based on OMPathBase and, therefore, on pathlib.PurePosixPath. This is working well, but it is not the
1740+
correct implementation on Windows systems. To get a valid Windows representation of the path, use the conversion
1741+
via pathlib.Path(<OMCPathDummy>.as_posix()).
1742+
"""
1743+
1744+
def is_file(self) -> bool:
1745+
"""
1746+
Check if the path is a regular file.
1747+
"""
1748+
return self._path().is_file()
1749+
1750+
def is_dir(self) -> bool:
1751+
"""
1752+
Check if the path is a directory.
1753+
"""
1754+
return self._path().is_dir()
1755+
1756+
def is_absolute(self):
1757+
"""
1758+
Check if the path is an absolute path.
1759+
"""
1760+
return self._path().is_absolute()
1761+
1762+
def read_text(self) -> str:
1763+
"""
1764+
Read the content of the file represented by this path as text.
1765+
"""
1766+
return self._path().read_text(encoding='utf-8')
1767+
1768+
def write_text(self, data: str):
1769+
"""
1770+
Write text data to the file represented by this path.
1771+
"""
1772+
return self._path().write_text(data=data, encoding='utf-8')
1773+
1774+
def mkdir(self, parents: bool = True, exist_ok: bool = False):
1775+
"""
1776+
Create a directory at the path represented by this class.
1777+
1778+
The argument parents with default value True exists to ensure compatibility with the fallback solution for
1779+
Python < 3.12. In this case, pathlib.Path is used directly and this option ensures, that missing parent
1780+
directories are also created.
1781+
"""
1782+
return self._path().mkdir(parents=parents, exist_ok=exist_ok)
1783+
1784+
def cwd(self):
1785+
"""
1786+
Returns the current working directory as an OMPathBase object.
1787+
"""
1788+
return self._path().cwd()
1789+
1790+
def unlink(self, missing_ok: bool = False) -> None:
1791+
"""
1792+
Unlink (delete) the file or directory represented by this path.
1793+
"""
1794+
return self._path().unlink(missing_ok=missing_ok)
1795+
1796+
def resolve(self, strict: bool = False):
1797+
"""
1798+
Resolve the path to an absolute path. This is done based on available OMC functions.
1799+
"""
1800+
path_resolved = self._path().resolve(strict=strict)
1801+
return type(self)(path_resolved, session=self._session)
1802+
1803+
def size(self) -> int:
1804+
"""
1805+
Get the size of the file in bytes - implementation baseon on pathlib.Path.
1806+
"""
1807+
if not self.is_file():
1808+
raise OMCSessionException(f"Path {self.as_posix()} is not a file!")
1809+
1810+
path = self._path()
1811+
return path.stat().st_size
1812+
1813+
1814+
if sys.version_info < (3, 12):
1815+
OMPathRunnerLocal = OMPathCompatibility
1816+
else:
1817+
OMPathRunnerLocal = _OMPathRunnerLocal # OMPathRunnerLocal
1818+
1819+
1820+
class OMSessionRunner(OMSessionABC):
1821+
"""
1822+
Implementation based on OMSessionABC without any use of an OMC server.
1823+
"""
1824+
1825+
def __init__(
1826+
self,
1827+
timeout: float = 10.00,
1828+
version: str = "1.27.0"
1829+
) -> None:
1830+
super().__init__(timeout=timeout)
1831+
self.model_execution_local = True
1832+
self._version = version
1833+
1834+
def __post_init__(self) -> None:
1835+
"""
1836+
No connection to an OMC server is created by this class!
1837+
"""
1838+
1839+
def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]:
1840+
"""
1841+
Helper function which returns a command prefix.
1842+
"""
1843+
return []
1844+
1845+
def get_version(self) -> str:
1846+
"""
1847+
We can not provide an OM version as we are not link to an OMC server. Thus, the provided version string is used
1848+
directly.
1849+
"""
1850+
return self._version
1851+
1852+
def set_workdir(self, workdir: OMPathABC) -> None:
1853+
"""
1854+
Set the workdir for this session.
1855+
"""
1856+
os.chdir(workdir.as_posix())
1857+
1858+
def omcpath(self, *path) -> OMPathABC:
1859+
"""
1860+
Create an OMCPath object based on the given path segments and the current OMCSession* class.
1861+
"""
1862+
return OMPathRunnerLocal(*path, session=self)
1863+
1864+
def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC:
1865+
"""
1866+
Get a temporary directory without using OMC.
1867+
"""
1868+
if tempdir_base is None:
1869+
tempdir_str = tempfile.gettempdir()
1870+
tempdir_base = self.omcpath(tempdir_str)
1871+
1872+
return self._tempdir(tempdir_base=tempdir_base)
1873+
1874+
def sendExpression(self, command: str, parsed: bool = True) -> Any:
1875+
raise OMCSessionException(f"{self.__class__.__name__} does not uses an OMC server!")

0 commit comments

Comments
 (0)