Merge pull request #44 from splightplatform/mgaragiola/hypersim-reader

Mgaragiola/hypersim reader

Author: marianogaragiola

simulation/hypersim/__init__.py

@@ -0,0 +1,25 @@
+from datetime import datetime, timezone
+from logging import getLogger
+
+from hypersim.interfaces import DataReader, DataSaver
+
+logger = getLogger("HypersimOperator")
+
+
+class HypersimConnector:
+    def __init__(self, reader: DataReader):
+        self._reader = reader
+        self._data_savers: list[DataSaver] = []
+
+    def add_data_saver(
+        self,
+        saver: DataSaver,
+    ) -> None:
+        self._data_savers.append(saver)
+
+    def process(self) -> None:
+        logger.debug("Reading data from Hypersim")
+        now = datetime.now(timezone.utc)
+        data = self._reader.read()
+        for saver in self._data_savers:
+            saver.process_data(data, now)

simulation/hypersim/data_connector.py

@@ -0,0 +1,42 @@
+from datetime import datetime
+from logging import getLogger
+
+from splight_lib.models import Asset
+
+# TODO: Update to use v4 models
+from splight_lib.models._v3.native import Number
+
+logger = getLogger("HypersimOperator")
+
+
+class DeviceDataSaver:
+    def __init__(self, asset: Asset):
+        self._asset = asset
+        self._attributes = {item.name: item for item in asset.attributes}
+        self._attr_sensor_map: dict[str, str] = {}
+
+    def add_attribute(self, attribute: str, sensor: str) -> None:
+        if attribute not in self._attributes:
+            raise ValueError(f"Attribute {attribute} not found in asset.")
+        if attribute in self._attr_sensor_map:
+            logger.debug(f"Attribute {attribute} already added.")
+        self._attr_sensor_map[sensor] = self._attributes[attribute]
+
+    def process_data(
+        self, data: dict[str, float], date: datetime
+    ) -> dict[str, float]:
+        for sensor, attribute in self._attr_sensor_map.items():
+            sensor_value = data.get(sensor, None)
+            if sensor_value is None:
+                logger.debug(f"Sensor {sensor} not found in data.")
+                continue
+            logger.debug(
+                f"Saving data for sensor {sensor} with value {sensor_value}"
+            )
+            attr_value = Number(
+                timestamp=date,
+                asset=self._asset.id,
+                attribute=attribute.id,
+                value=sensor_value,
+            )
+            attr_value.save()

simulation/hypersim/data_saver.py

@@ -0,0 +1,11 @@
+from typing import Protocol
+
+
+class DataReader(Protocol):
+    def read(self) -> list[float]:
+        pass
+
+
+class DataSaver(Protocol):
+    def save(self, data: list[float]) -> None:
+        pass

simulation/hypersim/interfaces.py

