Merge pull request #14 from SOPTIM/feature/convert-link-with-measurements

Feature/convert link with measurements #13

Author: frie-soptim

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "cgmes2pgm_converter"
-version = "0.2.3"
+version = "0.3.0"
 description = "Library to convert Common Grid Model Exchange Standard (CGMES) datasets to PowerGridModel (PGM) format"
 authors = [
     { name = "Lars Friedrich, Eduard Fried, Udo Schmitz", email = "powergridmodel@soptim.de" },

src/cgmes2pgm_converter/common/__init__.py

@@ -32,6 +32,7 @@
     BranchMeasurements,
     DefaultSigma,
     IncompleteMeasurements,
+    LinkAsShortLineOptions,
     MeasSub,
     MeasurementSubstitutionOptions,
     PassiveNodeOptions,

src/cgmes2pgm_converter/common/converter_options.py

@@ -15,7 +15,10 @@
 from dataclasses import dataclass, field
 from enum import Enum
 
-from .measurement_substitution import MeasurementSubstitutionOptions
+from .measurement_substitution import (
+    LinkAsShortLineOptions,
+    MeasurementSubstitutionOptions,
+)
 from .network_splitting import NetworkSplittingOptions
 
 
@@ -60,6 +63,9 @@ class ConverterOptions:
     network_splitting: NetworkSplittingOptions = field(
         default_factory=NetworkSplittingOptions
     )
+    link_as_short_line: LinkAsShortLineOptions = field(
+        default_factory=LinkAsShortLineOptions
+    )
 
     # Deprecated: support for disabling the use of generic branches has been removed
     # Generic branches are used for all branch types

src/cgmes2pgm_converter/common/measurement_substitution.py

@@ -53,6 +53,23 @@ class SshSubstitutionOptions:
     sigma: float = 20 * 1e6
 
 
+@dataclass
+class LinkAsShortLineOptions:
+    """
+    Generate measurements for generators and loads
+    without an existing power measurement using SSH values.
+
+    Attributes:
+        enable (bool): Whether to use SSH values for substitution.
+        sigma (float): The default sigma value for the substituted measurements (MW).
+    """
+
+    enable: bool = False
+    sigma_factor: float = 1.5
+    r: float = 0.01
+    x: float = 0.01
+
+
 @dataclass
 class UMeasurementSubstitutionOptions:
     """

src/cgmes2pgm_converter/components/branch/__init__.py

@@ -17,6 +17,7 @@
 """
 
 from .equivalent_branch import EquivalentBranchBuilder
+from .gb_from_link import GenericBranchFromLinkBuilder
 from .line import LineBuilder
 from .link import LinkBuilder
 from .transformer import *

src/cgmes2pgm_converter/components/branch/gb_from_link.py

@@ -0,0 +1,105 @@
+# Copyright [2025] [SOPTIM AG]
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import numpy as np
+from power_grid_model import ComponentType, initialize_array
+
+from cgmes2pgm_converter.common import AbstractCgmesIdMapping, BranchType, CgmesDataset
+
+from ..component import AbstractPgmComponentBuilder
+
+
+class GenericBranchFromLinkBuilder(AbstractPgmComponentBuilder):
+
+    def __init__(
+        self,
+        cgmes_source: CgmesDataset,
+        id_mapping: AbstractCgmesIdMapping,
+        data_type: str = "input",
+    ):
+        super().__init__(cgmes_source, id_mapping, data_type)
+        self._component_name = ComponentType.generic_branch
+
+    def is_active(self) -> bool:
+        return self._converter_options.link_as_short_line.enable
+
+    def build_from_cgmes(self, input_data) -> tuple[np.ndarray, dict | None]:
+
+        filtered_sensors = [
+            s
+            for s in input_data[ComponentType.sym_power_sensor]
+            if self.filter_sensor(s)
+        ]
+        # link_with_sensors_indices = [i for i in input_data[ComponentType.sym_power_sensor]["measured_object"] if self.filter_link(i)]
+        link_with_sensors_indices = [
+            i
+            for i in [m["measured_object"] for m in filtered_sensors]
+            if self.filter_link(i)
+        ]
+        mask = np.isin(input_data[ComponentType.link]["id"], link_with_sensors_indices)
+        filtered_links = input_data[ComponentType.link][mask]
+
+        # Remove links
+        input_data[ComponentType.link] = input_data[ComponentType.link][~mask]
+
+        # -> add GenericBranch
+        arr = initialize_array(
+            self._data_type, self.component_name(), filtered_links.shape[0]
+        )
+
+        # arr["id"] = self._id_mapping.add_cgmes_iris(res["line"], res["name"])
+        arr["id"] = filtered_links["id"]
+        arr["from_node"] = filtered_links["from_node"]
+        arr["to_node"] = filtered_links["to_node"]
+        arr["from_status"] = filtered_links["from_status"]
+        arr["to_status"] = filtered_links["to_status"]
+        arr["r1"] = self._converter_options.link_as_short_line.r
+        arr["x1"] = self._converter_options.link_as_short_line.x
+        arr["g1"] = 0.0
+        arr["b1"] = 0.0
+        arr["k"] = 1.0
+        arr["theta"] = 0.0
+
+        for s in filtered_sensors:
+            s["p_sigma"] = (
+                self._converter_options.link_as_short_line.sigma_factor * s["p_sigma"]
+            )
+            s["q_sigma"] = (
+                self._converter_options.link_as_short_line.sigma_factor * s["q_sigma"]
+            )
+            s["power_sigma"] = (
+                self._converter_options.link_as_short_line.sigma_factor
+                * s["power_sigma"]
+            )
+
+        ## TODO: _type = kopieren aus extra_data von Link
+        extra_info = {}
+        return arr, extra_info
+
+    def filter_sensor(self, sensor) -> bool:
+        meas = sensor["measured_object"]
+        return self.filter_link(meas)
+
+    def filter_link(self, idx: int) -> bool:
+        extra = self._extra_info[idx]
+        if (
+            extra["_type"] == "Breaker"
+            or extra["_type"] == "Switch"
+            or extra["_type"] == "Disconnector"
+        ):
+            return True
+        return False
+
+    def component_name(self) -> ComponentType:
+        return self._component_name

src/cgmes2pgm_converter/components/measurement/power_sensor.py

@@ -39,6 +39,7 @@ class SymPowerBuilder(AbstractPgmComponentBuilder):
             (SAMPLE(?_eq) as ?eq)
             (SAMPLE(?_tn) as ?tn)
             ?term
+            (SAMPLE(?_eq_type) as ?eq_type)
             (SAMPLE(?_p) as ?p)
             (SAMPLE(?_q) as ?q)
             (SAMPLE(?_acc_p) as ?acc_p)
@@ -62,6 +63,8 @@ class SymPowerBuilder(AbstractPgmComponentBuilder):
         }
         FILTER(?_type_p = "ThreePhaseActivePower")
 
