-# -*- coding: iso-8859-1 -*-\r
-# Copyright (C) 2007-2020 CEA/DEN, EDF R&D, OPEN CASCADE\r
-#\r
-# This library is free software; you can redistribute it and/or\r
-# modify it under the terms of the GNU Lesser General Public\r
-# License as published by the Free Software Foundation; either\r
-# version 2.1 of the License, or (at your option) any later version.\r
-#\r
-# This library is distributed in the hope that it will be useful,\r
-# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
-# Lesser General Public License for more details.\r
-#\r
-# You should have received a copy of the GNU Lesser General Public\r
-# License along with this library; if not, write to the Free Software\r
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
-#\r
-# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
-#\r
-\r
-# ---\r
-# File : salome_utils.py\r
-# Author : Vadim SANDLER, Open CASCADE S.A.S. (vadim.sandler@opencascade.com)\r
-# ---\r
-\r
-## @package salome_utils\r
-# \brief Set of utility functions used by SALOME python scripts.\r
-\r
-#\r
-# Exported functions\r
-#\r
-\r
-__all__ = [\r
- 'getORBcfgInfo',\r
- 'getHostFromORBcfg',\r
- 'getPortFromORBcfg',\r
- 'getUserName',\r
- 'getHostName',\r
- 'getShortHostName',\r
- 'getAppName',\r
- 'getPortNumber',\r
- 'getLogDir',\r
- 'getTmpDir',\r
- 'getHomeDir',\r
- 'generateFileName',\r
- 'makeTmpDir',\r
- 'uniteFiles',\r
- ]\r
-\r
-# ---\r
-\r
-def _try_bool( arg ):\r
- """\r
- Check if specified parameter represents boolean value and returns its value.\r
- String values like 'True', 'TRUE', 'YES', 'Yes', 'y', 'NO', 'false', 'n', etc\r
- are supported.\r
- If <arg> does not represent a boolean, an exception is raised.\r
- """\r
- if isinstance(arg, bool) :\r
- return arg\r
- elif isinstance(arg, (str, bytes)):\r
- v = str( arg ).lower()\r
- if v in [ "yes", "y", "true" ]: return True\r
- elif v in [ "no", "n", "false" ]: return False\r
- pass\r
- raise Exception("Not boolean value")\r
-\r
-# ---\r
-\r
-def getORBcfgInfo():\r
- """\r
- Get omniORB current configuration.\r
- Returns a list of three values: [ orb_version, host_name, port_number ].\r
-\r
- The information is retrieved from the omniORB configuration file defined\r
- by the OMNIORB_CONFIG environment variable.\r
- If omniORB configuration file can not be accessed, a list of three empty\r
- strings is returned.\r
- """\r
- import os, re\r
- ret = [ "", "", "" ]\r
- try:\r
- f = open( os.getenv( "OMNIORB_CONFIG" ) )\r
- lines = f.readlines()\r
- f.close()\r
- regvar = re.compile( "(ORB)?InitRef.*corbaname::(.*):(\d+)\s*$" )\r
- for l in lines:\r
- try:\r
- m = regvar.match( l )\r
- if m:\r
- if m.group(1) is None:\r
- ret[0] = "4"\r
- else:\r
- ret[0] = "3"\r
- pass\r
- ret[1] = m.group(2)\r
- ret[2] = m.group(3)\r
- break\r
- pass\r
- except:\r
- pass\r
- pass\r
- pass\r
- except:\r
- pass\r
- return ret\r
-\r
-# ---\r
-\r
-def getHostFromORBcfg():\r
- """\r
- Get current omniORB host.\r
- """\r
- return getORBcfgInfo()[1]\r
-# ---\r
-\r
-def getPortFromORBcfg():\r
- """\r
- Get current omniORB port.\r
- """\r
- return getORBcfgInfo()[2]\r
-\r
-# ---\r
-\r
-def getUserName():\r
- """\r
- Get user name:\r
- 1. try USER environment variable (USERNAME on windows)\r
- 2. if fails, try LOGNAME (un*x)\r
- 3. if fails return 'unknown' as default user name\r
- """\r
- import os, sys\r
- if sys.platform == "win32":\r
- return os.getenv("USERNAME", "unknown")\r
- else:\r
- user = os.getenv("USER")\r
- if user:\r
- return user\r
- return os.getenv("LOGNAME", "unknown")\r
-# ---\r
-\r
-def getHostName():\r
- """\r
- Get host name:\r
- 1. try socket python module gethostname() function\r
- 2. if fails, try HOSTNAME environment variable\r
- 3. if fails, try HOST environment variable\r
- 4. if fails, return 'unknown' as default host name\r
- """\r
- try:\r
- import socket\r
- host = socket.gethostname()\r
- except:\r
- host = None\r
- pass\r
- if not host: host = os.getenv("HOSTNAME")\r
- if not host: host = os.getenv("HOST")\r
- if not host: host = "unknown" # 'unknown' is default host name\r
- try:\r
- socket.gethostbyname(host)\r
- except:\r
- host = "localhost"\r
- pass\r
- return host\r
-\r
-# ---\r
-\r
-def getShortHostName():\r
- """\r
- Get short host name:\r
- 1. try socket python module gethostname() function\r
- 2. if fails, try HOSTNAME environment variable\r
- 3. if fails, try HOST environment variable\r
- 4. if fails, return 'unknown' as default host name\r
- """\r
- try:\r
- return getHostName().split('.')[0]\r
- except:\r
- pass\r
- return "unknown" # 'unknown' is default host name\r
-\r
-# ---\r
-\r
-def getAppName():\r
- """\r
- Get application name:\r
- 1. try APPNAME environment variable\r
- 2. if fails, return 'SALOME' as default application name\r
- """\r
- import os\r
- return os.getenv( "APPNAME", "SALOME" ) # 'SALOME' is default user name\r
-\r
-# ---\r
-\r
-def getPortNumber(use_default=True):\r
- """\r
- Get current naming server port number:\r
- 1. try NSPORT environment variable\r
- 1. if fails, try to parse config file defined by OMNIORB_CONFIG environment variable\r
- 2. if fails, return 2809 as default port number (if use_default is True) or None (id use_default is False)\r
- """\r
- import os\r
- try:\r
- return int( os.getenv( "NSPORT" ) )\r
- except:\r
- pass\r
- try:\r
- port = int( getPortFromORBcfg() )\r
- if port is not None: return port\r
- except:\r
- pass\r
- if use_default: return 2809 # '2809' is default port number\r
- return None\r
-\r
-# ---\r
-\r
-def getHomeDir():\r
- """\r
- Get home directory.\r
- """\r
- import os\r
- return os.path.realpath(os.path.expanduser('~'))\r
-# ---\r
-\r
-def getLogDir():\r
- """\r
- Get directory to be used for the log files.\r
- """\r
- import os\r
- return os.path.join(getTmpDir(), "logs", getUserName())\r
-# ---\r
-\r
-def getTmpDir():\r
- """\r
- Get directory to be used for the temporary files.\r
- """\r
- import os, tempfile\r
- f = tempfile.NamedTemporaryFile()\r
- tmpdir = os.path.dirname(f.name)\r
- f.close()\r
- return tmpdir\r
-# ---\r
-\r
-def generateFileName( dir, prefix = None, suffix = None, extension = None,\r
- unique = False, separator = "_", hidden = False, **kwargs ):\r
- """\r
- Generate file name by specified parameters. If necessary, file name\r
- can be generated to be unique.\r
-\r
- Parameters:\r
- - dir : directory path\r
- - prefix : file prefix (not added by default)\r
- - suffix : file suffix (not added by default)\r
- - extension : file extension (not added by default)\r
- - unique : if this parameter is True, the unique file name is generated:\r
- in this case, if the file with the generated name already exists\r
- in the <dir> directory, an integer suffix is added to the end of the\r
- file name. This parameter is False by default.\r
- - separator : separator of the words ('_' by default)\r
- - hidden : if this parameter is True, the file name is prepended by . (dot)\r
- symbol. This parameter is False by default.\r
-\r
- Other keyword parameters are:\r
- - with_username : 'add user name' flag/option:\r
- * boolean value can be passed to determine user name automatically\r
- * string value to be used as user name\r
- - with_hostname : 'add host name' flag/option:\r
- * boolean value can be passed to determine host name automatically\r
- * string value to be used as host name\r
- - with_port : 'add port number' flag/option:\r
- * boolean value can be passed to determine port number automatically\r
- * string value to be used as port number\r
- - with_app : 'add application name' flag/option:\r
- * boolean value can be passed to determine application name automatically\r
- * string value to be used as application name\r
- All <with_...> parameters are optional.\r
- """\r
- supported = [ 'with_username', 'with_hostname', 'with_port', 'with_app' ]\r
- from launchConfigureParser import verbose\r
- filename = []\r
- # separator\r
- if separator is None:\r
- separator = ""\r
- pass\r
- else:\r
- separator = str( separator )\r
- pass\r
- # prefix (if specified)\r
- if prefix is not None:\r
- filename.append( str( prefix ) )\r
- pass\r
- # additional keywords\r
- ### check unsupported parameters\r
- for kw in kwargs:\r
- if kw not in supported and verbose():\r
- print('Warning! salome_utilitie.py: generateFileName(): parameter %s is not supported' % kw)\r
- pass\r
- pass\r
- ### process supported keywords\r
- for kw in supported:\r
- if kw not in kwargs: continue\r
- ### user name\r
- if kw == 'with_username':\r
- try:\r
- # auto user name ?\r
- if _try_bool( kwargs[kw] ): filename.append( getUserName() )\r
- pass\r
- except:\r
- # user name given as parameter\r
- filename.append( kwargs[kw] )\r
- pass\r
- pass\r
- ### host name\r
- elif kw == 'with_hostname':\r
- try:\r
- # auto host name ?\r
- if _try_bool( kwargs[kw] ): filename.append( getShortHostName() )\r
- pass\r
- except:\r
- # host name given as parameter\r
- filename.append( kwargs[kw] )\r
- pass\r
- pass\r
- ### port number\r
- elif kw == 'with_port':\r
- try:\r
- # auto port number ?\r
- if _try_bool( kwargs[kw] ): filename.append( str( getPortNumber() ) )\r
- pass\r
- except:\r
- # port number given as parameter\r
- filename.append( str( kwargs[kw] ) )\r
- pass\r
- pass\r
- ### application name\r
- elif kw == 'with_app':\r
- try:\r
- # auto application name ?\r
- if _try_bool( kwargs[kw] ): filename.append( getAppName() )\r
- pass\r
- except:\r
- # application name given as parameter\r
- filename.append( kwargs[kw] )\r
- pass\r
- pass\r
- pass\r
- # suffix (if specified)\r
- if suffix is not None:\r
- filename.append( str( suffix ) )\r
- pass\r
- # raise an exception if file name is empty\r
- if not filename:\r
- raise Exception("Empty file name")\r
- #\r
- if extension is not None and extension.startswith("."): extension = extension[1:]\r
- #\r
- import os\r
- name = separator.join( filename )\r
- if hidden: name = "." + name # add dot for hidden files\r
- if extension: name = name + "." + str( extension ) # add extension if defined\r
- name = os.path.join( dir, name )\r
- if unique:\r
- # create unique file name\r
- index = 0\r
- while os.path.exists( name ):\r
- index = index + 1\r
- name = separator.join( filename ) + separator + str( index )\r
- if hidden: name = "." + name # add dot for hidden files\r
- if extension: name = name + "." + str( extension ) # add extension if defined\r
- name = os.path.join( dir, name )\r
- pass\r
- pass\r
- return os.path.normpath(name)\r
-\r
-# ---\r
-\r
-def makeTmpDir( path, mode=0o777 ):\r
- """\r
- Make temporary directory with the specified path.\r
- If the directory exists then clear its contents.\r
-\r
- Parameters:\r
- - path : absolute path to the directory to be created.\r
- - mode : access mode\r
- """\r
- import os\r
- if os.path.exists( path ):\r
- import sys\r
- if sys.platform == "win32":\r
- os.system( "rmdir /S /Q " + '"' + path + '"' )\r
- os.system( "mkdir " + '"' + path + '"' )\r
- else:\r
- os.system( "rm -rf " + path + "/*" )\r
- else:\r
- dirs = path.split("/")\r
- shift1 = shift2 = 0\r
- if not dirs[0]: shift1 = 1\r
- if dirs[-1]: shift2 = 1\r
- for i in range(1+shift1,len(dirs)+shift2):\r
- p = "/".join(dirs[:i])\r
- try:\r
- os.mkdir(p, mode)\r
- os.chmod(p, mode)\r
- except:\r
- pass\r
-\r
-# ---\r
-\r
-def uniteFiles( src_file, dest_file ):\r
- """\r
- Unite contents of the source file with contents of the destination file\r
- and put result of the uniting to the destination file.\r
- If the destination file does not exist then the source file is simply\r
- copied to its path.\r
-\r
- Parameters:\r
- - src_file : absolute path to the source file\r
- - dest_file : absolute path to the destination file\r
- """\r
- import os\r
-\r
- if not os.path.exists( src_file ):\r
- return\r
- pass\r
-\r
- if os.path.exists( dest_file ):\r
- # add a symbol of new line to contents of the destination file (just in case)\r
- dest = open( dest_file, 'r' )\r
- dest_lines = dest.readlines()\r
- dest.close()\r
-\r
- dest_lines.append( "\n" )\r
-\r
- dest = open( dest_file, 'w' )\r
- dest.writelines( dest_lines )\r
- dest.close()\r
-\r
- import sys\r
- if sys.platform == "win32":\r
- command = "type " + '"' + src_file + '"' + " >> " + '"' + dest_file + '"'\r
- else:\r
- command = "cat " + src_file + " >> " + dest_file\r
- pass\r
- pass\r
- else:\r
- import sys\r
- if sys.platform == "win32":\r
- command = "copy " + '"' + src_file + '"' + " " + '"' + dest_file + '"' + " > nul"\r
- else:\r
- command = "cp " + src_file + " " + dest_file\r
- pass\r
- pass\r
-\r
- os.system( command )\r
-\r
-# --\r
-\r
-_verbose = None\r
-\r
-def verbose():\r
- """\r
- Get verbosity level. Default verbosity level is specified via the environment variable\r
- SALOME_VERBOSE, e.g.:\r
- [bash %] export SALOME_VERBOSE=1\r
- The function setVerbose() can be used to change verbosity level explicitly.\r
- """\r
- global _verbose\r
- # verbose has already been called\r
- if _verbose is not None:\r
- return _verbose\r
- # first time\r
- try:\r
- from os import getenv\r
- _verbose = int(getenv('SALOME_VERBOSE'))\r
- except:\r
- _verbose = 0\r
- pass\r
- #\r
- return _verbose\r
-# --\r
-\r
-def setVerbose(level):\r
- """\r
- Change verbosity level. The function verbose() can be used to get current verbosity level.\r
- """\r
- global _verbose\r
- _verbose = level\r
- return\r
-# --\r
-\r
-import signal\r
-def killpid(pid, sig = 9):\r
- """\r
- Send signal sig to the process by pid.\r
-\r
- Parameters:\r
- - pid : PID of process\r
- - sig : signal for sending\r
- Possible values of signals: \r
- 9 means kill the process\r
- 0 only check existing of the process\r
- NOTE: Other values are not processed on Windows\r
- Returns:\r
- 1 Success\r
- 0 Fail, no such process\r
- -1 Fail, another reason\r
-\r
- """\r
- if not pid: return\r
- import os, sys\r
- if sig != 0:\r
- if verbose(): print("######## killpid pid = ", pid)\r
- try:\r
- if sys.platform == "win32":\r
- import ctypes\r
- if sig == 0:\r
- # PROCESS_QUERY_INFORMATION (0x0400) Required to retrieve certain information about a process\r
- SYNCHRONIZE = 0x100000\r
- handle = ctypes.windll.kernel32.OpenProcess(SYNCHRONIZE, False, int(pid))\r
- waitObj = ctypes.windll.kernel32.WaitForSingleObject(handle, 0)\r
- if waitObj:\r
- ret = 1\r
- ctypes.windll.kernel32.CloseHandle(handle)\r
- else:\r
- ret = 0\r
- if sig == 9:\r
- # PROCESS_TERMINATE (0x0001) Required to terminate a process using TerminateProcess.\r
- handle = ctypes.windll.kernel32.OpenProcess(0x0001, False, int(pid))\r
- ret = ctypes.windll.kernel32.TerminateProcess(handle, -1)\r
- ctypes.windll.kernel32.CloseHandle(handle)\r
- pass\r
- pass\r
- else:\r
- # Default: signal.SIGKILL = 9\r
- os.kill(int(pid),sig)\r
- ret = 1\r
- pass\r
- pass\r
- except OSError as e:\r
- # errno.ESRCH == 3 is 'No such process'\r
- if e.errno == 3:\r
- ret = 0\r
- else:\r
- ret = -1\r
- pass\r
- pass\r
- return ret\r
-# --\r
-\r
-def getOmniNamesPid(port):\r
- """\r
- Return OmniNames pid by port number.\r
- """\r
- import sys,subprocess,re\r
- if sys.platform == "win32":\r
- # Get process list by WMI Command Line Utility(WMIC)\r
- # Output is formatted with each value listed on a separate line and with the name of the property:\r
- # ...\r
- # Caption=<caption0>\r
- # CommandLine=<commandline0>\r
- # ProcessId=<processid0>\r
- #\r
- #\r
- #\r
- # Caption=<caption1>\r
- # CommandLine=<commandline1>\r
- # ProcessId=<processid1>\r
- # ...\r
- cmd = 'WMIC PROCESS get Caption,Commandline,Processid /VALUE'\r
- proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)\r
- # Get stdout\r
- allProc = proc.communicate()[0].decode()\r
- # find Pid of omniNames\r
- pid = re.findall(r'Caption=.*omniNames.*\n?CommandLine=.*omniNames.*\D%s\D.*\n?ProcessId=(\d*)'%(port),allProc)[0]\r
- else: \r
- cmd = "ps -eo pid,command | grep -v grep | grep -E \"omniNames.*%s\" | awk '{print $1}'"%(port)\r
- proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)\r
- pid = proc.communicate()[0]\r
- pass\r
-\r
- return pid\r
-# --\r
-\r
-def killOmniNames(port):\r
- """\r
- Kill OmniNames process by port number.\r
- """\r
- try:\r
- pid = getOmniNamesPid(port)\r
- if pid: killpid(pid)\r
- except:\r
- pass\r
- pass\r
-# --\r
+# -*- coding: iso-8859-1 -*-
+# Copyright (C) 2007-2023 CEA, EDF, OPEN CASCADE
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+#
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
+#
+
+## @package salome_utils
+# @brief Set of utility functions used by SALOME python scripts.
+
+"""
+Various utilities for SALOME.
+"""
+
+# pragma pylint: disable=invalid-name
+
+import os
+import os.path as osp
+import re
+import shutil
+import socket
+import sys
+import tempfile
+from contextlib import suppress
+
+import psutil
+
+def _try_bool(arg):
+ """
+ Convert given `arg` to a boolean value.
+ String values like 'True', 'TRUE', 'YES', 'Yes', 'y', 'NO', 'false', 'n', etc.
+ are supported.
+ If `arg` does not represent a boolean, an exception is raised.
+ :param arg : value being converted
+ :return result of conversion: `True` or `False`
+ """
+ if isinstance(arg, bool):
+ return arg
+ if isinstance(arg, bytes):
+ arg = arg.decode('utf-8', errors='ignore')
+ if isinstance(arg, str):
+ if arg.lower() in ('yes', 'y', 'true', 't'):
+ return True
+ if arg.lower() in ('no', 'n', 'false', 'f'):
+ return False
+ raise ValueError('Not a boolean value')
+
+# ---
+
+def getORBcfgInfo():
+ """
+ Get current omniORB configuration.
+
+ The information is retrieved from the omniORB configuration file defined
+ by the OMNIORB_CONFIG environment variable.
+ If omniORB configuration file can not be accessed, a tuple of three empty
+ strings is returned.
+
+ :return tuple of three strings: (orb_version, host_name, port_number)
+ """
+ orb_version = ''
+ hostname = ''
+ port_number = ''
+ with suppress(Exception), open(os.getenv('OMNIORB_CONFIG')) as forb:
+ regvar = re.compile(r'(ORB)?InitRef.*corbaname::(.*):(\d+)\s*$')
+ for line in forb.readlines():
+ match = regvar.match(line)
+ if match:
+ orb_version = '4' if match.group(1) is None else '3'
+ hostname = match.group(2)
+ port_number = match.group(3)
+ break
+ return orb_version, hostname, port_number
+
+# ---
+
+def getHostFromORBcfg():
+ """
+ Get current omniORB host name.
+ :return host name
+ """
+ return getORBcfgInfo()[1]
+
+# ---
+
+def getPortFromORBcfg():
+ """
+ Get current omniORB port.
+ :return port number
+ """
+ return getORBcfgInfo()[2]
+
+# ---
+
+def getUserName():
+ """
+ Get user name.
+
+ The following procedure is perfomed to deduce user name:
+ 1. try USER (USERNAME on Windows) environment variable.
+ 2. if (1) fails, try LOGNAME (un*x only).
+ 3. if (2) fails, return 'unknown' as default user name
+
+ :return user name
+ """
+ if sys.platform == 'win32':
+ username = os.getenv('USERNAME')
+ else:
+ username = os.getenv('USER', os.getenv('LOGNAME'))
+ if username is None:
+ import getpass
+ username = getpass.getuser()
+ return username
+
+# ---
+
+def getHostName():
+ """
+ Get host name.
+
+ The following procedure is perfomed to deduce host name:
+ 1. try socket python module, gethostname() function
+ 2. if (1) fails, try HOSTNAME environment variable
+ 3. if (2) fails, try HOST environment variable
+ 4. if (3) fails, tries 'unknown' as default host name
+ 5. finally, checks that IP is configured for hostname; if not, returns 'localhost'
+
+ :return host name
+ """
+ host = None
+ with suppress(Exception):
+ host = socket.gethostname()
+ if not host:
+ host = os.getenv('HOSTNAME', os.getenv('HOST', 'unknown'))
+ try:
+ # the following line just checks that IP is configured for hostname
+ socket.gethostbyname(host)
+ except (TypeError, OSError):
+ host = 'localhost'
+ return host
+
+# ---
+
+def getShortHostName():
+ """
+ Get short host name (with domain stripped).
+ See `getHostName()` for more details.
+ :return short host name
+ """
+ with suppress(AttributeError, IndexError):
+ return getHostName().split('.')[0]
+ return 'unknown' # default host name
+
+# ---
+
+def getAppName():
+ """
+ Get application name.
+ The following procedure is perfomed to deduce application name:
+ 1. try APPNAME environment variable
+ 2. if (1) fails, return 'SALOME' as default application name
+ :return application name
+ """
+ return os.getenv('APPNAME', 'SALOME') # 'SALOME' is default user name
+
+def getPid():
+ return os.getpid()
+
+# ---
+
+def getPortNumber(use_default=True):
+ """
+ Get currently used omniORB port.
+ The following procedure is perfomed to deduce port number:
+ 1. try NSPORT environment variable
+ 2. if (1) fails, try to parse config file defined by OMNIORB_CONFIG environment variable
+ 3. if (2) fails, return 2809 as default port number (if use_default is `True`) or `None`
+ (if use_default is `False`)
+ :return port number
+ """
+ with suppress(TypeError, ValueError):
+ return int(os.getenv('NSPORT'))
+ with suppress(TypeError, ValueError):
+ port = int(getPortFromORBcfg())
+ if port:
+ return port
+ return 2809 if use_default else None
+
+# ---
+
+def getHomeDir():
+ """
+ Get home directory.
+ :return home directory path
+ """
+ return osp.realpath(osp.expanduser('~'))
+
+# ---
+
+def getLogDir():
+ """
+ Get directory that stores log files.
+ :return path to the log directory
+ """
+ return osp.join(getTmpDir(), 'logs', getUserName())
+
+# ---
+
+def getTmpDir():
+ """
+ Get directory to store temporary files.
+ :return temporary directory path
+ """
+ with tempfile.NamedTemporaryFile() as tmp:
+ return osp.dirname(tmp.name)
+ return None
+
+# ---
+
+# pragma pylint: disable=too-many-arguments
+def generateFileName(path, prefix=None, suffix=None, extension=None,
+ unique=False, separator='_', hidden=False, **kwargs):
+ """
+ Generate file name.
+
+ :param path : directory path
+ :param prefix : file name prefix (none by default)
+ :param suffix : file name suffix (none by default)
+ :param extension : file extension (none by default)
+ :param unique : if `True`, function generates unique file name -
+ in this case, if file with the generated name already
+ exists in `path` directory, an integer suffix is appended
+ to the file name (`False` by default)
+ :param separator : words separator ('_' by default)
+ :param hidden : if `True`, file name is prepended with dot symbol
+ (`False` by default)
+ :param kwargs : additional keywrods arguments (see below)
+ :return generated file name
+
+ Additionally supported keyword parameters:
+ - with_username : use user name:
+ - with_hostname : use host name:
+ - with_port : use port number:
+ - with_app : use application name:
+ - with_pid : use current pid
+
+ Any of these keyword arguments can accept either explicit string value,
+ or `True` to automatically deduce value from current configuration.
+ """
+ filename = []
+
+ def _with_str(_str):
+ _str = '' if _str is None else str(_str)
+ if _str:
+ filename.append(_str)
+
+ def _with_kwarg(_kwarg, _func):
+ _value = kwargs.get(_kwarg, False)
+ try:
+ if _try_bool(_value):
+ filename.append(str(_func()))
+ except ValueError:
+ _with_str(_value)
+
+ _with_str(prefix)
+ _with_kwarg('with_username', getUserName)
+ _with_kwarg('with_hostname', getShortHostName)
+ _with_kwarg('with_port', getPortNumber)
+ _with_kwarg('with_app', getAppName)
+ _with_kwarg('with_pid', getPid)
+ _with_str(suffix)
+
+ # raise an exception if file name is empty
+ if not filename:
+ raise ValueError('Empty file name')
+
+ # extension
+ extension = '' if extension is None else str(extension)
+ if extension.startswith('.'):
+ extension = extension[1:]
+
+ # separator
+ separator = '' if separator is None else str(separator)
+
+ def _generate(_index=None):
+ # join all components together, add index if necessary
+ if _index is not None:
+ _name = separator.join(filename+[str(_index)])
+ else:
+ _name = separator.join(filename)
+ # prepend with dot if necessary
+ if hidden:
+ _name = '.' + _name
+ # append extension if ncessary
+ if extension:
+ _name = _name + '.' + extension
+ # now get full path
+ return osp.join(path, _name)
+
+ name = _generate()
+ if unique:
+ index = 0
+ while osp.exists(name):
+ index = index + 1
+ name = _generate(index)
+ return osp.normpath(name)
+
+# ---
+
+def cleanDir(path):
+ """
+ Clear contents of directory.
+ :param path directory path
+ """
+ if osp.exists(path):
+ for filename in os.listdir(path):
+ file_path = osp.join(path, filename)
+ with suppress(OSError):
+ if osp.isdir(file_path):
+ shutil.rmtree(file_path)
+ else:
+ os.unlink(file_path)
+
+# ---
+
+def makeDir(path, mode=0o777):
+ """
+ Make directory with the specified path.
+ :param path : directory path
+ :param mode : access mode
+ """
+ try:
+ oldmask = os.umask(0)
+ os.makedirs(path, mode=mode, exist_ok=True)
+ except IOError:
+ pass
+ finally:
+ os.umask(oldmask)
+
+# ---
+
+def makeTmpDir(path, mode=0o777):
+ """
+ Make temporary directory with the specified path.
+ If the directory exists, clear all its contents.
+ :param path : directory path
+ :param mode : access mode
+ """
+ makeDir(path, mode)
+ cleanDir(path)
+
+# ---
+
+def uniteFiles(src_file, dest_file):
+ """
+ Join contents of `src_file` and `dest_file` and put result to `dest_file`.
+ File `dest_file` may not exist.
+ :param src_file : source file path
+ :param dest_file : destination file path
+ """
+ if not osp.exists(src_file):
+ return
+
+ if osp.exists(dest_file):
+ with suppress(OSError), open(src_file, 'rb') as src, open(dest_file, 'ab') as dest:
+ dest.write(b'\n')
+ dest.write(src.read())
+ else:
+ with suppress(OSError):
+ shutil.copy(src_file, dest_file)
+
+# --
+
+def verbose():
+ """
+ Get current verbosity level.
+
+ Default verbosity level is specified via the environment variable SALOME_VERBOSE,
+ e.g. in bash:
+
+ $ export SALOME_VERBOSE=1
+
+ The function `setVerbose()` can be used to explicitly set verbosity level.
+
+ :return current verbosity level
+ """
+ if not hasattr(verbose, 'verbosity_level'):
+ verbose.verbosity_level = 0 # default value
+ with suppress(TypeError, ValueError):
+ # from SALOME_VERBOSE environment variable
+ verbose.verbosity_level = int(os.getenv('SALOME_VERBOSE', '0'))
+ return verbose.verbosity_level
+# --
+
+def setVerbose(level):
+ """
+ Change verbosity level.
+ The function `verbose()` can be used to get current verbosity level.
+ :param level : verbosity level
+ """
+ with suppress(TypeError, ValueError):
+ verbose.verbosity_level = int(level)
+# --
+
+def killPid(pid, sig=9):
+ """
+ Send signal `sig` to the process with given `pid`.
+
+ :param pid : PID of the process
+ :param sig : signal to send; some of possible values:
+ - 9 : kill process
+ - 0 : do nothing, just check process existence (see below)
+ NOTE: other values are not processed on Windows
+ :return result of execution:
+ - 1 : success
+ - 0 : fail, no such process
+ - -1 : fail, another reason
+ """
+ if not pid:
+ return -1
+
+ with suppress(ValueError):
+ pid = int(pid)
+
+ if sig == 0:
+ ret = 1 if psutil.pid_exists(pid) else 0
+ else:
+ if verbose():
+ print("######## killPid pid = ", pid)
+ try:
+ process = psutil.Process(pid)
+ process.terminate()
+ _, alive = psutil.wait_procs([process], timeout=5)
+ for proc in alive:
+ proc.kill()
+ ret = 1
+ except psutil.NoSuchProcess:
+ ret = 0
+ except OSError:
+ ret = -1
+ return ret
+# --
+
+def getOmniNamesPid(port):
+ """
+ Get PID of omniNames process running on given `port`.
+ :param port : port number
+ :return omniNames process's PID
+ """
+ processes = {p.info['pid']: p.info['name'] for p in psutil.process_iter(['pid', 'name'])}
+ return next((c.pid for c in psutil.net_connections(kind='inet') \
+ if str(c.laddr.port) == str(port) and processes.get(c.pid).startswith('omniNames')), None)
+# --
+
+def killOmniNames(port):
+ """
+ Kill omniNames process running on given `port`.
+ :param port : port number
+ """
+ with suppress(Exception):
+ killPid(getOmniNamesPid(port))
+# --