@@ -0,0 +1,205 @@
+from datetime import datetime, timezone
+from logging import getLogger
+from time import time
+from typing import Optional, Tuple, TypedDict
+
+import HyWorksApiGRPC as HyWorksApi
+import requests
+from splight_lib.models import Asset
+from splight_lib.settings import workspace_settings
+
+from hypersim.data_saver import DeviceDataSaver
+from hypersim.interfaces import DataReader
+from hypersim.reader import AssetAttributeDataReader
+
+logger = getLogger("HypersimOperator")
+GENERATOR_VECTOR_NAME = "generators_vector"
+LINE_CONTINGENCY = "contingency"
+
+
+class LineInfo(TypedDict):
+    name: str
+    asset: str
+    breaker: str
+
+
+class GeneratorInfo(TypedDict):
+    name: str
+    asset: str
+    breaker: str
+
+
+def set_device_value(
+    device: str, variable: str, value: str | int | float
+) -> None:
+    """Sets the value of a device variable in Hypersim."""
+    HyWorksApi.setComponentParameter(device, variable, str(value))
+    logger.debug(f"Setting {device}.{variable} to {value}")
+
+
+class DCMHypersimOperator:
+    def __init__(
+        self,
+        grid: str,
+        lines: list[LineInfo],
+        generators: list[GeneratorInfo],
+        hypersim_reader: DataReader,
+    ):
+        self._grid = grid
+        addresses = []
+        breakers = []
+        self._savers = []
+        for line_info in lines:
+            line = Asset.retrieve(line_info["asset"])
+            attrs = {attr.name: attr for attr in line.attributes}
+            address = {
+                "asset": line.id,
+                "attribute": attrs[GENERATOR_VECTOR_NAME].id,
+            }
+            addresses.append(address)
+            breakers.append(line_info["breaker"])
+            data_saver = DeviceDataSaver(line)
+            data_saver.add_attribute(
+                LINE_CONTINGENCY, line_info["breaker"]
+            )
+            self._savers.append(data_saver)
+        self._lines = lines
+        self._lines_breakers = breakers
+        self._generators = {item["name"]: item for item in generators}
+        self._generators_vector: dict[str, list[int]] = {}
+        self._hy_reader = hypersim_reader
+        self._spl_reader = AssetAttributeDataReader(
+            addresses, data_type="String", limit=1
+        )
+
+        self._in_contingency = False
+        self._last_contingency: datetime | None = None
+
+    def run(self) -> None:
+        t0 = time()
+        contingency = self._check_for_contingency()
+        new_contingency = self._process_contingency(contingency)
+        t1 = time()
+        if new_contingency:
+            logger.info(
+                f"\n\n\nOperation time: {(t1 - t0) * 1000:.3f} ms\n\n\n"
+            )
+
+    def _process_contingency(
+        self, contingency: Optional[Tuple[str, int]]
+    ) -> bool:
+        new_contingency = False
+        if self._in_contingency:
+            if contingency is None:
+                logger.info("Recovering system from contingency")
+                self._recover_system()
+                self._in_contingency = False
+            else:
+                logger.info(
+                    "System still in contingency. Waiting for recovery"
+                )
+        else:
+            if contingency is not None:
+                logger.info(f"Contingency found on line {contingency[0]}")
+                self._run_operation(contingency[0])
+                self._set_contingency(contingency[0])
+                new_contingency = True
+                self._in_contingency = True
+                self._last_contingency = datetime.now(timezone.utc)
+            else:
+                logger.debug("No contingency found.")
+        return new_contingency
+
+    def _set_contingency(self, line_breaker: str) -> None:
+        date = datetime.now(timezone.utc)
+        for saver in self._savers:
+            saver.process_data({line_breaker: True}, date)
+
+    def _check_for_contingency(self) -> Optional[Tuple[str, int]]:
+        breakers_status = self._hy_reader.read()
+        line_in_contingency = next(
+            filter(
+                lambda x: x[1] == 0 and x[0] in self._lines_breakers,
+                breakers_status.items(),
+            ),
+            None,
+        )
+        return line_in_contingency
+
+    def update_operation_vectors(self) -> None:
+        data = self._spl_reader.read()
+        for line_id, vector in data.items():
+            line_name = next(
+                filter(lambda x: x["asset"] == line_id, self._lines)
+            )
+            self._generators_vector.update(
+                {line_name["breaker"]: self._parse_generator_vector(vector)}
+            )
+        logger.info(f"Operation vectors updated: {self._generators_vector}")
+
+    def _run_operation(self, line_breaker: str) -> None:
+        vector = self._generators_vector.get(line_breaker, None)
+        if vector is None:
+            line_name = next(
+                filter(lambda x: x["breaker"] == line_breaker, self._lines)
+            )["name"]
+            raise ValueError(f"No operation vector found for line {line_name}")
+        self._apply_vector(vector)
+
+    def _apply_vector(self, vector: dict[str, int]) -> None:
+        logger.info(f"Applying operation vector: {vector}")
+        for gen_name, value in vector.items():
+            if value == 0:
+                continue
+            # In Hypersim, the setpoint is 0 to open the breaker and 7 to
+            # close it
+            setpoint = 0 if value == 1 else 7
+            generator = self._generators.get(gen_name, None)
+            # TODO: Check if generator is None
+            block, variable = generator["breaker"].split(".")
+            set_device_value(block, variable, setpoint)
+            logger.debug(f"Setting generator {gen_name} to {setpoint}")
+
+    def _parse_generator_vector(self, vector: str) -> list[int]:
+        generator_ordering = self._fetch_gen_ordering(self._grid)
+        splitted_vector = [int(x) for x in vector.split(",")]
+        sorted_gens = []
+        for gen in generator_ordering:
+            gen_id = gen["id"]
+            full_gen = next(
+                filter(
+                    lambda x: x["asset"] == gen_id, self._generators.values()
+                )
+            )
+            sorted_gens.append(full_gen)
+        parsed_vector = {
+            gen["name"]: value
+            for gen, value in zip(sorted_gens, splitted_vector)
+        }
+        return parsed_vector
+
+    def _recover_system(self) -> None:
+        date = datetime.now(timezone.utc)
+        for generator in self._generators.values():
+            block, variable = generator["breaker"].split(".")
+            set_device_value(block, variable, 7)
+            logger.debug(f"Closing breaker for generator {generator['name']}")
+
+        for saver in self._savers:
+            saver.process_data(
+                {breaker: False for breaker in self._lines_breakers}, date
+            )
+
+    @staticmethod
+    def _fetch_gen_ordering(grid_id: str) -> list[str]:
+        host = workspace_settings.SPLIGHT_PLATFORM_API_HOST
+        url = f"{host}/v4/engine/asset/grids/{grid_id}/structure/"
+        access_id = workspace_settings.SPLIGHT_ACCESS_ID
+        secret_key = workspace_settings.SPLIGHT_SECRET_KEY
+        header = {
+            "Authorization": f"Splight {access_id} {secret_key}",
+        }
+        response = requests.get(url, headers=header)
+        response.raise_for_status()
+        data = response.json()
+        return data["generators"]

