Source code for

import logging
import typing as t
from abc import ABC, abstractmethod
from import Iterable
from copy import copy
from pathlib import Path
from time import perf_counter
from typing import Union

from flatten_dict import flatten
from yaml import dump, safe_load

from agora.logging import timer

atomic = t.Union[int, float, str, bool]

[docs]class ParametersABC(ABC): """ Defines parameters as attributes and allows parameters to be converted to either a dictionary or to yaml. No attribute should be called "parameters"! """
[docs] def __init__(self, **kwargs): """ Defines parameters as attributes """ assert ( "parameters" not in kwargs ), "No attribute should be named parameters" for k, v in kwargs.items(): setattr(self, k, v)
[docs] def to_dict(self, iterable="null") -> t.Dict: """ Recursive function to return a nested dictionary of the attributes of the class instance. """ if isinstance(iterable, dict): if any( [ True for x in iterable.values() if isinstance(x, Iterable) or hasattr(x, "to_dict") ] ): return { k: v.to_dict() if hasattr(v, "to_dict") else self.to_dict(v) for k, v in iterable.items() } else: return iterable elif iterable == "null": # use instance's built-in __dict__ dictionary of attributes return self.to_dict(self.__dict__) else: return iterable
[docs] def to_yaml(self, path: Union[Path, str] = None): """ Returns a yaml stream of the attributes of the class instance. If path is provided, the yaml stream is saved there. Parameters ---------- path : Union[Path, str] Output path. """ if path: with open(Path(path), "w") as f: dump(self.to_dict(), f) return dump(self.to_dict())
@classmethod def from_dict(cls, d: dict): return cls(**d)
[docs] @classmethod def from_yaml(cls, source: Union[Path, str]): """ Returns instance from a yaml filename or stdin """ is_buffer = True try: if Path(source).exists(): is_buffer = False except Exception as _: assert isinstance(source, str), "Invalid source type." if is_buffer: params = safe_load(source) else: with open(source) as f: params = safe_load(f) return cls(**params)
@classmethod def default(cls, **kwargs): overriden_defaults = copy(cls._defaults) for k, v in kwargs.items(): overriden_defaults[k] = v return cls.from_dict(overriden_defaults)
[docs] def update(self, name: str, new_value): """ Update values recursively if name is a dictionary, replace data where existing found or add if not. It warns against type changes. If the existing structure under name is a dictionary, it looks for the first occurrence and modifies it accordingly. If a leaf node that is to be changed is a collection, it adds the new elements. """ assert name not in ( "parameters", "params", ), "Attribute can't be named params or parameters" if name in self.__dict__: if check_type_recursive(getattr(self, name), new_value): print("Warnings:Type changes are risky") if isinstance(getattr(self, name), dict): flattened = flatten(self.to_dict()) names_found = [k for k in flattened.keys() if name in k] found_idx = [keys.index(name) for keys in names_found] assert len(names_found), f"{name} not found as key." keys = None if len(names_found) > 1: for level in zip(found_idx, names_found): if level == min(found_idx): keys = level print( f"Warning: {name} was found in multiple keys. Selected {keys}" ) break else: keys = names_found.pop() if keys: current_val = flattened.get(keys, None) # if isinstance(current_val, t.Collection): elif isinstance(getattr(self, name), t.Collection): add_to_collection(getattr(self, name), new_value) elif isinstance(getattr(self, name), set): pass # TODO implement new_d = getattr(self, name) new_d.update(new_value) setattr(self, name, new_d) else: setattr(self, name, new_value)
[docs]def add_to_collection( collection: t.Collection, value: t.Union[atomic, t.Collection] ): # Adds element(s) in place. if not isinstance(value, t.Collection): value = [value] if isinstance(collection, list): collection += value elif isinstance(collection, set): collection.update(value)
[docs]class ProcessABC(ABC): """ Base class for processes. Defines parameters as attributes and requires run method to be defined. """
[docs] def __init__(self, parameters): """ Arguments --------- parameters: instance of ParametersABC """ self._parameters = parameters # convert parameters to dictionary # and then define each parameter as an attribute for k, v in parameters.to_dict().items(): setattr(self, k, v)
@property def parameters(self): return self._parameters @abstractmethod def run(self): pass def _log(self, message: str, level: str = "warning"): # Log messages in the corresponding level logger = logging.getLogger("aliby") getattr(logger, level)(f"{self.__class__.__name__}: {message}")
[docs]def check_type_recursive(val1, val2): same_types = True if not isinstance(val1, type(val2)) and not all( type(x) in (Path, str) for x in (val1, val2) # Ignore str->path ): return False if not isinstance(val1, t.Iterable) and not isinstance(val2, t.Iterable): return isinstance(val1, type(val2)) elif isinstance(val1, (tuple, list)) and isinstance(val2, (tuple, list)): return bool( sum([check_type_recursive(v1, v2) for v1, v2 in zip(val1, val2)]) ) elif isinstance(val1, dict) and isinstance(val2, dict): if not len(val1) or not len(val2): return False for k in val2.keys(): same_types = same_types and check_type_recursive(val1[k], val2[k]) return same_types
[docs]class StepABC(ProcessABC): """ Base class that expands on ProcessABC to include tools used by Aliby steps. It adds a setup step, logging and benchmarking for time benchmarks. """
[docs] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs)
@abstractmethod def _run_tp(self): pass @timer def run_tp(self, tp: int, **kwargs): """ Time and log the timing of a step. """ return self._run_tp(tp, **kwargs) def run(self): # Replace run with run_tp raise Warning("Steps use run_tp instead of run")