diff --git a/slic/utils/eco_components/config.py b/slic/utils/eco_components/config.py deleted file mode 100644 index 49ba288e..00000000 --- a/slic/utils/eco_components/config.py +++ /dev/null @@ -1,247 +0,0 @@ -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)) diff --git a/slic/utils/eco_components/ecocnf.py b/slic/utils/eco_components/ecocnf.py deleted file mode 100644 index 9934e584..00000000 --- a/slic/utils/eco_components/ecocnf.py +++ /dev/null @@ -1,15 +0,0 @@ -#import logging -#import json -#from pathlib import Path -#from .utilities.config import Configuration - -startup_lazy = False - -scopes = [ - {"name": "Alvra", "facility": "SwissFEL", "module": "alvra"}, - {"name": "Bernina", "facility": "SwissFEL", "module": "bernina"}, - {"name": "SwissMX", "facility": "SwissFEL", "module": "swissmx"}, -] - - -# settings = Configuration(Path.home() / '.ecorc', name='eco_startup_settings') diff --git a/slic/utils/eco_components/ecoinit.py b/slic/utils/eco_components/ecoinit.py deleted file mode 100644 index e4915ee9..00000000 --- a/slic/utils/eco_components/ecoinit.py +++ /dev/null @@ -1,49 +0,0 @@ -from .config import initFromConfigList -from . import ecocnf - - -def ecoinit(*args, _mod=None, alias_namespaces=None, components=None, lazy=None): - - #TODO: workaround for *args being used here - msg = "ecoinit: \"{}\" is mandatory, even though it is a keyword argument" - if _mod is None: - raise TypeError(msg.format("_mod")) - if alias_namespaces is None: - raise TypeError(msg.format("alias_namespaces")) - if components is None: - raise TypeError(msg.format("components")) - - if args: - allnames = [tc['name'] for tc in components] - comp_toinit = [] - for arg in args: - if not arg in allnames: - raise Exception(f'The component {arg} has no configuration defined!') - else: - comp_toinit.append(components[allnames.index(arg)]) - else: - comp_toinit = components - - if lazy is None: - lazy=ecocnf.startup_lazy - - op = {} - for key, value in initFromConfigList(comp_toinit, components, lazy=lazy).items(): - # _namespace[key] = value - _mod.__dict__[key] = value - op[key]= value - - - if not ecocnf.startup_lazy: - try: - for ta in value.alias.get_all(): - alias_namespaces.bernina.update( - ta["alias"], ta["channel"], ta["channeltype"] - ) - except: - pass -# alias_namespaces.bernina.store() - return op - - -