Transition elements to BaseElement subclasses (#16)

* Add common functionality class

* Allow elements to retrieve sensors

* Add import/export

* Add tests for design day solar

* Ignore mypy concretion errors

* Simplify thermal resistance calculation

* Add coverage

.travis.yml

@@ -2,8 +2,11 @@ language: python
 python:
   - "3.7"
   - "3.8"
-install: pip install tox-travis
+install:
+  - pip install tox-travis
+  - pip install codecov
 script: tox
+after_success: codecov
 branches:
   only:
     - master

ambient/construction.py

@@ -6,16 +6,18 @@
 # Wang and Chen
 # Building and Environment 38 (2003)
 
+from abc import abstractmethod
 from collections import defaultdict
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from functools import reduce
-from typing import Dict, List, Tuple, Union
+from typing import Dict, List, Tuple
 
 import numpy as np
 
 from numpy.polynomial.polynomial import polyzero, polyone
 
-from ambient.material import Material, MaterialResistanceOnly
+from ambient.core import BaseElement
+from ambient.material import MaterialBase, MaterialResistanceOnly
 
 _EPS = 1.0e-9
 
@@ -101,20 +103,21 @@ def _calculate_residuals(
     return delta
 
 
-@dataclass
-class ConstructionBase:
+@dataclass  # type: ignore
+class ConstructionBase(BaseElement):
     """Construction base class."""
 
     @property
+    @abstractmethod
     def thermal_resistance(self) -> float:
         """Return the total thermal resistance of the construction."""
-        raise NotImplementedError()
 
     @property
     def thermal_transmittance(self) -> float:
         """Return the total thermal transimittance of the construction."""
         return 1.0 / self.thermal_resistance
 
+    @abstractmethod
     def calculate_heat_flux_inside(
         self,
         outside_temps: np.ndarray,
@@ -123,8 +126,8 @@ def calculate_heat_flux_inside(
         current_index: int,
     ) -> float:
         """Calculated the inside heat fluxes."""
-        raise NotImplementedError()
 
+    @abstractmethod
     def calculate_heat_flux_outside(
         self,
         outside_temps: np.ndarray,
@@ -133,7 +136,6 @@ def calculate_heat_flux_outside(
         current_index: int,
     ) -> float:
         """Calculated the inside heat fluxes."""
-        raise NotImplementedError()
 
 
 @dataclass
@@ -143,28 +145,24 @@ class ConstructionLayered(ConstructionBase):
     The order of layers is outside to inside.
     """
 
-    materials: List[Union[Material, MaterialResistanceOnly]]
+    materials: List[MaterialBase] = field(default_factory=list)
     timestep: int = 3600  #: The time step for simulation [s]
 
     def __post_init__(self) -> None:
         """Set construction data not stored in fields."""
-        self._thermal_resistance = np.sum(
-            np.fromiter((m.thermal_resistance for m in self.materials), dtype=np.float)
-        )
-
         # allow fixing the minimum and maximum order of ctfs for testing
         self._min_ctf_order = 1
         self._max_ctf_order = 6
-
         self._ctfs_internal = None
-        self._is_resistance_only = all(
-            isinstance(m, MaterialResistanceOnly) for m in self.materials
-        )
 
     @property
     def thermal_resistance(self) -> float:
         """Return the thermal transimittance of the construction."""
-        return self._thermal_resistance
+        return sum(m.thermal_resistance or 0.0 for m in self.materials)
+
+    @property
+    def _is_resistance_only(self) -> bool:
+        return all(isinstance(m, MaterialResistanceOnly) for m in self.materials)
 
     @property
     def _ctfs(self) -> np.ndarray:

ambient/core.py

@@ -0,0 +1,184 @@
+"""Base classes."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field, fields
+from uuid import UUID, uuid4
+from typing import Any, ClassVar, Dict, List, Optional, Type
+
+import numpy as np
+
+
+_FUNCTION_MAP = {
+    "AVG": np.average,
+    "COUNT": np.size,
+    "MAX": np.amax,
+    "MIN": np.amin,
+    "SUM": np.sum,
+}
+
+
+@dataclass
+class BaseElement(ABC):
+    """Implements required generic functionality for all elements."""
+
+    factory: ClassVar[Dict[str, Type]] = {}
+
+    guid: UUID = field(default_factory=uuid4)
+
+    def __init_subclass__(cls: Type, **kwargs: Any) -> None:
+        """Register the subclass for file loading."""
+        super().__init_subclass__(**kwargs)  # type: ignore
+        BaseElement.factory[cls.__qualname__] = cls
+
+    def __getattribute__(self, name: str) -> None:
+        """Allow sensors to be used as input to classes."""
+        attr = super().__getattribute__(name)
+
+        if isinstance(attr, SensorBase):
+            return attr.value
+
+        return attr
+
+    def resolve_references(self, references: Dict[UUID, "BaseElement"] = None) -> None:
+        """Resolve loaded guids within the simulation."""
+        if references is None:
+            return
+
+        for fld in fields(self):
+            if fld.name == "guid":
+                continue
+
+            attr = getattr(self, fld.name)
+
+            if not isinstance(attr, str):
+                continue
+
+            try:
+                attr = UUID(attr)
+            except ValueError:
+                continue
+
+            setattr(self, fld.name, references[attr])
+
+    def get_sensor(self, name: str) -> Any:
+        """Retrieve the sensor connected to an attribute."""
+        attr = super().__getattribute__(name)
+
+        if isinstance(attr, SensorBase):
+            return attr
+
+        return None
+
+    def get_dependencies(self) -> List["BaseElement"]:
+        """Get all sensors which act as inputs to this element."""
+        return [f for f in fields(self) if isinstance(f, SensorBase)]
+
+
+@dataclass  # type: ignore
+class SensorBase(BaseElement):
+    """Base class for communication between elements."""
+
+    @property
+    @abstractmethod
+    def value(self) -> Any:
+        """Return the value of the sensor."""
+
+
+@dataclass  # type: ignore
+class SensorAttributeBase(SensorBase):
+    """Sensor for retrieving a single attribute."""
+
+    attribute: Optional[str] = None
+    function: Optional[str] = None
+
+    def __post_init__(self) -> None:
+        """Check the input."""
+        assert self.attribute or self.attribute is None
+        assert self.function is None or self.function in _FUNCTION_MAP
+
+
+@dataclass
+class SensorSISO(SensorAttributeBase):
+    """A sensor for extracting a single element attribute."""
+
+    source: Optional[BaseElement] = None
+
+    def __post_init__(self) -> None:
+        """Check the input."""
+        super().__post_init__()
+        assert self.source
+        assert isinstance(self.source, BaseElement)
+        assert self.attribute is None or hasattr(self.source, self.attribute)
+
+    def get_dependencies(self) -> List[BaseElement]:
+        """All source elements are dependencies."""
+        if self.source is None:
+            return []
+
+        if not isinstance(self.source, BaseElement):
+            raise ValueError(
+                "source is not a subclass of BaseElement. "
+                "Have references been resolved?"
+            )
+
+        return [self.source]
+
+    @property
+    def value(self) -> Any:
+        """Return the extracted value."""
+        arg = self.source
+
+        if self.attribute is not None:
+            arg = getattr(arg, self.attribute)
+
+        return arg if self.function is None else _FUNCTION_MAP[self.function](arg)
+
+
+@dataclass
+class SensorMISO(SensorAttributeBase):
+    """A sensor for extracting and aggregating element attributes."""
+
+    source: List[BaseElement] = field(default_factory=list)
+
+    def __post_init__(self) -> None:
+        """Check the input."""
+        super().__post_init__()
+        assert self.source
+        assert isinstance(self.source, (list, tuple))
+        assert self.attribute is None or all(
+            hasattr(e, self.attribute) for e in self.source
+        )
+
+    def resolve_references(self, references: Dict[UUID, BaseElement] = None) -> None:
+        """Resolve loaded guids within the simulation."""
+        super().resolve_references(references)
+
+        if references is None:
+            return
+
+        self.source = [
+            element
+            if isinstance(element, BaseElement)
+            else references[UUID(str(element))]
+            for element in self.source
+        ]
+
+    def get_dependencies(self) -> List[BaseElement]:
+        """All source elements are dependencies."""
+        if any(not isinstance(element, BaseElement) for element in self.source):
+            raise ValueError(
+                "an element of source is not a subclass of BaseElement. "
+                "Have references been resolved?"
+            )
+
+        return self.source
+
+    @property
+    def value(self) -> Any:
+        """Return the aggregated value."""
+        args = self.source
+
+        if self.attribute is not None:
+            args = [getattr(e, self.attribute) for e in args]
+
+        return args if self.function is None else _FUNCTION_MAP[self.function](args)

ambient/io.py

@@ -0,0 +1,53 @@
+"""Input output module."""
+
+import json
+from dataclasses import fields
+from typing import Any, Dict
+from uuid import UUID
+
+from ambient.core import BaseElement
+from ambient.simulation import Simulation
+
+
+# don't use dataclasses asdict function since we don't want
+# to recurse into each instance.
+def _field_dict(obj: Any) -> Dict[str, Any]:
+    return {
+        "type": type(obj).__qualname__,
+        "data": {
+            f.name: obj.get_sensor(f.name) or getattr(obj, f.name) for f in fields(obj)
+        },
+    }
+
+
+class SimulationEncoder(json.JSONEncoder):
+    """Encode simulations as JSON."""
+
+    def default(self, o: Any) -> Any:  # pylint: disable=method-hidden
+        """Handle simulation specific classes."""
+        if isinstance(o, UUID):
+            return f"{o.urn}"
+
+        if isinstance(o, Simulation):
+            sim = _field_dict(o)
+            sim["data"]["elements"] = {
+                e.guid.urn: _field_dict(e) for e in o.elements.values()
+            }
+            return sim
+
+        if isinstance(o, BaseElement):
+            return o.guid
+
+        return json.JSONEncoder.default(self, o)
+
+
+def simulation_decoder(dct: Dict[str, Any]) -> Dict[str, Any]:
+    """Rebuild simulation classes on import."""
+    if "guid" in dct:
+        dct["guid"] = UUID(dct["guid"])
+
+    if "type" in dct and "data" in dct:
+        factory = BaseElement.factory[dct["type"]]
+        return factory(**dct["data"])
+
+    return dct