|
| 1 | +# License: MIT |
| 2 | +# Copyright © 2024 Frequenz Energy-as-a-Service GmbH |
| 3 | + |
| 4 | +"""Base component from which all other components inherit.""" |
| 5 | + |
| 6 | +import dataclasses |
| 7 | +import logging |
| 8 | +from collections.abc import Mapping |
| 9 | +from datetime import datetime, timezone |
| 10 | +from typing import Any, Self |
| 11 | + |
| 12 | +from frequenz.client.common.microgrid import MicrogridId |
| 13 | +from frequenz.client.common.microgrid.components import ComponentId |
| 14 | + |
| 15 | +from .._lifetime import Lifetime |
| 16 | +from ..metrics._bounds import Bounds |
| 17 | +from ..metrics._metric import Metric |
| 18 | +from ._category import ComponentCategory |
| 19 | +from ._status import ComponentStatus |
| 20 | + |
| 21 | +_logger = logging.getLogger(__name__) |
| 22 | + |
| 23 | + |
| 24 | +@dataclasses.dataclass(frozen=True, kw_only=True) |
| 25 | +class Component: # pylint: disable=too-many-instance-attributes |
| 26 | + """A base class for all components.""" |
| 27 | + |
| 28 | + id: ComponentId |
| 29 | + """This component's ID.""" |
| 30 | + |
| 31 | + microgrid_id: MicrogridId |
| 32 | + """The ID of the microgrid this component belongs to.""" |
| 33 | + |
| 34 | + category: ComponentCategory | int |
| 35 | + """The category of this component. |
| 36 | +
|
| 37 | + Note: |
| 38 | + This should not be used normally, you should test if a component |
| 39 | + [`isinstance`][] of a concrete component class instead. |
| 40 | +
|
| 41 | + It is only provided for using with a newer version of the API where the client |
| 42 | + doesn't know about a new category yet, and in case some low level code needs to |
| 43 | + know the category of a component. |
| 44 | + """ |
| 45 | + |
| 46 | + status: ComponentStatus | int = ComponentStatus.UNSPECIFIED |
| 47 | + """The status of this component. |
| 48 | +
|
| 49 | + Tip: |
| 50 | + You can also use |
| 51 | + [`is_active_now()`][frequenz.client.microgrid.component.Component.is_active_now] |
| 52 | + or |
| 53 | + [`is_active_at()`][frequenz.client.microgrid.component.Component.is_active_at], |
| 54 | + which also checks if the component is operational. |
| 55 | + """ |
| 56 | + |
| 57 | + name: str | None = None |
| 58 | + """The name of this component.""" |
| 59 | + |
| 60 | + manufacturer: str | None = None |
| 61 | + """The manufacturer of this component.""" |
| 62 | + |
| 63 | + model_name: str | None = None |
| 64 | + """The model name of this component.""" |
| 65 | + |
| 66 | + operational_lifetime: Lifetime = dataclasses.field(default_factory=Lifetime) |
| 67 | + """The operational lifetime of this component.""" |
| 68 | + |
| 69 | + rated_bounds: Mapping[Metric | int, Bounds] = dataclasses.field( |
| 70 | + default_factory=dict, |
| 71 | + # dict is not hashable, so we don't use this field to calculate the hash. This |
| 72 | + # shouldn't be a problem since it is very unlikely that two components with all |
| 73 | + # other attributes being equal would have different category specific metadata, |
| 74 | + # so hash collisions should be still very unlikely. |
| 75 | + hash=False, |
| 76 | + ) |
| 77 | + """List of rated bounds present for the component identified by Metric.""" |
| 78 | + |
| 79 | + category_specific_metadata: Mapping[str, Any] = dataclasses.field( |
| 80 | + default_factory=dict, |
| 81 | + # dict is not hashable, so we don't use this field to calculate the hash. This |
| 82 | + # shouldn't be a problem since it is very unlikely that two components with all |
| 83 | + # other attributes being equal would have different category specific metadata, |
| 84 | + # so hash collisions should be still very unlikely. |
| 85 | + hash=False, |
| 86 | + ) |
| 87 | + """The category specific metadata of this component. |
| 88 | +
|
| 89 | + Note: |
| 90 | + This should not be used normally, it is only useful when accessing a newer |
| 91 | + version of the API where the client doesn't know about the new metadata fields |
| 92 | + yet. |
| 93 | + """ |
| 94 | + |
| 95 | + def __new__(cls, *_: Any, **__: Any) -> Self: |
| 96 | + """Prevent instantiation of this class.""" |
| 97 | + if cls is Component: |
| 98 | + raise TypeError(f"Cannot instantiate {cls.__name__} directly") |
| 99 | + return super().__new__(cls) |
| 100 | + |
| 101 | + def is_active_at(self, timestamp: datetime) -> bool: |
| 102 | + """Check whether this component is active at a specific timestamp. |
| 103 | +
|
| 104 | + A component is considered active if it is in the active state and is |
| 105 | + operational at the given timestamp. The operational lifetime is used to |
| 106 | + determine whether the component is operational at the given timestamp. |
| 107 | +
|
| 108 | + If a component has an unspecified status, it is assumed to be active |
| 109 | + and a warning is logged. |
| 110 | +
|
| 111 | + Args: |
| 112 | + timestamp: The timestamp to check. |
| 113 | +
|
| 114 | + Returns: |
| 115 | + Whether this component is active at the given timestamp. |
| 116 | + """ |
| 117 | + if self.status is ComponentStatus.UNSPECIFIED: |
| 118 | + _logger.warning( |
| 119 | + "Component %s has an unspecified status. Assuming it is active.", |
| 120 | + self, |
| 121 | + ) |
| 122 | + return self.operational_lifetime.is_operational_at(timestamp) |
| 123 | + |
| 124 | + return ( |
| 125 | + self.status is ComponentStatus.ACTIVE |
| 126 | + and self.operational_lifetime.is_operational_at(timestamp) |
| 127 | + ) |
| 128 | + |
| 129 | + def is_active_now(self) -> bool: |
| 130 | + """Check whether this component is currently active. |
| 131 | +
|
| 132 | + A component is considered active if it is in the active state and is |
| 133 | + operational at the current time. The operational lifetime is used to |
| 134 | + determine whether the component is operational at the current time. |
| 135 | +
|
| 136 | + If a component has an unspecified status, it is assumed to be active |
| 137 | + and a warning is logged. |
| 138 | +
|
| 139 | + Returns: |
| 140 | + Whether this component is active at the current time. |
| 141 | + """ |
| 142 | + return self.is_active_at(datetime.now(timezone.utc)) |
| 143 | + |
| 144 | + @property |
| 145 | + def identity(self) -> tuple[ComponentId, MicrogridId]: |
| 146 | + """The identity of this component. |
| 147 | +
|
| 148 | + This uses the component ID and microgrid ID to identify a component |
| 149 | + without considering the other attributes, so even if a component state |
| 150 | + changed, the identity remains the same. |
| 151 | + """ |
| 152 | + return (self.id, self.microgrid_id) |
| 153 | + |
| 154 | + def __str__(self) -> str: |
| 155 | + """Return a human-readable string representation of this instance.""" |
| 156 | + name = f":{self.name}" if self.name else "" |
| 157 | + return f"{self.id}<{type(self).__name__}>{name}" |
0 commit comments