update discovery methods to more reliably add subresources to their parents for serialization

Author: tipatterson-dev

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oshconnect"
-version = "0.3.0a5.post1"
+version = "0.3.0a6"
 description = "Library for interfacing with OSH, helping guide visualization efforts, and providing a place to store configurations."
 readme = "README.md"
 authors = [

src/oshconnect/oshconnectapi.py

@@ -5,7 +5,7 @@
 #  Contact email:  ian@botts-inc.com
 #   ==============================================================================
 import logging
-import shelve
+import json
 from uuid import UUID
 
 from .csapi4py.default_api_helpers import APIHelper
@@ -26,7 +26,7 @@ class OSHConnect:
     _cs_api_builder: APIHelper = None
     # _datasource_handler: DataStreamHandler = None
     _datastreams: list[Datastream] = []
-    _datataskers: list[DataStore] = []
+    _controlstreams: list[ControlStream] = []
     _datagroups: list = []
     _tasks: list = []
     _playback_mode: TemporalModes = TemporalModes.REAL_TIME
@@ -68,18 +68,27 @@ def remove_node(self, node_id: str):
         self._nodes = [node for node in self._nodes if
                        node.get_id() != node_id]
 
-    def save_config(self, config: dict):
+    def save_config(self):
         logging.info(f"Saving configuration for {self._name}")
-        with shelve.open(f"{self._name}_config") as db:
-            db['app_config'] = self
-            db.close()
+
+        data = {}
+        for node in self._nodes:
+            node_dict = node.serialize()
+            data.update({node.get_id(): node_dict})
+
+        # write to JSON file
+        file_path = f"{self._name}_config.json"
+        with open(file_path, 'w', encoding='utf-8') as f:
+            json.dump({"app_config": data}, f, ensure_ascii=False, indent=2)
 
     @classmethod
     def load_config(cls, file_name: str) -> 'OSHConnect':
-        with shelve.open(file_name, 'r') as db:
-            app = db['app_config']
-            db.close()
-            return app
+        """Load configuration data from a JSON file and return the stored config dict.
+        Note: Despite the return type hint, this returns the configuration dictionary.
+        """
+        with open(file_name, 'r', encoding='utf-8') as f:
+            obj = json.load(f)
+            return obj.get('app_config', obj)
 
     def share_config(self, config: dict):
         pass
@@ -114,18 +123,6 @@ def visualize_streams(self, streams: list):
     def get_visualization_recommendations(self, streams: list):
         pass
 
-    def discover_datastreams(self):
-        for system in self._systems:
-            res_datastreams = system.discover_datastreams()
-            datastreams = list(
-                map(lambda ds: Datastream(parent_node=system.get_parent_node(), id=ds.ds_id, datastream_resource=ds),
-                    res_datastreams))
-
-            for ds in datastreams:
-                ds.set_parent_resource_id(system.get_underlying_resource().system_id)
-            # datastreams = [ds.set_parent_resource_id(system.get_underlying_resource().system_id) for ds in datastreams]
-            self._datastreams.extend(datastreams)
-
     def discover_systems(self, nodes: list[str] = None):
         """
         Discover systems from the nodes that have been added to the OSHConnect instance. They are associated with the
@@ -142,15 +139,16 @@ def discover_systems(self, nodes: list[str] = None):
             res_systems = node.discover_systems()
             self._systems.extend(res_systems)
 
+    def discover_datastreams(self):
+        for system in self._systems:
+            datastreams = system.discover_datastreams()
+            self._datastreams.extend(datastreams)
+
     def discover_controlstreams(self, streams: list):
         for system in self._systems:
-            res_controlstreams = system.discover_controlstreams()
-            controlstreams = list(
-                map(lambda cs: ControlStream(parent_node=system.get_parent_node(), id=cs.cs_id,
-                                             controlstream_resource=cs), res_controlstreams))
-            for cs in controlstreams:
-                cs.set_parent_resource_id(system.get_underlying_resource().system_id)
-            self._datataskers.extend(controlstreams)
+            controlstreams = system.discover_controlstreams()
+
+            self._controlstreams.extend(controlstreams)
 
     def authenticate_user(self, user: dict):
         pass

src/oshconnect/streamableresource.py

@@ -194,10 +194,10 @@ def discover_systems(self):
             print(system_objs)
             for system_json in system_objs:
                 print(system_json)
-                system = SystemResource.model_validate(system_json)
+                system = SystemResource.model_validate(system_json, by_alias=True)
                 sys_obj = System(label=system.properties['name'],
                                  name=to_lower_camel(system.properties['name'].replace(" ", "_")),
-                                 urn=system.properties['uid'], parent_node=self)
+                                 urn=system.properties['uid'], parent_node=self, resource_id=system.system_id)
 
                 self._systems.append(sys_obj)
                 new_systems.append(sys_obj)
@@ -247,6 +247,64 @@ def register_streamable(self, streamable: StreamableResource):
     def get_session(self) -> OSHClientSession:
         return self._client_session
 
+    def serialize(self) -> dict:
+        data = {
+            "_id": self._id,
+            "protocol": self.protocol,
+            "address": self.address,
+            "port": self.port,
+            "server_root": self.server_root,
+            "is_secure": self.is_secure,
+            "username": getattr(self._api_helper, "username", None),
+            "password": getattr(self._api_helper, "password", None),
+            "_systems": [system.serialize() for system in self._systems] if self._systems is not None else None,
+        }
+        data["name"] = getattr(self, "name", None)
+        data["label"] = getattr(self, "label", None)
+        data["urn"] = getattr(self, "urn", None)
+        data["description"] = getattr(self, "description", None)
+        datastreams = getattr(self, "datastreams", None)
+        if datastreams is not None:
+            data["datastreams"] = [ds.serialize() for ds in datastreams]
+        else:
+            data["datastreams"] = None
+        control_channels = getattr(self, "control_channels", None)
+        if control_channels is not None:
+            data["control_channels"] = [cc.serialize() for cc in control_channels]
+        else:
+            data["control_channels"] = None
+        underlying = getattr(self, "_underlying_resource", None)
+        if underlying is not None:
+            dump = getattr(underlying, 'model_dump', None)
+            if callable(dump):
+                data["underlying_resource"] = underlying.model_dump(by_alias=True, exclude_none=True)
+            elif hasattr(underlying, 'to_dict'):
+                data["underlying_resource"] = underlying.to_dict()
+            else:
+                data["underlying_resource"] = str(underlying)
+        else:
+            data["underlying_resource"] = None
+        # Remove any 'resource' key if present
+        data.pop("resource", None)
+        return data
+
+    @classmethod
+    def deserialize(cls, data: dict, session_manager: 'SessionManager' = None) -> 'Node':
+        node = cls(
+            protocol=data["protocol"],
+            address=data["address"],
+            port=data["port"],
+            username=data.get("username"),
+            password=data.get("password"),
+            server_root=data.get("server_root", "sensorhub"),
+            session_manager=session_manager
+        )
+        node._id = data["_id"]
+        node.is_secure = data.get("is_secure", False)
+        node._systems = [System.deserialize(sys, node) for sys in data.get("_systems", [])] if data.get(
+            "_systems") is not None else []
+        return node
+
 
 class Status(Enum):
     INITIALIZING = "initializing"
@@ -269,7 +327,7 @@ class StreamableModes(Enum):
 class StreamableResource(Generic[T], ABC):
     _id: UUID
     _resource_id: str
-    _canonical_link: str
+    # _canonical_link: str
     _topic: str
     _status: str = Status.STOPPED.value
     ws_url: str
@@ -536,6 +594,39 @@ def get_inbound_deque(self):
     def get_outbound_deque(self):
         return self._outbound_deque
 
+    def serialize(self) -> dict:
+        """Serializes common attributes of StreamableResource, safely handling missing/None attributes."""
+        topic = getattr(self, "_topic", None)
+        status = getattr(self, "_status", None)
+        parent_resource_id = getattr(self, "_parent_resource_id", None)
+        connection_mode = getattr(self, "_connection_mode", None)
+        resource_id = getattr(self, "_resource_id", None)
+        if isinstance(connection_mode, Enum):
+            connection_mode = connection_mode.value
+
+        return {
+            "id": str(getattr(self, "_id", None)),
+            "resource_id": resource_id,
+            # "canonical_link": getattr(self, "_canonical_link", None),
+            "topic": topic,
+            "status": status,
+            "parent_resource_id": parent_resource_id,
+            "connection_mode": connection_mode,
+        }
+
+    @classmethod
+    def deserialize(cls, data: dict, node: 'Node') -> 'StreamableResource':
+        """Deserializes common attributes. Subclasses should override and call super()."""
+        obj = cls(node=node)
+        obj._id = uuid.UUID(data["id"])
+        obj._resource_id = data.get("resource_id")
+        # obj._canonical_link = data.get("canonical_link")
+        obj._topic = data.get("topic")
+        obj._status = data.get("status")
+        obj._parent_resource_id = data.get("parent_resource_id")
+        obj._connection_mode = StreamableModes(data.get("connection_mode", StreamableModes.PUSH.value)),
+        return obj
+
 
 class System(StreamableResource[SystemResource]):
     name: str
@@ -567,29 +658,37 @@ def __init__(self, name: str, label: str, urn: str, parent_node: Node, **kwargs)
 
         self._underlying_resource = self.to_system_resource()
 
-    def discover_datastreams(self) -> list[DatastreamResource]:
+    def discover_datastreams(self) -> list[Datastream]:
         res = self._parent_node.get_api_helper().get_resource(APIResourceTypes.SYSTEM, self._resource_id,
                                                               APIResourceTypes.DATASTREAM)
         datastream_json = res.json()['items']
-        ds_resources = []
+        datastreams = []
 
         for ds in datastream_json:
-            datastream_objs = DatastreamResource.model_validate(ds)
-            ds_resources.append(datastream_objs)
+            datastream_objs = DatastreamResource.model_validate(ds, by_alias=True)
+            new_ds = Datastream(self._parent_node, datastream_objs)
+            datastreams.append(new_ds)
 
-        return ds_resources
+            if not [ds.get_underlying_resource() != datastream_objs for ds in self.datastreams]:
+                self.datastreams.append(new_ds)
 
-    def discover_controlstreams(self) -> list[ControlStreamResource]:
+        return datastreams
+
+    def discover_controlstreams(self) -> list[ControlStream]:
         res = self._parent_node.get_api_helper().get_resource(APIResourceTypes.SYSTEM, self._resource_id,
                                                               APIResourceTypes.CONTROL_CHANNEL)
         controlstream_json = res.json()['items']
-        cs_resources = []
+        controlstreams = []
+
+        for cs_json in controlstream_json:
+            controlstream_objs = ControlStreamResource.model_validate(cs_json)
+            new_cs = ControlStream(self._parent_node, controlstream_objs)
+            controlstreams.append(new_cs)
 
-        for cs in controlstream_json:
-            controlstream_objs = ControlStreamResource.model_validate(cs)
-            cs_resources.append(controlstream_objs)
+            if not [cs.get_underlying_resource() != controlstream_objs for cs in self.control_channels]:
+                self.control_channels.append(new_cs)
 
-        return cs_resources
+        return controlstreams
 
     @staticmethod
     def from_system_resource(system_resource: SystemResource, parent_node: Node) -> System:
@@ -737,6 +836,53 @@ def retrieve_resource(self):
             self._underlying_resource = system_resource
             return None
 
+    def serialize(self) -> dict:
+        data = super().serialize()
+        data["name"] = getattr(self, "name", None)
+        data["label"] = getattr(self, "label", None)
+        data["urn"] = getattr(self, "urn", None)
+        data["description"] = getattr(self, "description", None)
+        datastreams = getattr(self, "datastreams", None)
+        if datastreams is not None:
+            data["datastreams"] = [ds.serialize() for ds in datastreams]
+        else:
+            data["datastreams"] = None
+        control_channels = getattr(self, "control_channels", None)
+        if control_channels is not None:
+            data["control_channels"] = [cc.serialize() for cc in control_channels]
+        else:
+            data["control_channels"] = None
+        underlying = getattr(self, "_underlying_resource", None)
+        if underlying is not None:
+            dump = getattr(underlying, 'model_dump', None)
+            if callable(dump):
+                data["underlying_resource"] = underlying.model_dump(by_alias=True, exclude_none=True)
+            elif hasattr(underlying, 'to_dict'):
+                data["underlying_resource"] = underlying.to_dict()
+            else:
+                data["underlying_resource"] = str(underlying)
+        else:
+            data["underlying_resource"] = None
+        # Remove any 'resource' key if present
+        data.pop("resource", None)
+        return data
+
+    @classmethod
+    def deserialize(cls, data: dict, node: 'Node') -> 'System':
+        obj = cls(
+            name=data["name"],
+            label=data["label"],
+            urn=data["urn"],
+            parent_node=node,
+            description=data.get("description"),
+            resource_id=data.get("resource_id")
+        )
+        obj._id = uuid.UUID(data["id"])
+        obj.datastreams = [Datastream.deserialize(ds, node) for ds in data.get("datastreams", [])]
+        obj.control_channels = [ControlStream.deserialize(cc, node) for cc in data.get("control_channels", [])]
+        obj._underlying_resource = SystemResource.model_validate(data.get("_underlying_resource"))
+        return obj
+
 
 class Datastream(StreamableResource[DatastreamResource]):
     should_poll: bool
@@ -813,6 +959,32 @@ def insert(self, data: dict):
         encoded = json.dumps(data).encode('utf-8')
         self._publish_mqtt(self._topic, encoded)
 
+    def serialize(self) -> dict:
+        data = super().serialize()
+        data["should_poll"] = getattr(self, "should_poll", None)
+        underlying = getattr(self, "_underlying_resource", None)
+        if underlying is not None:
+            dump = getattr(underlying, 'model_dump', None)
+            if callable(dump):
+                data["underlying_resource"] = underlying.model_dump(by_alias=True, exclude_none=True)
+            elif hasattr(underlying, 'to_dict'):
+                data["underlying_resource"] = underlying.to_dict()
+            else:
+                data["underlying_resource"] = str(underlying)
+        else:
+            data["underlying_resource"] = None
+
+        return data
+
+    @classmethod
+    def deserialize(cls, data: dict, node: 'Node') -> 'Datastream':
+        ds_resource = DatastreamResource.model_validate(data["resource"]) if data.get("resource") else None
+        obj = cls(parent_node=node, datastream_resource=ds_resource)
+        obj._id = uuid.UUID(data["id"])
+        obj.should_poll = data.get("should_poll", False)
+        obj._underlying_resource = DatastreamResource.model_validate(data["_underlying_resource"])
+        return obj
+
 
 class ControlStream(StreamableResource[ControlStreamResource]):
     _status_topic: str
@@ -905,3 +1077,29 @@ def subscribe(self, topic=None, callback=None, qos=0):
             self._mqtt_client.subscribe(t, qos=qos, msg_callback=self._mqtt_sub_callback)
         else:
             self._mqtt_client.subscribe(t, qos=qos, msg_callback=callback)
+
+    def serialize(self) -> dict:
+        data = super().serialize()
+        data["status_topic"] = getattr(self, "_status_topic", None)
+        underlying = getattr(self, "_underlying_resource", None)
+        if underlying is not None:
+            dump = getattr(underlying, 'model_dump', None)
+            if callable(dump):
+                data["underlying_resource"] = underlying.model_dump(by_alias=True, exclude_none=True)
+            elif hasattr(underlying, 'to_dict'):
+                data["underlying_resource"] = underlying.to_dict()
+            else:
+                data["underlying_resource"] = str(underlying)
+        else:
+            data["underlying_resource"] = None
+
+        return data
+
+    @classmethod
+    def deserialize(cls, data: dict, node: 'Node') -> 'ControlStream':
+        cs_resource = ControlStreamResource.model_validate(data["resource"]) if data.get("resource") else None
+        obj = cls(node=node, controlstream_resource=cs_resource)
+        obj._id = uuid.UUID(data["id"])
+        obj._status_topic = data.get("status_topic")
+        obj._underlying_resource = ControlStreamResource.model_validate(data["underlying_resource"])
+        return obj