Updated context manager to read SETO topo areas

gridappsd-field-bus-lib/gridappsd/field_interface/agents/__init__.py

@@ -2,6 +2,6 @@
 
 from gridappsd.field_interface.agents.agents import (FeederAgent, DistributedAgent,
                                                      CoordinatingAgent, SwitchAreaAgent,
-                                                     SecondaryAreaAgent)
+                                                     SecondaryAreaAgent, SubstationAgent)
 
 __all__: List[str] = ["FeederAgent", "DistributedAgent", "CoordinatingAgent"]

gridappsd-field-bus-lib/gridappsd/field_interface/agents/agents.py

@@ -233,7 +233,28 @@ def on_control(self, control):
         self.control_device(device_id, command)
 '''
 
+class SubstationAgent(DistributedAgent):
 
+    def __init__(self,
+                 upstream_message_bus_def: MessageBusDefinition,
+                 downstream_message_bus_def: MessageBusDefinition,
+                 agent_config: Dict,
+                 substation_dict=None,
+                 simulation_id=None):
+        super().__init__(upstream_message_bus_def, downstream_message_bus_def, agent_config,
+                         substation_dict, simulation_id)
+        self.substation_area = None
+        self.downstream_message_bus_def = downstream_message_bus_def
+
+        self._connect()
+
+        if self.agent_area_dict is not None:
+            substation = cim.EquipmentContainer(mRID=self.downstream_message_bus_def.id)
+            self.substation_area = DistributedArea(connection=self.connection,
+                                               container=substation,
+                                               distributed=True)
+            self.substation_area.build_from_topo_message(topology_dict=self.agent_area_dict,
+                                                     centralized_graph=None)
 class FeederAgent(DistributedAgent):
 
     def __init__(self,

gridappsd-field-bus-lib/gridappsd/field_interface/context_managers/__init__.py

@@ -0,0 +1,9 @@
+from typing import List
+
+from gridappsd.field_interface.context_managers.context_manager_agents import (SubstationAreaContextManager,
+                                                                FeederAreaContextManager,
+                                                                SwitchAreaContextManager,
+                                                                SecondaryAreaContextManager)
+
+
+__all__: List[str] = ["SubstationAreaContextManager","FeederAreaContextManager","SwitchAreaContextManager","SecondaryAreaContextManager"]

gridappsd-field-bus-lib/gridappsd/field_interface/context_managers/centralized_context_managers.py

@@ -0,0 +1,93 @@
+import argparse
+import logging
+import json
+import os
+import time
+from typing import Dict
+
+
+from cimgraph.data_profile import CIM_PROFILE
+from gridappsd import GridAPPSD
+import gridappsd.topics as t
+import gridappsd.field_interface.agents.agents as agents_mod
+from gridappsd.field_interface.context_managers.utils import REQUEST_FIELD, get_MessageBusDefinition
+from gridappsd.field_interface.context_managers.context_manager_agents import FeederAreaContextManager, SwitchAreaContextManager, SecondaryAreaContextManager
+
+cim_profile = CIM_PROFILE.RC4_2021.value
+agents_mod.set_cim_profile(cim_profile=cim_profile, iec61970_301=7)
+cim = agents_mod.cim
+
+logging.basicConfig(level=logging.DEBUG)
+logging.getLogger('goss').setLevel(logging.ERROR)
+logging.getLogger('stomp.py').setLevel(logging.ERROR)
+
+_log = logging.getLogger(__name__)
+
+def _main():
+
+    time.sleep(10)
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--simulation_id",
+        help="Simulation id to use for communicating with simulated devices on the message bus. \
+                        If simulation_id is not provided then Context Manager assumes to run on deployed field with real devices.",
+        required=False)
+    opts = parser.parse_args()
+    simulation_id = opts.simulation_id
+
+    agent_config = {
+        "app_id":
+        "context_manager",
+        "description":
+        "This agent provides topological context information like neighboring agents and devices to other distributed agents"
+    }
+
+    gapps = GridAPPSD()
+    response = gapps.get_response(t.PLATFORM_STATUS, {"isField": True})
+    field_model_mrid = response['fieldModelMrid']
+
+    is_field_initialized = False
+
+    while not is_field_initialized:
+        response = gapps.get_response(REQUEST_FIELD, {"request_type": "is_initilized"})
+        print(response)
+        is_field_initialized = response['data']['initialized']
+        time.sleep(1)
+
+    system_message_bus_def = get_MessageBusDefinition(field_model_mrid)
+    feeder_message_bus_def = get_MessageBusDefinition(field_model_mrid)
+
+    #TODO: create access control for agents for different layers
+    feeder_agent = FeederAreaContextManager(system_message_bus_def,
+                                            feeder_message_bus_def,
+                                            agent_config,
+                                            simulation_id=simulation_id)
+
+    for switch_area in feeder_agent.agent_area_dict['switch_areas']:
+        switch_area_message_bus_def = get_MessageBusDefinition(str(switch_area['message_bus_id']))
+        print("Creating switch area agent " + str(switch_area['message_bus_id']))
+        switch_area_agent = SwitchAreaContextManager(feeder_message_bus_def,
+                                                     switch_area_message_bus_def,
+                                                     agent_config,
+                                                     simulation_id=simulation_id)
+
+        # create secondary area distributed agents
+        for secondary_area in switch_area['secondary_areas']:
+            secondary_area_message_bus_def = get_MessageBusDefinition(
+                str(secondary_area['message_bus_id']))
+            print("Creating secondary area agent " + str(secondary_area['message_bus_id']))
+            secondary_area_agent = SecondaryAreaContextManager(switch_area_message_bus_def,
+                                                               secondary_area_message_bus_def,
+                                                               agent_config,
+                                                               simulation_id=simulation_id)
+
+    while True:
+        try:
+            time.sleep(0.1)
+        except KeyboardInterrupt:
+            print("Exiting sample")
+            break
+
+
+if __name__ == "__main__":
+    _main()

gridappsd-field-bus-lib/gridappsd/field_interface/context_manager.py → gridappsd-field-bus-lib/gridappsd/field_interface/context_managers/context_manager_agents.py

@@ -1,30 +1,55 @@
-import argparse
 import logging
-import os
 import time
 from typing import Dict
 
-import gridappsd.field_interface.agents.agents as agents_mod
 import gridappsd.topics as t
-from cimgraph.data_profile import CIM_PROFILE
 from gridappsd import GridAPPSD
-from gridappsd.field_interface.agents import (FeederAgent, SecondaryAreaAgent, SwitchAreaAgent)
+from gridappsd.field_interface.agents import (SubstationAgent, FeederAgent, SecondaryAreaAgent, SwitchAreaAgent)
 from gridappsd.field_interface.interfaces import MessageBusDefinition
-
-cim_profile = CIM_PROFILE.RC4_2021.value
-
-agents_mod.set_cim_profile(cim_profile=cim_profile, iec61970_301=7)
-
-cim = agents_mod.cim
+from gridappsd.field_interface.context_managers.utils import REQUEST_FIELD
 
 logging.basicConfig(level=logging.DEBUG)
 logging.getLogger('goss').setLevel(logging.ERROR)
 logging.getLogger('stomp.py').setLevel(logging.ERROR)
 
 _log = logging.getLogger(__name__)
 
-#FieldBusManager's request topics. To be used only by context manager user role only.
-REQUEST_FIELD = ".".join((t.PROCESS_PREFIX, "request.field"))
+class SubstationAreaContextManager(SubstationAgent):
+
+    def __init__(self,
+                 upstream_message_bus_def: MessageBusDefinition,
+                 downstream_message_bus_def: MessageBusDefinition,
+                 agent_config: Dict,
+                 substation_dict: Dict = None,
+                 simulation_id: str = None):
+
+        self.ot_connection = GridAPPSD()
+        if substation_dict is None:
+            request = {'request_type': 'get_context', 'modelId': downstream_message_bus_def.id}
+            substation_dict = None
+            while substation_dict is None:
+                self.ot_connection.get_logger().debug(f"Requesting topology for {self.__class__}")
+                response = self.ot_connection.get_response(REQUEST_FIELD, request, timeout=10)
+                if 'DistributionArea' in response:
+                    substation_dict = response['DistributionArea']['Substation']['@id']
+                    self.ot_connection.get_logger().debug("Topology received at Substation Area Context Manager")
+                else:
+                    time.sleep(5)
+        super().__init__(upstream_message_bus_def, downstream_message_bus_def, agent_config,
+                         substation_dict, simulation_id)
+
+        #Override agent_id to a static value
+        self.agent_id = downstream_message_bus_def.id + '.context_manager'
+
+        self.context = {'data':substation_dict}
+
+        self.registered_agents = {}
+        self.registered_agents[self.agent_id] = self.get_registration_details()
+
+        self.neighbouring_agents = {}
+        self.upstream_agents = {}
+        self.downstream_agents = {}
+        self.ot_connection.get_logger().info("Substation Area Context Manager Created")
 
 
 class FeederAreaContextManager(FeederAgent):
@@ -205,90 +230,3 @@ def on_request(self, message_bus, headers: Dict, message):
         elif message['request_type'] == 'control_command':
             simulation_id = message['input']['simulation_id']
             self.ot_connection.send(t.simulation_input_topic(simulation_id), message)
-
-
-def get_MessageBusDefinition(area_id: str) -> MessageBusDefinition:
-
-    connection_args = {
-        "GRIDAPPSD_ADDRESS": os.environ.get('GRIDAPPSD_ADDRESS', "tcp://gridappsd:61613"),
-        "GRIDAPPSD_USER": os.environ.get('GRIDAPPSD_USER'),
-        "GRIDAPPSD_PASSWORD": os.environ.get('GRIDAPPSD_PASSWORD'),
-        "GRIDAPPSD_APPLICATION_ID": os.environ.get('GRIDAPPSD_APPLICATION_ID')
-    }
-
-    bus = MessageBusDefinition(id=area_id,
-                               is_ot_bus=True,
-                               connection_type="GRIDAPPSD_TYPE_GRIDAPPSD",
-                               conneciton_args=connection_args)
-
-    return bus
-
-
-def _main():
-
-    time.sleep(10)
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "--simulation_id",
-        help="Simulation id to use for communicating with simulated devices on the message bus. \
-                        If simulation_id is not provided then Context Manager assumes to run on deployed field with real devices.",
-        required=False)
-    opts = parser.parse_args()
-    simulation_id = opts.simulation_id
-
-    agent_config = {
-        "app_id":
-        "context_manager",
-        "description":
-        "This agent provides topological context information like neighboring agents and devices to other distributed agents"
-    }
-
-    gapps = GridAPPSD()
-    response = gapps.get_response(t.PLATFORM_STATUS, {"isField": True})
-    field_model_mrid = response['fieldModelMrid']
-
-    is_field_initialized = False
-
-    while not is_field_initialized:
-        response = gapps.get_response(REQUEST_FIELD, {"request_type": "is_initilized"})
-        print(response)
-        is_field_initialized = response['data']['initialized']
-        time.sleep(1)
-
-    system_message_bus_def = get_MessageBusDefinition(field_model_mrid)
-    feeder_message_bus_def = get_MessageBusDefinition(field_model_mrid)
-
-    #TODO: create access control for agents for different layers
-    feeder_agent = FeederAreaContextManager(system_message_bus_def,
-                                            feeder_message_bus_def,
-                                            agent_config,
-                                            simulation_id=simulation_id)
-
-    for switch_area in feeder_agent.agent_area_dict['switch_areas']:
-        switch_area_message_bus_def = get_MessageBusDefinition(str(switch_area['message_bus_id']))
-        print("Creating switch area agent " + str(switch_area['message_bus_id']))
-        switch_area_agent = SwitchAreaContextManager(feeder_message_bus_def,
-                                                     switch_area_message_bus_def,
-                                                     agent_config,
-                                                     simulation_id=simulation_id)
-
-        # create secondary area distributed agents
-        for secondary_area in switch_area['secondary_areas']:
-            secondary_area_message_bus_def = get_MessageBusDefinition(
-                str(secondary_area['message_bus_id']))
-            print("Creating secondary area agent " + str(secondary_area['message_bus_id']))
-            secondary_area_agent = SecondaryAreaContextManager(switch_area_message_bus_def,
-                                                               secondary_area_message_bus_def,
-                                                               agent_config,
-                                                               simulation_id=simulation_id)
-
-    while True:
-        try:
-            time.sleep(0.1)
-        except KeyboardInterrupt:
-            print("Exiting sample")
-            break
-
-
-if __name__ == "__main__":
-    _main()

gridappsd-field-bus-lib/gridappsd/field_interface/context_managers/substation.py

@@ -0,0 +1,78 @@
+import argparse
+import json
+import logging
+import time
+
+from cimgraph.data_profile import CIM_PROFILE
+
+import gridappsd.field_interface.agents.agents as agents_mod
+from gridappsd.field_interface.interfaces import MessageBusDefinition
+from gridappsd.field_interface.context_managers.context_manager_agents import SubstationAreaContextManager
+
+
+cim_profile = CIM_PROFILE.RC4_2021.value
+agents_mod.set_cim_profile(cim_profile=cim_profile, iec61970_301=7)
+cim = agents_mod.cim
+
+logging.basicConfig(level=logging.DEBUG)
+logging.getLogger('goss').setLevel(logging.ERROR)
+logging.getLogger('stomp.py').setLevel(logging.ERROR)
+
+_log = logging.getLogger(__name__)
+
+def _main():
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--simulation_id",
+        help="Simulation id to use for communicating with simulated devices on the message bus. \
+                        If simulation_id is not provided then Context Manager assumes to run on deployed field with real devices.",
+        required=False)
+    parser.add_argument(
+        "--system_message_bus",
+        help="Yaml file to connect with upstream system(OT) message bus.",
+        required=True)
+
+    parser.add_argument(
+        "--substation_message_bus",
+        help="Yaml file to connect with downstream substation area message bus.",
+        required=True)
+
+    parser.add_argument(
+        "--substation_dict",
+        help="JSON file containing substation topology dictionary. If this file is not provided then disctionary is requested by Field Bus Manager using upstream message bus.",
+        required=False)
+
+    opts = parser.parse_args()
+    simulation_id = opts.simulation_id
+
+    agent_config = {
+        "app_id":
+        "context_manager",
+        "description":
+        "This agent provides topological context information like neighboring agents and devices to other distributed agents"
+    }
+
+
+    system_message_bus_def = MessageBusDefinition.load(opts.system_message_bus)   
+    substation_message_bus_def = MessageBusDefinition.load(opts.substation_message_bus)
+
+    with open(opts.substation_dict,encoding="utf-8") as f:
+        substation_dict = json.load(f)
+
+    substation_agent = SubstationAreaContextManager(system_message_bus_def,
+                                            substation_message_bus_def,
+                                            agent_config,
+                                            substation_dict = substation_dict,
+                                            simulation_id=simulation_id)
+
+    while True:
+        try:
+            time.sleep(0.1)
+        except KeyboardInterrupt:
+            print("Exiting sample")
+            break
+
+
+if __name__ == "__main__":
+    _main()