Source code for simframe.io.writers.namespacewriter

from simframe.frame.field import Field
from simframe.io.reader import Reader
from simframe.io.writer import Writer
from simframe.utils.color import colorize
from simframe.utils.simplenamespace import SimpleNamespace

from collections import deque
import copy
import numbers
import numpy as np


[docs] class namespacewriter(Writer): """Class to write ``Frame`` object to namespace""" def __init__(self, *args, **kwargs): super().__init__(_writeframetonamespace, dumping=False, description="Temporary namespace writer", reader=namespacereader, *args, **kwargs) self._buffer = deque([]) def __repr__(self): ret = self.__str__()+"\n" ret += "-" * (len(ret)-1) + "\n" ret += " Data directory : {}\n".format(self.datadir) ret += " Dumping : {}\n".format( colorize(self.dumping, "yellow") if not self.dumping else self.dumping) ret += " Verbosity : {}".format(self.verbosity) return ret def _getfilename(self): """Filenames are not required for this class.""" pass
[docs] def write(self, owner, i=0, *args, **kwargs): """Writes output to namespace Parameters ---------- owner : Frame Parent frame object i : int Not used in this class forceoverwrite : boolean Not used in this class filename : string Not used in this class""" self._buffer.append(self._func(owner)) if self.verbosity > 0: num = str(i).zfill(self._zfill) msg = "Saving frame {}".format(num) print(msg) if self.dumping: self.writedump(owner)
[docs] def reset(self): """This resets the namespace. Notes ----- WARNING: This cannot be undone.""" self._buffer.clear()
class namespacereader(Reader): """Class to read namespace outputs""" def __init__(self, writer): super().__init__(writer) def all(self): """Functions that reads all output files and combines them into a single ``SimpleNamespace``. Returns ------- dataset : SimpleNamespace Namespace of data set. Notes ----- This function is reading one output files to get the structure of the data and calls ``read.sequence()`` for every field in the data structure.""" if self._writer._buffer == deque([]): raise RuntimeError("Writer buffer is empty.") # Read first file to get structure data0 = self._writer._buffer[0] return self._expand(data0) def output(self, i): """Reading a single output Parameters ---------- i : int Number of output to be read Returns ------- n : SimpleNamespace Namespace of desired output""" if i >= len(self._writer._buffer): raise RuntimeError("Output {} does not exist.".format(i)) return self._writer._buffer[i] def sequence(self, field): """Reading the entire sequence of a specific field. Parameters ---------- field : string String with location of requested field Returns ------- seq : array Array with requested values Notes ----- ``field`` is addressing the values just as in the parent frame object. E.g. ``"groupA.groupB.fieldC"`` is addressing ``Frame.groupA.groupB.fieldC``.""" if self._writer._buffer == deque([]): raise RuntimeError("Writer buffer is empty.") if not isinstance(field, str): raise TypeError("<field> has to be string.") loc = field.split(".") N = len(self._writer._buffer) ret = [] for i in range(N): A = np.array(_getvaluefrombuffer(self._writer._buffer[i], loc)) ret.append(A) return np.array(ret) def _getvaluefrombuffer(buf, loc): """Returns a requested value from buffer. Function is called recursively. Parameters ---------- buf : dict Buffer object loc : list List of strings with the requested location within buf Returns ------- ret : object Reqested value in buf at position loc""" if len(loc) > 1: return _getvaluefrombuffer(buf.__dict__[loc[0]], loc[1:]) if not hasattr(buf, loc[0]): raise RuntimeError("Requested <field> does not exist.") return buf.__dict__[loc[0]] def _converttonamespace(o): """Converts an object into a namespace Parameters ---------- o : object object Returns ------- ns : SimpleNamespace Nested namespace with the data in Frame object Notes ----- Attributes beginning with underscore _ are being ignored. So are fields with Field.save == False.""" ret = {} # These things are written directy into the dictionary. direct = (numbers.Number, np.number, tuple, list, np.ndarray, str) for key, val in o.__dict__.items(): # Ignore hidden variables if key.startswith("_"): continue # Skip fields that should not be stored if isinstance(val, Field) and val.save == False: continue if val is not None and isinstance(val, direct): ret[key] = copy.copy(val) else: ret[key] = _converttonamespace(val) return SimpleNamespace(**ret) def _writeframetonamespace(frame): """Takes a list of dicts and a frame, stitches them together and returns namespace. Paramters --------- frame : Frame Frame object to add Returns ------- n : SimpleNamespace Namespace with data""" d = _converttonamespace(frame) return d