fixed syntax err

Author: omari91

cim_parser.py

@@ -2,7 +2,7 @@
 import time
 from typing import List
 
-from main import ConnectivityNode, ACLineSegment, EnergyConsumer, GridTopology, FastGridEngine
+from main import ConnectivityNode, ACLineSegment, EnergyConsumer, GridTopology, FastGridEngine, Terminal, PerLengthPhaseImpedance
 
 CIM_NS = {'cim': 'http://iec.ch/TC57/2013/CIM-schema-cim16#',
           'rdf': 'http://www.w3.org/1999/02/22-rdf-syntax-ns#'}
@@ -11,82 +11,75 @@
 def parse_cim_xml(file_path: str) -> GridTopology:
     """
     Parses a CIM XML (RDF) file and returns a validated GridTopology object.
-    Note: Real CIM standard uses Terminals to link ConductingEquipment to ConnectivityNodes.
-    For this simplified example, we're using direct schema extensions. A full CIMTool profile
-    would generate the precise mapping required.
+    Translating basic XML elements into the new strict Terminal linkages and Impedance properties.
     """
     print(f"[*] Parsing CIM XML file: {file_path}")
     tree = ET.parse(file_path)
     root = tree.getroot()
 
     nodes: List[ConnectivityNode] = []
+    terminals: List[Terminal] = []
     segments: List[ACLineSegment] = []
     consumers: List[EnergyConsumer] = []
+    impedance_catalog = {}
 
     # 1. Parse Connectivity Nodes
     for node_elem in root.findall('cim:ConnectivityNode', CIM_NS):
         node_id = node_elem.attrib.get(f"{{{CIM_NS['rdf']}}}ID")
-
-        voltage_elem = node_elem.find(
-            'cim:ConnectivityNode.voltageLevelKV', CIM_NS)
+        voltage_elem = node_elem.find('cim:ConnectivityNode.voltageLevelKV', CIM_NS)
         v_kv = float(voltage_elem.text) if voltage_elem is not None else 20.0
-
         type_elem = node_elem.find('cim:ConnectivityNode.type', CIM_NS)
         n_type = type_elem.text if type_elem is not None else 'PQ'
 
-        nodes.append(
-            ConnectivityNode(
-                id=node_id,
-                voltage_level_kv=v_kv,
-                type=n_type))
+        nodes.append(ConnectivityNode(id=node_id, voltage_level_kv=v_kv, type=n_type))
 
     # 2. Parse ACLineSegments
     for line_elem in root.findall('cim:ACLineSegment', CIM_NS):
         line_id = line_elem.attrib.get(f"{{{CIM_NS['rdf']}}}ID")
 
         from_node = line_elem.find('cim:ACLineSegment.fromNode', CIM_NS).attrib.get(
-            f"{CIM_NS['rdf']}  resource").strip('#')
+            f"{{{CIM_NS['rdf']}}}resource").strip('#')
         to_node = line_elem.find('cim:ACLineSegment.toNode', CIM_NS).attrib.get(
-            f"{CIM_NS['rdf']}  resource").strip('#')
+            f"{{{CIM_NS['rdf']}}}resource").strip('#')
 
-        length_km = float(
-            line_elem.find(
-                'cim:ACLineSegment.length',
-                CIM_NS).text)
+        length_km = float(line_elem.find('cim:ACLineSegment.length', CIM_NS).text)
         r = float(line_elem.find('cim:ACLineSegment.r', CIM_NS).text)
         x = float(line_elem.find('cim:ACLineSegment.x', CIM_NS).text)
 
+        # Map to catalog pattern
+        imp_id = f"imp_{line_id}"
+        impedance_catalog[imp_id] = PerLengthPhaseImpedance(id=imp_id, r_ohm_per_km=r, x_ohm_per_km=x)
+
         segments.append(ACLineSegment(
             id=line_id,
-            from_node=from_node,
-            to_node=to_node,
             length_km=length_km,
-            r_ohm_per_km=r,
-            x_ohm_per_km=x
+            impedance_ref=imp_id
         ))
 
+        # Build linkages via Terminals
+        terminals.append(Terminal(id=f"t_{line_id}_1", connectivity_node=from_node, conducting_equipment=line_id))
+        terminals.append(Terminal(id=f"t_{line_id}_2", connectivity_node=to_node, conducting_equipment=line_id))
+
     # 3. Parse EnergyConsumers
     for load_elem in root.findall('cim:EnergyConsumer', CIM_NS):
         load_id = load_elem.attrib.get(f"{{{CIM_NS['rdf']}}}ID")
 
         node_ref = load_elem.find('cim:EnergyConsumer.node', CIM_NS).attrib.get(
-            f"{CIM_NS['rdf']}  resource").strip('#')
+            f"{{{CIM_NS['rdf']}}}resource").strip('#')
         p = float(load_elem.find('cim:EnergyConsumer.p', CIM_NS).text)
         q = float(load_elem.find('cim:EnergyConsumer.q', CIM_NS).text)
 
         consumers.append(EnergyConsumer(
             id=load_id,
-            node=node_ref,
             p_mw=p,
             q_mvar=q
         ))
 
