Skip to content

Commit 08c0dff

Browse files
committed
[OMCSession] add *Runner related classes for OMPath and OMSession
1 parent 843c5ae commit 08c0dff

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
@@ -1735,3 +1735,156 @@ def _omc_port_get(self) -> str:
17351735
f"pid={self._omc_process.pid if isinstance(self._omc_process, subprocess.Popen) else '?'}")
17361736

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

0 commit comments

Comments
 (0)