Merge branch 'v2.0' into wk9874/offline_fixes

Author: wk9874

simvue/api/objects/base.py

@@ -165,10 +165,14 @@ def __init__(
             )
         )
 
-        self._headers: dict[str, str] = {
-            "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}",
-            "User-Agent": _user_agent or f"Simvue Python client {__version__}",
-        }
+        self._headers: dict[str, str] = (
+            {
+                "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}",
+                "User-Agent": _user_agent or f"Simvue Python client {__version__}",
+            }
+            if not self._offline
+            else {}
+        )
 
         self._params: dict[str, str] = {}
 

simvue/api/objects/metrics.py

@@ -37,7 +37,7 @@ def __init__(
     def new(
         cls, *, run: str, offline: bool = False, metrics: list[MetricSet], **kwargs
     ):
-        """Create a new Events entry on the Simvue server"""
+        """Create a new Metrics entry on the Simvue server"""
         return Metrics(
             run=run,
             metrics=[metric.model_dump() for metric in metrics],
@@ -51,27 +51,23 @@ def get(
         cls,
         metrics: list[str],
         xaxis: typing.Literal["timestamp", "step", "time"],
+        runs: list[str],
         *,
         count: pydantic.PositiveInt | None = None,
         offset: pydantic.PositiveInt | None = None,
         **kwargs,
     ) -> typing.Generator[MetricSet, None, None]:
         _class_instance = cls(_read_only=True, _local=True)
-        if (
-            _data := cls._get_all_objects(
-                count,
-                offset,
-                metrics=json.dumps(metrics),
-                xaxis=xaxis,
-                **kwargs,
-            ).get("data")
-        ) is None:
-            raise RuntimeError(
-                f"Expected key 'data' for retrieval of {_class_instance.__class__.__name__.lower()}s"
-            )
-
-        for _entry in _data:
-            yield MetricSet(**_entry)
+        _data = cls._get_all_objects(
+            count,
+            offset,
+            metrics=json.dumps(metrics),
+            runs=json.dumps(runs),
+            xaxis=xaxis,
+            **kwargs,
+        )
+        # TODO: Temp fix, just return the dictionary. Not sure what format we really want this in...
+        return _data
 
     @pydantic.validate_call
     def span(self, run_ids: list[str]) -> dict[str, int | float]:

simvue/api/objects/run.py

@@ -215,14 +215,16 @@ def heartbeat_timeout(self, time_seconds: int | None) -> None:
 
     @property
     @staging_check
-    def notifications(self) -> typing.Literal["none", "email"]:
-        return self._get_attribute("notifications")
+    def notifications(self) -> typing.Literal["none", "all", "error", "lost"]:
+        return self._get_attribute("notifications")["state"]
 
     @notifications.setter
     @write_only
     @pydantic.validate_call
-    def notifications(self, notifications: typing.Literal["none", "email"]) -> None:
-        self._staging["notifications"] = notifications
+    def notifications(
+        self, notifications: typing.Literal["none", "all", "error", "lost"]
+    ) -> None:
+        self._staging["notifications"] = {"state": notifications}
 
     @property
     @staging_check

simvue/config/parameters.py

@@ -21,19 +21,23 @@
 
 
 class ServerSpecifications(pydantic.BaseModel):
-    url: pydantic.AnyHttpUrl
-    token: pydantic.SecretStr
+    url: pydantic.AnyHttpUrl | None
+    token: pydantic.SecretStr | None
 
     @pydantic.field_validator("url")
     @classmethod
-    def url_to_api_url(cls, v: typing.Any) -> str:
+    def url_to_api_url(cls, v: typing.Any) -> str | None:
+        if not v:
+            return
         if f"{v}".endswith("/api"):
             return f"{v}"
         _url = URL(f"{v}") / "api"
         return f"{_url}"
 
     @pydantic.field_validator("token")
-    def check_token(cls, v: typing.Any) -> str:
+    def check_token(cls, v: typing.Any) -> str | None:
+        if not v:
+            return
         value = v.get_secret_value()
         if not (expiry := get_expiry(value)):
             raise AssertionError("Failed to parse Simvue token - invalid token form")

simvue/config/user.py

@@ -212,10 +212,10 @@ def fetch(
 
         _run_mode = mode or _config_dict["run"].get("mode") or "online"
 
-        if not _server_url:
+        if not _server_url and _run_mode != "offline":
             raise RuntimeError("No server URL was specified")
 
-        if not _server_token:
+        if not _server_token and _run_mode != "offline":
             raise RuntimeError("No server token was specified")
 
         _config_dict["server"]["token"] = _server_token

simvue/run.py

@@ -30,8 +30,8 @@
 import psutil
 
 from simvue.api.objects.alert.fetch import Alert
-from simvue.api.objects.folder import Folder, get_folder_from_path
-from simvue.exception import ObjectNotFoundError, SimvueRunError
+from simvue.api.objects.folder import Folder
+from simvue.exception import SimvueRunError
 from simvue.utilities import prettify_pydantic
 
 
@@ -184,9 +184,13 @@ def __init__(
             if self._user_config.metrics.resources_metrics_interval < 1
             else self._user_config.metrics.resources_metrics_interval
         )
-        self._headers: dict[str, str] = {
-            "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}"
-        }
+        self._headers: dict[str, str] = (
+            {
+                "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}"
+            }
+            if mode != "offline"
+            else {}
+        )
         self._sv_obj: RunObject | None = None
         self._pid: int | None = 0
         self._shutdown_event: threading.Event | None = None
@@ -411,7 +415,9 @@ def _create_dispatch_callback(
         if self._user_config.run.mode == "online" and not self._id:
             raise RuntimeError("Expected identifier for run")
 
-        if not self._user_config.server.url or not self._sv_obj:
+        if (
+            self._user_config.run.mode != "offline" and not self._user_config.server.url
+        ) or not self._sv_obj:
             raise RuntimeError("Cannot commence dispatch, run not initialised")
 
         def _dispatch_callback(
@@ -459,7 +465,7 @@ def _start(self, reconnect: bool = False) -> bool:
 
         self._start_time = time.time()
 
-        if self._sv_obj:
+        if self._sv_obj and self._sv_obj.status != "running":
             self._sv_obj.status = self._status
             self._sv_obj.started = self._start_time
             self._sv_obj.commit()
@@ -558,6 +564,7 @@ def init(
         folder: typing.Annotated[
             str, pydantic.Field(None, pattern=FOLDER_REGEX)
         ] = None,
+        notification: typing.Literal["none", "all", "error", "lost"] = "none",
         running: bool = True,
         retention_period: str | None = None,
         timeout: int | None = 180,
@@ -579,6 +586,9 @@ def init(
             description of the run, by default None
         folder : str, optional
             folder within which to store the run, by default "/"
+        notification: typing.Literal["none", "all", "error", "lost"], optional
+            whether to notify the user by email upon completion of the run if
+            the run is in the specified state, by default "none"
         running : bool, optional
             whether to set the status as running or created, the latter implying
             the run will be commenced at a later time. Default is True.
@@ -614,13 +624,10 @@ def init(
 
         self._term_color = not no_color
 
-        try:
-            self._folder = get_folder_from_path(path=folder)
-        except ObjectNotFoundError:
-            self._folder = Folder.new(
-                path=folder, offline=self._user_config.run.mode == "offline"
-            )
-            self._folder.commit()  # type: ignore
+        self._folder = Folder.new(
+            path=folder, offline=self._user_config.run.mode == "offline"
+        )
+        self._folder.commit()  # type: ignore
 
         if isinstance(visibility, str) and visibility not in ("public", "tenant"):
             self._error(
@@ -631,7 +638,9 @@ def init(
             self._error("invalid mode specified, must be online, offline or disabled")
             return False
 
-        if not self._user_config.server.token or not self._user_config.server.url:
+        if self._user_config.run.mode != "offline" and (
+            not self._user_config.server.token or not self._user_config.server.url
+        ):
             self._error(
                 "Unable to get URL and token from environment variables or config file"
             )
@@ -683,6 +692,7 @@ def init(
         self._sv_obj.heartbeat_timeout = timeout
         self._sv_obj.alerts = []
         self._sv_obj.created = time.time()
+        self._sv_obj.notifications = notification
 
         if self._status == "running":
             self._sv_obj.system = get_system()

simvue/utilities.py

@@ -50,16 +50,17 @@ def find_first_instance_of_file(
     if isinstance(file_names, str):
         file_names = [file_names]
 
-    for root, _, files in os.walk(os.getcwd(), topdown=False):
-        for file_name in file_names:
-            if file_name in files:
-                return pathlib.Path(root).joinpath(file_name)
+    for file_name in file_names:
+        _user_file = pathlib.Path.cwd().joinpath(file_name)
+        if _user_file.exists():
+            return _user_file
 
     # If the user is running on different mounted volume or outside
     # of their user space then the above will not return the file
     if check_user_space:
         for file_name in file_names:
-            if os.path.exists(_user_file := pathlib.Path.home().joinpath(file_name)):
+            _user_file = pathlib.Path.home().joinpath(file_name)
+            if _user_file.exists():
                 return _user_file
 
     return None

tests/functional/test_dispatch.py

@@ -67,9 +67,10 @@ def callback(___: list[typing.Any], _: str, args=check_dict, var=variable) -> No
     event.set()
 
     dispatcher.join()
+    time.sleep(0.1)
 
     for variable in variables:
-        assert check_dict[variable]["counter"] >= 2 if overload_buffer else 1, f"Check of counter for dispatcher '{variable}' failed with count = {check_dict[variable]['counter']}"
+        assert check_dict[variable]["counter"] >= (2 if overload_buffer else 1), f"Check of counter for dispatcher '{variable}' failed with count = {check_dict[variable]['counter']}"
     assert time.time() - start_time < time_threshold
 
 

tests/unit/test_events.py

@@ -7,6 +7,7 @@
 
 from simvue.api.objects import Events, Folder, Run
 from simvue.models import DATETIME_FORMAT
+from simvue.sender import sender
 
 @pytest.mark.api
 @pytest.mark.online
@@ -30,3 +31,39 @@ def test_events_creation_online() -> None:
     _run.delete()
     _folder.delete(recursive=True, delete_runs=True, runs_only=False)
 
+@pytest.mark.api
+@pytest.mark.offline
+def test_events_creation_offline() -> None:
+    _uuid: str = f"{uuid.uuid4()}".split("-")[0]
+    _folder_name = f"/simvue_unit_testing/{_uuid}"
+    _folder = Folder.new(path=_folder_name, offline=True)
+    _run = Run.new(folder=_folder_name, offline=True)
+    _folder.commit()
+    _run.commit()
+    _timestamp = datetime.datetime.now().strftime(DATETIME_FORMAT)
+    _events =   Events.new(
+        run=_run.id,
+        events=[
+            {"message": "This is a test!", "timestamp": _timestamp}
+        ],
+        offline=True
+    )
+    _events.commit()
+    with _events._local_staging_file.open() as in_f:
+        _local_data = json.load(in_f)
+        
+    assert _local_data.get("run") == _run.id
+    assert _local_data.get("events")[0].get("message") == "This is a test!"
+    assert _local_data.get("events")[0].get("timestamp") == _timestamp
+    
+    _id_mapping = sender(_events._local_staging_file.parents[1], 1, 10, ["folders", "runs", "events"])
+    time.sleep(1)
+    
+    # Get online version of events
+    _online_events = Events(_id_mapping.get(_events.id))
+    _event_content = next(_online_events.get(run_id=_id_mapping.get(_run.id)))
+    assert _event_content.message == "This is a test!"
+    assert _event_content.timestamp == _timestamp
+
+    _run.delete()
+    _folder.delete(recursive=True, delete_runs=True, runs_only=False)

tests/unit/test_metrics.py

@@ -6,6 +6,8 @@
 import uuid
 
 from simvue.api.objects import Metrics, Folder, Run
+from simvue.models import DATETIME_FORMAT
+from simvue.sender import sender
 
 @pytest.mark.api
 @pytest.mark.online
@@ -38,9 +40,60 @@ def test_metrics_creation_online() -> None:
     )
     assert _metrics.to_dict()
     _metrics.commit()
-    assert _metrics.get(metrics=["x", "y", "z"], xaxis="step")
+    assert _metrics.get(metrics=["x", "y", "z"], xaxis="step", runs=[_run.id])
     assert _metrics.span(run_ids=[_run.id])
     assert _metrics.names(run_ids=[_run.id])
     _run.delete()
     _folder.delete(recursive=True, delete_runs=True, runs_only=False)
 
+@pytest.mark.api
+@pytest.mark.offline
+def test_metrics_creation_offline() -> None:
+    _uuid: str = f"{uuid.uuid4()}".split("-")[0]
+    _folder_name = f"/simvue_unit_testing/{_uuid}"
+    _folder = Folder.new(path=_folder_name, offline=True)
+    _run = Run.new(name="hello", folder=_folder_name, offline=True)
+    _folder.commit()
+    _run.commit()
+    
+    _values = {
+        "x": 1,
+        "y": 2.0,
+        "z": True
+    }
+    _time: int = 1
+    _step: int = 1
+    _timestamp = datetime.datetime.now().strftime(DATETIME_FORMAT)
+    _metrics = Metrics.new(
+        run=_run.id,
+        metrics=[
+            {
+                "timestamp": _timestamp,
+                "time": _time,
+                "step": _step,
+                "values": _values,
+            }
+        ],
+        offline=True
+    )
+    _metrics.commit()
+    with _metrics._local_staging_file.open() as in_f:
+        _local_data = json.load(in_f)
+        
+    assert _local_data.get("run") == _run.id
+    assert _local_data.get("metrics")[0].get("values") == _values
+    assert _local_data.get("metrics")[0].get("timestamp") == _timestamp
+    assert _local_data.get("metrics")[0].get("step") == _step
+    assert _local_data.get("metrics")[0].get("time") == _time
+    
+    _id_mapping = sender(_metrics._local_staging_file.parents[1], 1, 10, ["folders", "runs", "metrics"])
+    time.sleep(1)
+    
+    # Get online version of metrics
+    _online_metrics = Metrics(_id_mapping.get(_metrics.id))
+    _data = _online_metrics.get(metrics=["x", "y", "z"], runs=[_id_mapping.get(_run.id)], xaxis="step")
+    assert sorted(_online_metrics.names(run_ids=[_id_mapping.get(_run.id)])) == sorted(_values.keys())
+    assert _data.get(_id_mapping.get(_run.id)).get('y')[0].get('value') == 2.0
+    assert _data.get(_id_mapping.get(_run.id)).get('y')[0].get('step') == 1
+    _run.delete()
+    _folder.delete(recursive=True, delete_runs=True, runs_only=False)