partial stubs for Datasources and a small portion of Time and Styling

Author: tipatterson-dev

oshconnect/datasource/__init__.py

@@ -1,37 +1,14 @@
-def isDefined(v):
+from enum import Enum
+
+
+def is_defined(v):
     return v is not None
 
-def randomUUID():
-    return str(v)
 
-class Mode(enum):
+class Mode(Enum):
     REPLAY = "replay"
     BATCH = "batch"
     REAL_TIME = "realTime"
 
-DATASOURCE_DATA_TOPIC = 'datasource-data-'
-batchsize = 1
-
-
-class DataSourceWorker:
-    def __init__(self,dataSourceHandlers):
-        self.dataSourceHandlers = dataSourceHandlers
-
 class DataSourceHandler:
-    def __init__(self,context,topic,broadcastChannel,values,version,properties,initialized):
-        self.context = any
-        self.topic = any
-        self.broadcastChannel = any
-        self.values = []
-        self.values = 0
-        self.properties = batchsize
-        self.initialized = False
-
-
-def handleIsInit(self,eventData, resp):
-    dsId = eventData.dsId
-    resp.data = self.dataSourceHandlers(dsId).isInitalized()
-    self.postMessage(resp)
-
-
-        
+    datasources

oshconnect/datasource/datasource.py

@@ -0,0 +1,79 @@
+from uuid import uuid4
+
+from oshconnect.datasource import Mode
+
+
+class DataSource:
+
+    def __init__(self, name: str, mode: Mode, properties: dict):
+        self._id = f'datasource-{uuid4()}'
+        self.name = name
+        self.mode = mode
+        self.properties = properties
+
+    def get_id(self) -> str:
+        return self._id
+
+    def get_name(self):
+        pass
+
+    def create_process(self):
+        pass
+
+    def terminate_process(self):
+        pass
+
+    # Might not be necessary
+    def subscribe(self):
+        pass
+
+    def update_properties(self, properties: dict):
+        # TODO: need to stop in progress sub-processes and restart
+        self.properties = properties
+
+    def initialize(self):
+        pass
+
+    def connect(self):
+        pass
+
+    def disconnect(self):
+        pass
+
+    def reset(self):
+        pass
+
+    def get_status(self):
+        pass
+
+
+class DatasourceHandler:
+    datasource_map: dict[str, DataSource]
+
+    def __init__(self):
+        self.datasource_map = {}
+
+    def add_datasource(self, datasource: DataSource):
+        self.datasource_map[datasource.get_id()] = datasource
+
+    def initialize_ds(self, datasource_id: str, properties: dict):
+        ds = self.datasource_map.get(datasource_id)
+        ds.initialize()
+
+    def initialize_all(self):
+        # list comp is faster than for loop
+        [ds.initialize() for ds in self.datasource_map.values()]
+
+    def connect_ds(self, datasource_id: str):
+        ds = self.datasource_map.get(datasource_id)
+        ds.connect()
+
+    def connect_all(self):
+        [ds.connect() for ds in self.datasource_map.values()]
+
+    def disconnect_ds(self, datasource_id: str):
+        ds = self.datasource_map.get(datasource_id)
+        ds.disconnect()
+
+    def disconnect_all(self):
+        [ds.disconnect() for ds in self.datasource_map.values()]

oshconnect/datasource/handler.py

@@ -0,0 +1,59 @@
+from abc import ABC, abstractmethod
+
+
+# Might not be necessary due to differences in this implementation die to actual multiprocessing
+class DataSourceHandler(ABC):
+    def __init__(self):
+        self.context = None
+        self.topic = None
+        self.broadcast_channel = None
+        self.values = []
+        self.version = 0
+        self.properties = {
+            "batchsize": 1
+        }
+        self._initialized = False
+        self.datasource_id = None
+
+    @abstractmethod
+    def create_context(self, properties: dict):
+        pass
+
+    async def init(self, datasource_id: str, properties: dict, topics: list[str]):
+        self.datasource_id = datasource_id
+        self.properties.update(properties)
+        self.topic = self.set_topics(topics)
+        # Context doesn't really have to exist in python
+        self.context = self.create_context(properties)
+        self.context.on_change_status = self.on_change_status()
+        self.context.handle_data = self.handle_data()
+        await self.context.init(self.properties)
+        self._initialized = True
+
+    # TODO: topics may not be necessary
+    def set_topics(self, topics):
+        _topic = topics.data
+        if self.topic == _topic:
+            return
+        return
+
+    def on_change_status(self):
+        pass
+
+    def handle_data(self):
+        pass
+
+    def flush(self):
+        pass
+
+    def connect(self):
+        pass
+
+    def disconnect(self):
+        pass
+
+    def is_initialized(self):
+        return self._initialized
+
+    def is_connected(self):
+        pass

oshconnect/styling/__init__.py

@@ -0,0 +1,25 @@
+from abc import ABC
+from typing import Callable
+
+from pydantic import BaseModel
+
+
+class VisualizationDataLayer(BaseModel):
+    """
+    Represents the data portion of a particular visualization as well as its state information.
+    """
+
+    name: str
+    description: str
+    datasource_id: list[str]
+    visible: bool
+    timestamp: float
+    _get_timestamp: Callable
+    on_left_click: Callable
+    on_right_click: Callable
+    on_hover: Callable
+    _id: str
+    _type: str
+    _filter: dict
+    _datasources_to_fn: Callable
+    _no_datasources_fn: Callable

oshconnect/timemanagement/timemanagement.py

@@ -107,3 +107,10 @@ def get_buffer_time(self):
 
     def _compute_time_range(self):
         pass
+
+
+class Synchronizer:
+    _buffer: any
+
+    def synchronize(self, systems: list):
+        pass