Merge pull request #151 from simvue-io/feature/148-offline-online-subclasses

Use base class for Online and Offline

Author: kzscisoft

pyproject.toml

@@ -29,7 +29,7 @@ torch = ["torch"]
 plot = ["matplotlib", "plotly"]
 
 [tool.poetry.scripts]
-simvue_sender = "simvue.bin.simvue_sender:run"
+simvue_sender = "simvue.bin.sender:run"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0.0"

simvue/factory/__init__.py

@@ -0,0 +1,15 @@
+from .remote import Remote
+from .offline import Offline
+from .base import SimvueBaseClass
+
+
+def Simvue(
+    name: str,
+    uniq_id: str,
+    mode: str,
+    suppress_errors: bool = True
+) -> SimvueBaseClass:
+    if mode == "offline":
+        return Offline(name, uniq_id, suppress_errors)
+    else:
+        return Remote(name, uniq_id, suppress_errors)

simvue/factory/base.py

@@ -0,0 +1,68 @@
+import abc
+import uuid
+import typing
+import logging
+
+class SimvueBaseClass(abc.ABC):
+    @abc.abstractmethod
+    def __init__(self, name: str, uniq_id: uuid.UUID, suppress_errors: bool) -> None:
+        self._logger = logging.getLogger(f"simvue.{self.__class__.__name__}")
+        self._suppress_errors: bool = suppress_errors
+        self._uuid: str = uniq_id
+        self._name: str = name
+        self._id: typing.Optional[int] = None
+        self._aborted: bool = False
+
+    def _error(self, message: str) -> None:
+        """
+        Raise an exception if necessary and log error
+        """
+        if not self._suppress_errors:
+            raise RuntimeError(message)
+        else:
+            self._logger.error(message)
+            self._aborted = True
+
+    @abc.abstractmethod
+    def create_run(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def update(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def set_folder_details(self, data) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def save_file(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def add_alert(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def set_alert_state(self, alert_id: str, status: str) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def list_alerts(self) -> list[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def send_metrics(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def send_event(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def send_heartbeat(self) -> typing.Optional[dict[str, typing.Any]]:
+        pass
+
+    @abc.abstractmethod
+    def check_token(self) -> bool:
+        pass

simvue/factory/offline.py

@@ -0,0 +1,178 @@
+import json
+import os
+import time
+import uuid
+import typing
+import glob
+import logging
+import pathlib
+
+from simvue.utilities import get_offline_directory, create_file, prepare_for_api, skip_if_failed
+from simvue.factory.base import SimvueBaseClass
+
+logger = logging.getLogger(__name__)
+
+
+class Offline(SimvueBaseClass):
+    """
+    Class for offline runs
+    """
+    def __init__(self, name: str, uniq_id: str, suppress_errors: bool=True) -> None:
+        super().__init__(name, uniq_id, suppress_errors)
+
+        self._directory: str = os.path.join(get_offline_directory(), self._uuid)
+
+        os.makedirs(self._directory, exist_ok=True)
+
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def _write_json(self, filename: str, data: dict[str, typing.Any]) -> None:
+        """
+        Write JSON to file
+        """
+        if not os.path.isdir(os.path.dirname(filename)):
+            self._error("Cannot write file '{filename}', parent directory does not exist")
+
+        try:
+            with open(filename, 'w') as fh:
+                json.dump(data, fh)
+        except Exception as err:
+            self._error(f"Unable to write file {filename} due to {str(err)}")
+
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def _mock_api_post(self, prefix: str, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        unique_id = time.time()
+        filename = os.path.join(self._directory, f"{prefix}-{unique_id}.json")
+        self._write_json(filename, data)
+        return data
+
+    @skip_if_failed("_aborted", "_suppress_errors", (None, None))
+    def create_run(self, data) -> tuple[typing.Optional[str], typing.Optional[str]]:
+        """
+        Create a run
+        """
+        if not self._directory:
+            self._logger.error("No directory specified")
+            return (None, None)
+        try:
+            os.makedirs(self._directory, exist_ok=True)
+        except Exception as err:
+            self._logger.error('Unable to create directory %s due to: %s', self._directory, str(err))
+            return (None, None)
+        
+        filename = f"{self._directory}/run.json"
+
+        logger.debug(f"Creating run in '{filename}'")
+
+        if 'name' not in data:
+            data['name'] = None
+
+        self._write_json(filename, data)
+
+        status = data['status']
+        filename = f"{self._directory}/{status}"
+        create_file(filename)
+
+        return (self._name, self._id)
+
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def update(self, data) -> typing.Optional[dict[str, typing.Any]]:
+        """
+        Update metadata, tags or status
+        """
+        unique_id = time.time()
+        filename = f"{self._directory}/update-{unique_id}.json"
+        self._write_json(filename, data)
+
+        if 'status' in data:
+            status = data['status']
+            if not self._directory or not os.path.exists(self._directory):
+                self._error("No directory defined for writing")
+                return None
+            filename = f"{self._directory}/{status}"
+            
+            logger.debug(f"Writing API data to file '{filename}'")
+            
+            create_file(filename)
+
+            if status == 'completed':
+                status_running = f"{self._directory}/running"
+                if os.path.isfile(status_running):
+                    os.remove(status_running)
+
+        return data
+
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def set_folder_details(self, data) -> typing.Optional[dict[str, typing.Any]]:
+        """
+        Set folder details
+        """
+        unique_id = time.time()
+        filename = f"{self._directory}/folder-{unique_id}.json"
+        self._write_json(filename, data)
+        return data
+
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def save_file(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        """
+        Save file
+        """
+        if 'pickled' in data:
+            temp_file = f"{self._directory}/temp-{uuid.uuid4()}.pickle"
+            with open(temp_file, 'wb') as fh:
+                fh.write(data['pickled'])
+            data['pickledFile'] = temp_file
+        unique_id = time.time()
+        filename = os.path.join(self._directory, f"file-{unique_id}.json")
+        self._write_json(filename, prepare_for_api(data, False))
+        return data
+
+    def add_alert(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        """
+        Add an alert
+        """
+        return self._mock_api_post("alert", data)
+    
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def set_alert_state(self, alert_id: str, status: str) -> typing.Optional[dict[str, typing.Any]]:
+        if not os.path.exists(_alert_file := os.path.join(self._directory, f"alert-{alert_id}.json")):
+            self._error(f"Failed to retrieve alert '{alert_id}' for modification")
+            return None
+        
+        with open(_alert_file) as alert_in:
+            _alert_data = json.load(alert_in)
+        
+        _alert_data |= {"run": self._id, "alert": alert_id, "status": status}
+
+        self._write_json(_alert_file, _alert_data)
+
+        return _alert_data
+    
+    @skip_if_failed("_aborted", "_suppress_errors", [])
+    def list_alerts(self) -> list[dict[str, typing.Any]]:
+        return [
+            json.load(open(alert_file))
+            for alert_file
+            in glob.glob(os.path.join(self._directory, "alert-*.json"))
+        ]
+
+    def send_metrics(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        """
+        Send metrics
+        """
+        return self._mock_api_post("metrics", data)
+
+    def send_event(self, data: dict[str, typing.Any]) -> typing.Optional[dict[str, typing.Any]]:
+        """
+        Send event
+        """
+        return self._mock_api_post("event", data)
+    
+    @skip_if_failed("_aborted", "_suppress_errors", None)
+    def send_heartbeat(self) -> typing.Optional[dict[str, typing.Any]]:
+        logger.debug(f"Creating heartbeat file: {os.path.join(self._directory, 'heartbeat')}")
+        pathlib.Path(os.path.join(self._directory, "heartbeat"), "a").touch()
+        return {"success": True}
+
+    @skip_if_failed("_aborted", "_suppress_errors", False)
+    def check_token(self) -> bool:
+        return True