248 lines
7.1 KiB
Python
248 lines
7.1 KiB
Python
import json
|
|
import importlib
|
|
from pathlib import Path
|
|
import sys
|
|
from colorama import Fore as _color
|
|
from functools import partial
|
|
from .aliases import Alias
|
|
import getpass
|
|
import colorama
|
|
import socket
|
|
|
|
|
|
#TODO: add lazy_object_proxy as dep?
|
|
try:
|
|
from lazy_object_proxy import Proxy
|
|
except ImportError:
|
|
print(_color.RED + "Cannot import lazy_object_proxy, thus cannot create lazy objects" + _color.RESET)
|
|
Proxy = None
|
|
|
|
|
|
class Component:
|
|
def __init__(self, namestring):
|
|
self.name = namestring
|
|
|
|
|
|
def init_name_obj(obj, args, kwargs, name=None):
|
|
try:
|
|
return obj(*args, **kwargs, name=name)
|
|
except TypeError:
|
|
return obj(*args, **kwargs)
|
|
|
|
|
|
def init_device(type_string, name, args=[], kwargs={}, verbose=True, lazy=True):
|
|
if verbose:
|
|
print(("Configuring %s " % (name)).ljust(25), end="")
|
|
sys.stdout.flush()
|
|
imp_p, type_name = type_string.split(sep=":")
|
|
imp_p = imp_p.split(sep=".")
|
|
if verbose:
|
|
print(("(%s)" % (type_name)).ljust(25), end="")
|
|
sys.stdout.flush()
|
|
try:
|
|
tg = importlib.import_module(".".join(imp_p)).__dict__[type_name]
|
|
|
|
if lazy and Proxy:
|
|
tdev = Proxy(partial(init_name_obj, tg, args, kwargs, name=name))
|
|
if verbose:
|
|
print((_color.YELLOW + "LAZY" + _color.RESET).rjust(5))
|
|
sys.stdout.flush()
|
|
else:
|
|
tdev = init_name_obj(tg, args, kwargs, name=name)
|
|
if verbose:
|
|
print((_color.GREEN + "OK" + _color.RESET).rjust(5))
|
|
sys.stdout.flush()
|
|
return tdev
|
|
except Exception as expt:
|
|
# tb = traceback.format_exc()
|
|
if verbose:
|
|
print((_color.RED + "FAILED" + _color.RESET).rjust(5))
|
|
# print(sys.exc_info())
|
|
raise expt
|
|
|
|
|
|
def get_dependencies(inp):
|
|
outp = []
|
|
if isinstance(inp, dict):
|
|
inp = inp.values()
|
|
for ta in inp:
|
|
if isinstance(ta, Component):
|
|
outp.append(ta.name)
|
|
elif isinstance(ta, dict) or isinstance(ta, list):
|
|
outp.append(get_dependencies(ta))
|
|
return outp
|
|
|
|
|
|
def replaceComponent(inp, dict_all, config_all, lazy=False):
|
|
if isinstance(inp, list):
|
|
outp = []
|
|
for ta in inp:
|
|
if isinstance(ta, Component):
|
|
if ta.name in dict_all.keys():
|
|
outp.append(dict_all[ta.name])
|
|
else:
|
|
ind = [ta.name==tca['name'] for tca in config_all].index(True)
|
|
outp.append(initFromConfigList(config_list[ind:ind+1],config_all,lazy=lazy))
|
|
elif isinstance(ta, dict) or isinstance(ta, list):
|
|
outp.append(replaceComponent(ta, dict_all,config_all, lazy=lazy))
|
|
else:
|
|
outp.append(ta)
|
|
elif isinstance(inp, dict):
|
|
outp = {}
|
|
for tk, ta in inp.items():
|
|
if isinstance(ta, Component):
|
|
if ta.name in dict_all.keys():
|
|
outp[tk] = dict_all[ta.name]
|
|
else:
|
|
ind = [tk.name==tca['name'] for tca in config_all].index(True)
|
|
outp[tk] = initFromConfigList(config_list[ind:ind+1],config_all,lazy=lazy)
|
|
elif isinstance(ta, dict) or isinstance(ta, list):
|
|
outp[tk] = replaceComponent(ta, dict_all, config_all, lazy=lazy)
|
|
else:
|
|
outp[tk] = ta
|
|
else:
|
|
return inp
|
|
return outp
|
|
|
|
|
|
def initFromConfigList(config_list, config_all, lazy=False):
|
|
op = {}
|
|
for td in config_list:
|
|
# args = [op[ta.name] if isinstance(ta, Component) else ta for ta in td["args"]]
|
|
# kwargs = {
|
|
# tkwk: op[tkwv.name] if isinstance(tkwv, Component) else tkwv
|
|
# for tkwk, tkwv in td["kwargs"].items()
|
|
# }
|
|
try:
|
|
tlazy = td["lazy"]
|
|
except:
|
|
tlazy = lazy
|
|
op[td["name"]] = init_device(
|
|
td["type"],
|
|
td["name"],
|
|
replaceComponent(td["args"], op, config_all, lazy=lazy),
|
|
replaceComponent(td["kwargs"], op, config_all, lazy=lazy),
|
|
lazy=tlazy,
|
|
)
|
|
return op
|
|
|
|
|
|
class Configuration:
|
|
"""Configuration collector object collecting important settings for arbitrary use,
|
|
linking to one or few standard config files in the file system. Sould also be used
|
|
for config file writing."""
|
|
|
|
def __init__(self, configFile, name=None):
|
|
self.name = name
|
|
self.configFile = Path(configFile)
|
|
self._config = {}
|
|
if self.configFile:
|
|
self.readConfigFile()
|
|
|
|
def readConfigFile(self):
|
|
self._config = loadConfig(self.configFile)
|
|
assert (
|
|
type(self._config) is dict
|
|
), f"Problem reading {self.configFile} json file, seems not to be a valid dictionary structure!"
|
|
# self.__dict__.update(self._config)
|
|
|
|
def __setitem__(self, key, item):
|
|
self._config[key] = item
|
|
# self.__dict__.update(self._config)
|
|
self.saveConfigFile()
|
|
|
|
def __getitem__(self, key):
|
|
return self._config[key]
|
|
|
|
def saveConfigFile(self, filename=None, force=False):
|
|
if not filename:
|
|
filename = self.configFile
|
|
if (not force) and filename.exists():
|
|
if (
|
|
not input(
|
|
f"File {filename.absolute().as_posix()} exists,\n would you like to overwrite? (y/n)"
|
|
).strip()
|
|
== "y"
|
|
):
|
|
return
|
|
writeConfig(filename, self._config)
|
|
|
|
def _ipython_key_completions_(self):
|
|
return list(self._config.keys())
|
|
|
|
def __repr__(self):
|
|
return json.dumps(self._config, indent=4)
|
|
|
|
|
|
def loadConfig(fina):
|
|
with open(fina, "r") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def writeConfig(fina, obj):
|
|
with open(fina, "w") as f:
|
|
json.dump(obj, f, indent=4)
|
|
|
|
|
|
class ChannelList(list):
|
|
def __init__(self,*args,**kwargs):
|
|
self.file_name = kwargs.pop("file_name")
|
|
# list.__init__(*args,**kwargs)
|
|
self.load()
|
|
|
|
def load(self):
|
|
self.clear()
|
|
self.extend(parseChannelListFile(self.file_name))
|
|
|
|
|
|
def parseChannelListFile(fina):
|
|
out = []
|
|
with open(fina, "r") as f:
|
|
done = False
|
|
while not done:
|
|
d = f.readline()
|
|
if not d:
|
|
done = True
|
|
if len(d) > 0:
|
|
if not d.isspace():
|
|
if not d[0] == "#":
|
|
out.append(d.strip())
|
|
return out
|
|
|
|
|
|
def append_to_path(*args):
|
|
for targ in args:
|
|
sys.path.append(targ)
|
|
|
|
|
|
def prepend_to_path(*args):
|
|
for targ in args:
|
|
sys.path.insert(0, targ)
|
|
|
|
class Terminal:
|
|
def __init__(self,title='eco',scope=None):
|
|
self.title = title
|
|
self.scope = scope
|
|
|
|
@property
|
|
def user(self):
|
|
return getpass.getuser()
|
|
|
|
@property
|
|
def host(self):
|
|
return socket.gethostname()
|
|
|
|
@property
|
|
def user(self):
|
|
return getpass.getuser()
|
|
|
|
def get_string(self):
|
|
s = f'{self.title}'
|
|
if self.scope:
|
|
s +=f'-{self.scope}'
|
|
s += f' ({self.user}@{self.host})'
|
|
return s
|
|
|
|
def set_title(self,extension=''):
|
|
print(colorama.ansi.set_title("♻️ "+self.get_string()+extension))
|