Skip to content

Commit ae240a5

Browse files
committed
[OMCSession] add *Runner related classes for OMPath and OMSession
1 parent 219a0a7 commit ae240a5

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

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

0 commit comments

Comments
 (0)