-    print(
-        f"[+] Loaded {
-            len(nodes)} Nodes, {
-            len(segments)} Lines, {
-                len(consumers)} Consumers.")
-    return GridTopology(nodes=nodes, segments=segments, consumers=consumers)
+        # Terminal for consumer
+        terminals.append(Terminal(id=f"t_{load_id}_1", connectivity_node=node_ref, conducting_equipment=load_id))
+
+    print(f"[+] Loaded {len(nodes)} Nodes, {len(terminals)} Terminals, {len(segments)} Lines, {len(consumers)} Consumers.")
+    return GridTopology(nodes=nodes, terminals=terminals, segments=segments, consumers=consumers, impedance_catalog=impedance_catalog)
 
 
 if __name__ == "__main__":

main.py

@@ -35,77 +35,121 @@ class ConnectivityNode(BaseModel):
     type: Literal['PQ', 'PV', 'SLACK'] = 'PQ'
 
 
-class ACLineSegment(BaseModel):
+class Terminal(BaseModel):
+    id: str
+    connectivity_node: str
+    conducting_equipment: str
+
+
+class PerLengthPhaseImpedance(BaseModel):
     id: str
-    from_node: str
-    to_node: str
-    length_km: float = Field(..., gt=0)
     r_ohm_per_km: float = Field(..., ge=0)
     x_ohm_per_km: float = Field(..., ge=0)
 
-    @property
-    def z_total(self) -> complex:
-        return complex(self.r_ohm_per_km, self.x_ohm_per_km) * self.length_km
-
 
-class EnergyConsumer(BaseModel):
+class ConductingEquipment(BaseModel):
     id: str
-    node: str
+
+
+class ACLineSegment(ConductingEquipment):
+    length_km: float = Field(..., gt=0)
+    impedance_ref: str
+
+    def get_z_total(self, catalog: Dict[str,
+                    PerLengthPhaseImpedance]) -> complex:
+        imp = catalog[self.impedance_ref]
+        return complex(imp.r_ohm_per_km, imp.x_ohm_per_km) * self.length_km
+
+
+class EnergyConsumer(ConductingEquipment):
     p_mw: float
     q_mvar: float
 
 
 class GridTopology(BaseModel):
     nodes: List[ConnectivityNode]
+    terminals: List[Terminal]
     segments: List[ACLineSegment]
     consumers: List[EnergyConsumer] = []
+    impedance_catalog: Dict[str, PerLengthPhaseImpedance] = {}
 
     @model_validator(mode='after')
     def validate_connectivity(self):
         node_ids = {n.id for n in self.nodes}
-        for segment in self.segments:
-            if segment.from_node not in node_ids or segment.to_node not in node_ids:
-                raise ValueError(
-                    f"[!] Orphaned segment detected: {
-                        segment.id}")
+        equip_ids = {e.id for e in self.segments + self.consumers}
+
+        for t in self.terminals:
+            if t.connectivity_node not in node_ids:
+                raise ValueError(f"[!] Invalid Node Ref on Terminal: {t.id}")
+            if t.conducting_equipment not in equip_ids:
+                raise ValueError(f"[!] Invalid Equipment Ref on Terminal: {t.id}")
+
         return self
 
 # ==========================================
 # 2. TOPOLOGY GENERATOR (Realistic Params)
 # ==========================================
 
 
-def generate_linear_feeder(n_nodes: int,
-                           voltage_kv: float = 20.0) -> Tuple[List[ConnectivityNode],
-                                                              List[ACLineSegment]]:
+def generate_linear_feeder(
+        n_nodes: int,
+        voltage_kv: float = 20.0) -> GridTopology:
     """
     Generates a realistic suburban feeder.
     Spacing: 0.6 km (Standard node/substation spacing)
     Cable: NA2XS2Y 1x150 mm² (R=0.20, X=0.12)
     """
+
+    # Standard Impedance Catalog
+    na2xs2y_150 = PerLengthPhaseImpedance(
+        id="cable_na2xs2y_1x150", r_ohm_per_km=0.20, x_ohm_per_km=0.12
+    )
+    catalog = {na2xs2y_150.id: na2xs2y_150}
+
     nodes = [
         ConnectivityNode(
             id="source",
             voltage_level_kv=voltage_kv,
             type='SLACK')]
     segments = []
+    terminals = []
 
-    prev_id = "source"
+    prev_node_id = "source"
     for i in range(1, n_nodes + 1):
-        curr_id = f"node_{i}"
-        nodes.append(ConnectivityNode(id=curr_id, voltage_level_kv=voltage_kv))
+        curr_node_id = f"node_{i}"
+        line_id = f"segment_{i}"
 
