112 lines
4.3 KiB
Python
112 lines
4.3 KiB
Python
import importlib.resources
|
|
from pathlib import Path
|
|
import sys
|
|
import shutil
|
|
import logging
|
|
import socket
|
|
|
|
|
|
class Singleton(type):
|
|
_instances = {}
|
|
|
|
def __call__(cls, *args, **kwargs):
|
|
if cls not in cls._instances:
|
|
cls._instances[cls] = super(type(cls), cls).__call__(*args, **kwargs) # pylint: disable=bad-super-call, no-member
|
|
return cls._instances[cls]
|
|
|
|
|
|
def _extract_res_dir_using_py_3_7(dir_resource_package: str, dest_path: Path):
|
|
dest_path.mkdir(parents=True, exist_ok=True)
|
|
dir_contents = importlib.resources.contents(dir_resource_package)
|
|
for child_resource in dir_contents:
|
|
# print(child_resource)
|
|
if child_resource in ['__init__.py', '__pycache__']:
|
|
continue
|
|
|
|
if importlib.resources.is_resource(dir_resource_package, child_resource):
|
|
logging.debug('extracting package %s resource %s to %s', dir_resource_package, child_resource, dest_path)
|
|
with importlib.resources.path(dir_resource_package, child_resource) as job_template_path:
|
|
shutil.copy(job_template_path, dest_path / child_resource)
|
|
else:
|
|
logging.debug('%s is not a resource... we assume its a directory', child_resource)
|
|
_extract_res_dir_using_py_3_7(dir_resource_package + '.' + child_resource, dest_path / child_resource)
|
|
|
|
|
|
def extract_resource_dir(resource_package: str, resource_dir_to_extract: str, dest_path: Path):
|
|
"""extracts (unpacks) a directory resource from a resource package to an actual directory
|
|
resource_package: eg 'iprbench.resources'
|
|
resource_dir_to_extract: the name of the directory resource in the resource package
|
|
"""
|
|
dest_path.mkdir(parents=True, exist_ok=True)
|
|
method = None
|
|
if sys.version_info >= (3, 12):
|
|
method = 'with-importlib-resources-as-files'
|
|
elif sys.version_info >= (3, 7):
|
|
# importlib.resources.as_files doesn't exist but importlib.resources exist
|
|
method = 'with-importlib-resources'
|
|
else:
|
|
method = 'with-old-pkg-resources'
|
|
|
|
if method == 'with-importlib-resources-as-files':
|
|
# https://stackoverflow.com/questions/58132947/extract-folder-from-python-package-resource
|
|
traversable = importlib.resources.files(resource_package) # pylint: disable=no-member
|
|
with importlib.resources.as_file(traversable) as path: # pylint: disable=no-member
|
|
shutil.copytree(path, dest_path)
|
|
elif method == 'with-importlib-resources':
|
|
_extract_res_dir_using_py_3_7(resource_package + '.' + resource_dir_to_extract, dest_path)
|
|
elif method == 'with-old-pkg-resources':
|
|
raise NotImplementedError()
|
|
else:
|
|
assert False, f'unexpected method : {method}'
|
|
|
|
|
|
def get_ip():
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.settimeout(0)
|
|
try:
|
|
# doesn't even have to be reachable
|
|
s.connect(('10.254.254.254', 1))
|
|
ip_address = s.getsockname()[0]
|
|
except Exception:
|
|
ip_address = '127.0.0.1'
|
|
finally:
|
|
s.close()
|
|
return ip_address
|
|
|
|
|
|
VlanId = str # eg 'SIMPA', 'REC_PHYS'
|
|
|
|
|
|
def get_vlan() -> VlanId:
|
|
ip_address = [int(s) for s in get_ip().split('.')]
|
|
logging.debug('ip address : %s', ip_address)
|
|
vlan_id = 'UNKNOWN'
|
|
if ip_address[0] == 129 and ip_address[1] == 20:
|
|
vlan_id = {
|
|
27: 'SIMPA',
|
|
79: 'REC-PHYS',
|
|
}[ip_address[2]]
|
|
elif ip_address[0] == 10 and ip_address[1] == 100:
|
|
# using univ-rennes' vpn
|
|
vlan_id = 'REC-PHYS'
|
|
else:
|
|
assert False
|
|
return vlan_id
|
|
|
|
|
|
def get_proxy_env_vars() -> str:
|
|
# fatal: unable to access 'https://github.com/hibridon/hibridon/': Failed to connect to proxy-nt.univ-rennes1.fr port 3128: Connection timed out
|
|
proxy_is_required = get_vlan() in ['SIMPA']
|
|
if proxy_is_required:
|
|
ur1_proxy_url = 'http://proxy-nt.univ-rennes1.fr:3128'
|
|
proxy_env_vars = ''
|
|
proxy_env_vars = f'{proxy_env_vars} HTTP_PROXY={ur1_proxy_url}'
|
|
proxy_env_vars = f'{proxy_env_vars} HTTPS_PROXY={ur1_proxy_url}'
|
|
proxy_env_vars = f'{proxy_env_vars} FTP_PROXY={ur1_proxy_url}'
|
|
proxy_env_vars = f'{proxy_env_vars} http_proxy={ur1_proxy_url}'
|
|
proxy_env_vars = f'{proxy_env_vars} https_proxy={ur1_proxy_url}'
|
|
proxy_env_vars = f'{proxy_env_vars} ftp_proxy={ur1_proxy_url}'
|
|
else:
|
|
proxy_env_vars = ''
|
|
return proxy_env_vars
|