Source code for ctapipe.core.container

from collections import defaultdict
from copy import deepcopy
from pprint import pformat
from textwrap import wrap
import warnings
import numpy as np
from astropy.units import UnitConversionError, Quantity, Unit

import logging


log = logging.getLogger(__name__)

__all__ = ["Container", "Field", "FieldValidationError", "Map"]


[docs]class FieldValidationError(ValueError): pass
[docs]class Field: """ Class for storing data in a `Container`. Parameters ---------- default: default value of the item (this will be set when the `Container` is constructed, as well as when ``Container.reset`` is called description: str Help text associated with the item unit: str or astropy.units.core.UnitBase unit to convert to when writing output, or None for no conversion ucd: str universal content descriptor (see Virtual Observatory standards) type: type expected type of value dtype: str or np.dtype expected data type of the value, None to ignore in validation. Means value is expected to be a numpy array or astropy quantity ndim: int or None expected dimensionality of the data, for arrays, None to ignore allow_none: if the value of None is given to this Field, skip validation """ def __init__( self, default=None, description="", unit=None, ucd=None, dtype=None, type=None, ndim=None, allow_none=True, ): self.default = default self.description = description self.unit = Unit(unit) if unit is not None else None self.ucd = ucd self.dtype = np.dtype(dtype) if dtype is not None else None self.type = type self.ndim = ndim self.allow_none = allow_none def __repr__(self): desc = f"{self.description}" if self.unit is not None: desc += f" [{self.unit}]" if self.ndim is not None: desc += f" as a {self.ndim}-D array" if self.dtype is not None: desc += f" with type {self.dtype}" return desc
[docs] def validate(self, value): """ check that a given value is appropriate for this Field Parameters ---------- value: Any the value to test Raises ------ FieldValidationError: if the value is not valid """ if self.allow_none and value is None: return errorstr = f"the value '{value}' ({type(value)}) is invalid: " if self.type is not None and not isinstance(value, self.type): raise FieldValidationError( f"{errorstr} Should be an instance of {self.type}" ) if self.unit is not None: if not isinstance(value, Quantity): raise FieldValidationError( f"{errorstr} Should have units of {self.unit}" ) from None try: value.to(self.unit) except UnitConversionError as err: raise FieldValidationError(f"{errorstr}: {err}") # strip off the units now, so we can test the rest without units value = value.value if self.ndim is not None: # should be a numpy array if not isinstance(value, np.ndarray): raise FieldValidationError(f"{errorstr} Should be an ndarray") if value.ndim != self.ndim: raise FieldValidationError( f"{errorstr} Should have dimensionality {self.ndim}" ) if value.dtype != self.dtype: raise FieldValidationError( f"{errorstr} Has dtype " f"{value.dtype}, should have dtype" f" {self.dtype}" ) else: # not a numpy array if self.dtype is not None: if not isinstance(value, self.dtype.type): raise FieldValidationError( f"{errorstr} Should have numpy dtype {self.dtype}" )
[docs]class DeprecatedField(Field): """ used to mark which fields may be removed in next version """ def __init__(self, default, description="", unit=None, ucd=None, reason=""): super().__init__(default=default, description=description, unit=unit, ucd=ucd) warnings.warn(f"Field {self} is deprecated. {reason}", DeprecationWarning) self.reason = reason
class ContainerMeta(type): """ The MetaClass for `Container` It reserves __slots__ for every class variable, that is of instance `Field` and sets all other class variables as read-only for the instances. This makes sure, that the metadata is immutable, and no new fields can be added to a container by accident. """ def __new__(cls, name, bases, dct): field_names = [k for k, v in dct.items() if isinstance(v, Field)] dct["__slots__"] = tuple(field_names + ["meta", "prefix"]) dct["fields"] = {} # inherit fields from baseclasses for b in bases: if issubclass(b, Container): for k, v in b.fields.items(): dct["fields"][k] = v for k in field_names: dct["fields"][k] = dct.pop(k) new_cls = type.__new__(cls, name, bases, dct) # if prefix was not set as a class variable, build a default one if "container_prefix" not in dct: new_cls.container_prefix = name.lower().replace("container", "") return new_cls
[docs]class Container(metaclass=ContainerMeta): """Generic class that can hold and accumulate data to be passed between Components. The purpose of this class is to provide a flexible data structure that works a bit like a dict or blank Python class, but prevents the user from accessing members that have not been defined a priori (more like a C struct), and also keeps metadata information such as a description, defaults, and units for each item in the container. Containers can transform the data into a `dict` using the ``as_dict`` method. This allows them to be written to an output table for example, where each Field defines a column. The `dict` conversion can be made recursively and even flattened so that a nested set of `Containers <Container>`_ can be translated into a set of columns in a flat table without naming conflicts (the name of the parent Field is pre-pended). Only members of instance `Field` will be used as output. For hierarchical data structures, Field can use `Container` subclasses or a `Map` as the default value. >>> import astropy.units as u >>> class MyContainer(Container): ... x = Field(100, "The X value") ... energy = Field(-1, "Energy measurement", unit=u.TeV) ... >>> cont = MyContainer() >>> print(cont.x) 100 >>> # metadata will become header keywords in an output file: >>> cont.meta["KEY"] = "value" `Fields <Field>`_ inside `Containers <Container>`_ can contain instances of other containers, to allow for a hierarchy of containers, and can also contain a `Map` for the case where one wants e.g. a set of sub-classes indexed by a value like the ``telescope_id``. Examples of this can be found in `ctapipe.containers` `Container` works by shadowing all class variables (which must be instances of `Field`) with instance variables of the same name that hold the actual data. If ``reset`` is called, all instance variables are reset to their default values as defined in the class. Finally, a Container can have associated metadata via its `meta` attribute, which is a `dict` of keywords to values. """ def __init__(self, **fields): self.meta = {} # __slots__ cannot be provided with defaults # via class variables, so we use a `container_prefix` class variable # and an instance variable `prefix` in `__slots__` self.prefix = self.container_prefix for k in set(self.fields).difference(fields): # deepcopy of None is surprisingly slow default = self.fields[k].default if default is not None: default = deepcopy(default) setattr(self, k, default) for k, v in fields.items(): setattr(self, k, v) def __getitem__(self, key): return getattr(self, key) def __setitem__(self, key, value): return setattr(self, key, value) def items(self, add_prefix=False): """Generator over (key, value) pairs for the items""" if not add_prefix or self.prefix == "": return ((k, getattr(self, k)) for k in self.fields.keys()) return ((self.prefix + "_" + k, getattr(self, k)) for k in self.fields.keys()) def keys(self): """Get the keys of the container""" return self.fields.keys() def values(self): """Get the keys of the container""" return (getattr(self, k) for k in self.fields.keys()) def as_dict(self, recursive=False, flatten=False, add_prefix=False): """ Convert the `Container` into a dictionary Parameters ---------- recursive: bool sub-Containers should also be converted to dicts flatten: type return a flat dictionary, with any sub-field keys generated by appending the sub-Container name. add_prefix: bool include the container's prefix in the name of each item """ if not recursive: return dict(self.items(add_prefix=add_prefix)) else: d = dict() for key, val in self.items(add_prefix=add_prefix): if isinstance(val, Container) or isinstance(val, Map): if flatten: d.update( { f"{key}_{k}": v for k, v in val.as_dict( recursive, add_prefix=add_prefix ).items() } ) else: d[key] = val.as_dict( recursive=recursive, flatten=flatten, add_prefix=add_prefix ) else: d[key] = val return d def reset(self, recursive=True): """ Reset all values back to their default values Parameters ---------- recursive: bool If true, also reset all sub-containers """ for name, value in self.fields.items(): if isinstance(value, Container): if recursive: getattr(self, name).reset() else: setattr(self, name, deepcopy(self.fields[name].default)) def update(self, **values): """ update more than one parameter at once (e.g. `update(x=3,y=4)` or `update(**dict_of_values)`) """ for key in values: self[key] = values[key] def __str__(self): return pformat(self.as_dict(recursive=True)) def __repr__(self): text = ["{}.{}:".format(type(self).__module__, type(self).__name__)] for name, item in self.fields.items(): extra = "" if isinstance(getattr(self, name), Container): extra = ".*" if isinstance(getattr(self, name), Map): extra = "[*]" desc = "{:>30s}: {}".format(name + extra, repr(item)) lines = wrap(desc, 80, subsequent_indent=" " * 32) text.extend(lines) return "\n".join(text) def validate(self): """ Check that all fields in the Container have the expected characterisics (as defined by the Field metadata). This is not intended to be run every time a Container is filled, since it is slow, only for testing a first event. Raises ------ ValueError: if the Container's values are not valid """ for name, field in self.fields.items(): try: field.validate(self[name]) except FieldValidationError as err: raise FieldValidationError( f"{self.__class__.__name__} Field '{name}': {err}" )
[docs]class Map(defaultdict): """A dictionary of sub-containers that can be added to a Container. This may be used e.g. to store a set of identical sub-Containers (e.g. indexed by ``tel_id`` or algorithm name). """
[docs] def as_dict(self, recursive=False, flatten=False, add_prefix=False): if not recursive: return dict(self.items()) else: d = dict() for key, val in self.items(): if isinstance(val, Container) or isinstance(val, Map): if flatten: d.update( { f"{key}_{k}": v for k, v in val.as_dict( recursive, add_prefix=add_prefix ).items() } ) else: d[key] = val.as_dict( recursive=recursive, flatten=flatten, add_prefix=add_prefix ) continue d[key] = val return d
[docs] def reset(self, recursive=True): """Calls all ``Container.reset`` for all values in the Map""" for val in self.values(): if isinstance(val, Container): val.reset(recursive=recursive)