+        nodes.append(
+            ConnectivityNode(
+                id=curr_node_id,
+                voltage_level_kv=voltage_kv))
+
+        # The Equipment
         segments.append(ACLineSegment(
-            id=f"segment_{i}",
-            from_node=prev_id,
-            to_node=curr_id,
-            length_km=0.6,      # <--- UPDATED: Realistic 600m spacing
-            r_ohm_per_km=0.20,  # <--- UPDATED: Realistic Aluminum Cable Res
-            x_ohm_per_km=0.12
+            id=line_id,
+            length_km=0.6,
+            impedance_ref=na2xs2y_150.id
         ))
-        prev_id = curr_id
 
-    return nodes, segments
+        # The Linkage via Terminals
+        terminals.append(
+            Terminal(
+                id=f"t_{line_id}_1",
+                connectivity_node=prev_node_id,
+                conducting_equipment=line_id))
+        terminals.append(
+            Terminal(
+                id=f"t_{line_id}_2",
+                connectivity_node=curr_node_id,
+                conducting_equipment=line_id))
+
+        prev_node_id = curr_node_id
+
+    return GridTopology(
+        nodes=nodes,
+        terminals=terminals,
+        segments=segments,
+        impedance_catalog=catalog)
 
 # ==========================================
 # 3. HIGH-PERFORMANCE PHYSICS KERNEL
@@ -174,10 +218,21 @@ def _compile_topology(self):
 
         n = len(self.grid.nodes)
 
-        # 2. Graph Analysis
+        # 2. Graph Analysis (Bridging via Terminals)
         G = nx.Graph()
+
+        # Build equipment-to-nodes map
+        equip_nodes_map = {}
+        for t in self.grid.terminals:
+            if t.conducting_equipment not in equip_nodes_map:
+                equip_nodes_map[t.conducting_equipment] = []
+            equip_nodes_map[t.conducting_equipment].append(t.connectivity_node)
+
         for segment in self.grid.segments:
-            G.add_edge(segment.from_node, segment.to_node, z=segment.z_total)
+            nodes = equip_nodes_map.get(segment.id, [])
+            if len(nodes) == 2:
+                z = segment.get_z_total(self.grid.impedance_catalog)
+                G.add_edge(nodes[0], nodes[1], z=z)
 
         root_id = self.grid.nodes[0].id
         bfs_tree = nx.bfs_tree(G, source=root_id)
@@ -199,8 +254,15 @@ def solve(
         q_inj = np.zeros(n)
 
         for consumer in active_consumers:
-            if consumer.node in self.node_map:
-                idx = self.node_map[consumer.node]
+            # Find the terminal for this consumer
+            consumer_node = None
+            for t in self.grid.terminals:
+                if t.conducting_equipment == consumer.id:
+                    consumer_node = t.connectivity_node
+                    break
+
+            if consumer_node and consumer_node in self.node_map:
+                idx = self.node_map[consumer_node]
                 p_inj[idx] += consumer.p_mw
                 q_inj[idx] += consumer.q_mvar
 
@@ -244,23 +306,25 @@ def compute_batch_q(self, v_pu_array: np.ndarray) -> np.ndarray:
     # A. SETUP (Realistic Suburban Feeder)
     #
     N_NODES = 20
-    nodes, segments = generate_linear_feeder(N_NODES, voltage_kv=20.0)
+    grid = generate_linear_feeder(N_NODES, voltage_kv=20.0)
 
     # 2.5 MW is a realistic heavy load (e.g., small factory or EV park)
-    end_node_id = nodes[-1].id
-    consumers = [
-        EnergyConsumer(
-            id="heavy_load",
-            node=end_node_id,
-            p_mw=2.5,
-            q_mvar=0.5)]
+    end_node_id = grid.nodes[-1].id
+    heavy_load = EnergyConsumer(id="heavy_load", p_mw=2.5, q_mvar=0.5)
+
+    # Link load to node via Terminal
+    load_terminal = Terminal(
+        id="t_heavy_load_1",
+        connectivity_node=end_node_id,
+        conducting_equipment=heavy_load.id)
 
-    grid = GridTopology(nodes=nodes, segments=segments, consumers=consumers)
+    grid.consumers.append(heavy_load)
+    grid.terminals.append(load_terminal)
 
     # B. SOLVE
     t0 = time.perf_counter()
     engine = FastGridEngine(grid)
-    res = engine.solve(consumers)
+    res = engine.solve(grid.consumers)
     solve_ms = (time.perf_counter() - t0) * 1000
 
     # C. ANALYZE
@@ -276,8 +340,8 @@ def compute_batch_q(self, v_pu_array: np.ndarray) -> np.ndarray:
 
     # D. VISUALIZE
     #
-    dist_km = [i * 0.6 for i in range(len(nodes))]  # 0.6km segments
-    v_profile = [abs(res[n.id]) / 20.0 for n in nodes]
+    dist_km = [i * 0.6 for i in range(len(grid.nodes))]  # 0.6km segments
+    v_profile = [abs(res[n.id]) / 20.0 for n in grid.nodes]
 
     plt.figure(figsize=(10, 5))
     plt.plot(dist_km, v_profile, 'o-', color='navy', label='Voltage Profile')