simulation/hypersim/operator.py

@@ -0,0 +1,99 @@
+from datetime import datetime, timedelta, timezone
+from logging import getLogger
+from typing import TypedDict
+
+import HyWorksApiGRPC as HyWorksApi
+from splight_lib.models._v3.datalake import DataRequest, PipelineStep, Trace
+from splight_lib.models._v3.native import Boolean, Number, String
+from tenacity import retry, wait_fixed
+
+logger = getLogger("HypersimOperator")
+
+TYPE_MAP = {
+    "number": Number,
+    "string": String,
+    "boolean": Boolean,
+}
+
+
+class DataAddress(TypedDict):
+    asset: str
+    attribute: str
+
+
+class HypersimDataReader:
+    def __init__(self, sensors: list[str] | None = None):
+        self._connect()
+        self._sensors: set[str] = set(sensors) if sensors else set()
+        self._data: dict[str, float] = {}
+
+    def add_sensor(self, sensor: str) -> None:
+        if sensor in self._sensors:
+            logger.debug(f"Sensor {sensor} already added.")
+        self._sensors.add(sensor)
+
+    def update_data(self) -> None:
+        try:
+            values = self._read_sensor_values()
+        except Exception as e:
+            logger.error(f"Error reading sensors: {e}")
+            self._connect()
+            raise e
+        if len(values) != len(self._sensors):
+            raise ValueError(
+                (
+                    "An error occurred while reading sensors. The number of "
+                    "read values does not match the number of sensors."
+                )
+            )
+        self._data = {key: value for key, value in zip(self._sensors, values)}
+
+    def read(self) -> dict[str, float]:
+        return self._data
+
+    @retry(wait=wait_fixed(0.1))
+    def _read_sensor_values(self) -> list[float]:
+        values = HyWorksApi.getLastSensorValues(list(self._sensors))
+        return values
+
+    def _connect(self) -> None:
+        HyWorksApi.startAndConnectHypersim()
+        return None
+
+
+class AssetAttributeDataReader:
+    def __init__(
+        self,
+        addresses: list[DataAddress],
+        data_type: str = "Number",
+        limit: int = 1000,
+    ):
+        self._addresses = addresses
+        self._limit = limit
+        self._data_class = TYPE_MAP.get(data_type.lower(), Number)
+
+    def add_address(self, address: DataAddress) -> None:
+        # TODO: validate if address already exists
+        self._addresses.append(address)
+
+    def read(self):
+        now = datetime.now(timezone.utc)
+        from_ts = now - timedelta(minutes=30)
+        request = DataRequest[self._data_class](
+            from_timestamp=from_ts,
+            to_timestamp=now,
+        )
+        for address in self._addresses:
+            trace = Trace.from_address(address["asset"], address["attribute"])
+            trace.add_step(PipelineStep.from_dict({"$limit": self._limit}))
+            request.add_trace(trace)
+        data = request.apply()
+        if len(data) != len(self._addresses):
+            raise ValueError(
+                (
+                    "An error occurred while reading attributes. "
+                    "The number of read values does not match the "
+                    "number of addresses."
+                )
+            )
+        return {item.asset: item.value for item in data}