-
Notifications
You must be signed in to change notification settings - Fork 36
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Creating base pydantic model and updating nodes.py to extend. Include…
…s CIM mapping
- Loading branch information
Showing
4 changed files
with
112 additions
and
245 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,137 +1,54 @@ | ||
| # -*- coding: utf-8 -*- | ||
| from __future__ import absolute_import, division, print_function | ||
| from builtins import super, range, zip, round, map | ||
| from pydantic import BaseModel, Field, ValidationError | ||
| from pydantic.json import isoformat, timedelta_isoformat | ||
|
|
||
| import warnings | ||
| from traitlets.traitlets import ( | ||
| _CallbackWrapper, | ||
| EventHandler, | ||
| ) | ||
| import traitlets as T | ||
|
|
||
| import logging | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class DiTToHasTraits(T.HasTraits): | ||
|
|
||
| response = T.Any(allow_none=True, help="default trait for managing return values") | ||
|
|
||
| def __init__(self, model, *args, **kwargs): | ||
| model.model_store.append(self) | ||
| self.build(model) | ||
| super().__init__(*args, **kwargs) | ||
|
|
||
| def set_name(self, model): | ||
| try: | ||
| name = self.name | ||
| if name in model.model_names: | ||
| warnings.warn("Duplicate name %s being set. Object overwritten." % name) | ||
| logger.debug("Duplicate name %s being set. Object overwritten." % name) | ||
| logger.debug(model.model_names[name], self) | ||
| model.model_names[name] = self | ||
| except AttributeError: | ||
| pass | ||
|
|
||
| def build(self, model): | ||
| raise NotImplementedError( | ||
| "Build function must be implemented by derived classes" | ||
| ) | ||
|
|
||
| def notify_access(self, bunch): | ||
| if not isinstance(bunch, T.Bunch): | ||
| # cast to bunch if given a dict | ||
| bunch = T.Bunch(bunch) | ||
| name, type = bunch.name, bunch.type | ||
|
|
||
| callables = [] | ||
| callables.extend(self._trait_notifiers.get(name, {}).get(type, [])) | ||
| callables.extend(self._trait_notifiers.get(name, {}).get(T.All, [])) | ||
| callables.extend(self._trait_notifiers.get(T.All, {}).get(type, [])) | ||
| callables.extend(self._trait_notifiers.get(T.All, {}).get(T.All, [])) | ||
|
|
||
| # Call them all now | ||
| # Traits catches and logs errors here. I allow them to raise | ||
| if len(callables) > 1: | ||
| raise TypeError( | ||
| "Maximum number of callables allowed for a single attribute using the 'fetch' event is 1. Please check the documentation of DiTTo" | ||
| ) | ||
| for c in callables: | ||
| # Bound methods have an additional 'self' argument. | ||
|
|
||
| if isinstance(c, _CallbackWrapper): | ||
| c = c.__call__ | ||
| elif isinstance(c, EventHandler) and c.name is not None: | ||
| c = getattr(self, c.name) | ||
|
|
||
| return c(bunch) | ||
|
|
||
|
|
||
| class DiTToTraitType(T.TraitType): | ||
|
|
||
| allow_none = True | ||
|
|
||
| def get(self, obj, cls=None): | ||
| # Call notify_access with event type fetch | ||
| # If and only if one event exists, a return value will be produced | ||
| # This return value is saved as the current value in obj._trait_values | ||
| # Then call super get | ||
| try: | ||
| r = obj.notify_access( | ||
| T.Bunch( | ||
| name=self.name, | ||
| value=obj._trait_values[self.name], | ||
| owner=self, | ||
| type="fetch", | ||
| ) | ||
| ) | ||
|
|
||
| old_value = obj._trait_values[self.name] | ||
| from .units import Voltage, VoltageUnit, Phase | ||
| from .position import Position | ||
|
|
||
| if r is not None and r != old_value: | ||
| logger.debug( | ||
| "Response from callback event 'fetch' on property {} does not match previous value. Overloading existing value {} with new value {}".format( | ||
| self.name, old_value, r | ||
| ) | ||
| ) | ||
| obj._trait_values[self.name] = r | ||
| except KeyError: | ||
| pass | ||
|
|
||
| return super().get(obj, cls=cls) | ||
|
|
||
|
|
||
| class Float(T.Float, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class Complex(T.Complex, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class Unicode(T.Unicode, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class Any(T.Any, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class Int(T.Int, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class List(T.List, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class Instance(T.Instance, DiTToTraitType): | ||
| pass | ||
|
|
||
|
|
||
| class Bool(T.Bool, DiTToTraitType): | ||
| pass | ||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| observe = T.observe | ||
| class DiTToBaseModel(BaseModel): | ||
| """ Base pydantic class for all DiTTo models. | ||
| A name is required for all DiTTo models. | ||
| """ | ||
|
|
||
| class Config: | ||
| """ Base pydantic configuration for all DiTTo models. | ||
| """ | ||
| title = "DiTToBaseModel" | ||
| validate_assignment = True | ||
| validate_all = True | ||
| extra = "forbid" | ||
| use_enum_values = True | ||
| arbitrary_types_allowed = True | ||
| allow_population_by_field_name = True | ||
|
|
||
| name: str = Field( | ||
| description="Name of the element in the DiTTo model", | ||
| title="name", | ||
| cim_value="name" | ||
| ) | ||
|
|
||
| substation_name: Optional[str] = Field( | ||
| description="Name of the substation the element is under", | ||
| title="substation_name" | ||
| cim_value="EquipmentContainer.Substation.name" | ||
| ) | ||
|
|
||
| feeder_name: Optional[str] = Field( | ||
| description="Name of the feeder the element is on", | ||
| title="feeder_name" | ||
| cim_value="EquipmentContainer.name" | ||
| ) | ||
|
|
||
| positions: Optional[List[Position]] = Field( | ||
| description="A list of positions of the element. For single point elements, this list should have a length of one. For lines, this list may contain intermediate points.", | ||
| title="positions" | ||
| cim_value="Location.PositionPoints" | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| # -*- coding: utf-8 -*- | ||
| from __future__ import absolute_import, division, print_function | ||
| from builtins import super, range, zip, round, map | ||
| from pydantic import BaseModel, Field, ValidationError | ||
| from pydantic.json import isoformat, timedelta_isoformat | ||
|
|
||
| import warnings | ||
| import logging | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| class Voltage(BaseModel): | ||
| """This class is used to represent the voltage at a node or a line. It is a simple container for the voltage value and the unit. | ||
| """ | ||
| value: float = Field( | ||
| title="value", | ||
| description="The value of the voltage", | ||
| ) | ||
| unit: VoltageUnit = Field( | ||
| description="The unit of the voltage", | ||
| title="unit", | ||
| default="V", | ||
| ) | ||
|
|
||
| class VoltageUnit(str, Enum): | ||
| """This class is used to represent the possible units of voltage. | ||
| """ | ||
| V = "V" | ||
| kV = "kV" | ||
| MV = "MV" | ||
|
|
||
|
|
||
| class Phase(str, Enum): | ||
| """This class is used to represent a single phase from a set of possible values. | ||
| """ | ||
| A = "A" | ||
| B = "B" | ||
| C = "C" | ||
| N = "N" | ||
| s1 = "s1" | ||
| s2 = "s2" |
Oops, something went wrong.