4848import sys
4949import tempfile
5050import time
51- from typing import Optional
51+ from typing import Any , Optional
5252import uuid
5353import warnings
5454import zmq
@@ -83,14 +83,14 @@ class OMCSessionException(Exception):
8383
8484class OMCSessionCmd :
8585
86- def __init__ (self , session : OMCSessionZMQ , readonly : Optional [ bool ] = False ):
86+ def __init__ (self , session : OMCSessionZMQ , readonly : bool = False ):
8787 if not isinstance (session , OMCSessionZMQ ):
8888 raise OMCSessionException ("Invalid session definition!" )
8989 self ._session = session
9090 self ._readonly = readonly
91- self ._omc_cache = {}
91+ self ._omc_cache : dict [ tuple [ str , bool ], Any ] = {}
9292
93- def _ask (self , question : str , opt : Optional [list [str ]] = None , parsed : Optional [ bool ] = True ):
93+ def _ask (self , question : str , opt : Optional [list [str ]] = None , parsed : bool = True ):
9494
9595 if opt is None :
9696 expression = question
@@ -270,18 +270,24 @@ def getClassNames(self, className=None, recursive=False, qualified=False, sort=F
270270
271271class OMCSessionZMQ :
272272
273- def __init__ (self , timeout = 10.00 ,
274- docker = None , dockerContainer = None , dockerExtraArgs = None , dockerOpenModelicaPath = "omc" ,
275- dockerNetwork = None , port = None , omhome : str = None ):
273+ def __init__ (self ,
274+ timeout : float = 10.00 ,
275+ docker : Optional [str ] = None ,
276+ dockerContainer : Optional [int ] = None ,
277+ dockerExtraArgs : Optional [list ] = None ,
278+ dockerOpenModelicaPath : str = "omc" ,
279+ dockerNetwork : Optional [str ] = None ,
280+ port : Optional [int ] = None ,
281+ omhome : Optional [str ] = None ):
276282 if dockerExtraArgs is None :
277283 dockerExtraArgs = []
278284
279285 self ._omhome = self ._get_omhome (omhome = omhome )
280286
281287 self ._omc_process = None
282288 self ._omc_command = None
283- self ._omc = None
284- self ._dockerCid = None
289+ self ._omc : Optional [ Any ] = None
290+ self ._dockerCid : Optional [ int ] = None
285291 self ._serverIPAddress = "127.0.0.1"
286292 self ._interactivePort = None
287293 self ._temp_dir = pathlib .Path (tempfile .gettempdir ())
@@ -465,7 +471,7 @@ def _set_omc_command(self, omc_path_and_args_list) -> list:
465471
466472 return omc_command
467473
468- def _get_omhome (self , omhome : str = None ):
474+ def _get_omhome (self , omhome : Optional [ str ] = None ):
469475 # use the provided path
470476 if omhome is not None :
471477 return pathlib .Path (omhome )
@@ -493,7 +499,9 @@ def _connect_to_omc(self, timeout) -> str:
493499 while True :
494500 if self ._dockerCid :
495501 try :
496- port = subprocess .check_output (["docker" , "exec" , self ._dockerCid , "cat" , self ._port_file ],
502+ port = subprocess .check_output (args = ["docker" ,
503+ "exec" , str (self ._dockerCid ),
504+ "cat" , str (self ._port_file )],
497505 stderr = subprocess .DEVNULL ).decode ().strip ()
498506 break
499507 except subprocess .CalledProcessError :
@@ -517,7 +525,7 @@ def _connect_to_omc(self, timeout) -> str:
517525
518526 port = port .replace ("0.0.0.0" , self ._serverIPAddress )
519527 logger .info (f"OMC Server is up and running at { omc_zeromq_uri } "
520- f"pid={ self ._omc_process .pid } cid={ self ._dockerCid } " )
528+ f"pid={ self ._omc_process .pid if self . _omc_process else '?' } cid={ self ._dockerCid } " )
521529
522530 # Create the ZeroMQ socket and connect to OMC server
523531 context = zmq .Context .instance ()
0 commit comments