+        ?_eq rdf:type ?_eq_type.
+
         ?term cim:Terminal.TopologicalNode ?_tn.
 
         $TOPO_ISLAND
@@ -509,6 +512,23 @@ def get_terminal_types(
                 elif meas_node == node_to:
                     terminal_types[i] = MeasuredTerminalType.branch_to
 
+            elif (
+                eq_id in input_data[ComponentType.link]["id"]
+                and self._converter_options.link_as_short_line.enable
+            ):
+
+                node_from = input_data[ComponentType.link]["from_node"][
+                    input_data[ComponentType.link]["id"] == eq_id
+                ]
+                node_to = input_data[ComponentType.link]["to_node"][
+                    input_data[ComponentType.link]["id"] == eq_id
+                ]
+
+                if meas_node == node_from:
+                    terminal_types[i] = MeasuredTerminalType.branch_from
+                elif meas_node == node_to:
+                    terminal_types[i] = MeasuredTerminalType.branch_to
+
             elif eq_id in input_data[ComponentType.generic_branch]["id"]:
 
                 node_from = input_data[ComponentType.generic_branch]["from_node"][

src/cgmes2pgm_converter/converter.py

@@ -147,6 +147,7 @@ def _get_component_builders(
             c.SymPowerSensorIncompleteAppliance,
             c.SymLoadOrGenForPassiveNodeBuilder,
             c.SymPowerForPassiveNodeBuilder,
+            c.GenericBranchFromLinkBuilder,
         ]
         return [
             builder(self._datasource, self._id_mapping, self._options)