Merge branch 'v2.0' into wk9874/alert_deduplication

Author: wk9874

.pre-commit-config.yaml

@@ -23,7 +23,7 @@ repos:
         args: [--branch, main, --branch, dev]
       - id: check-added-large-files
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.9.4
+    rev: v0.9.6
     hooks:
       - id: ruff
         args: [ --fix, --exit-non-zero-on-fix, "--ignore=C901" ]

poetry.lock

@@ -53,6 +53,7 @@ dependencies = [
   "psutil (>=6.1.1,<7.0.0)",
   "tenacity (>=9.0.0,<10.0.0)",
   "typing-extensions (>=4.12.2,<5.0.0) ; python_version < \"3.11\"",
+  "deepmerge (>=2.0,<3.0)",
 ]
 
 [project.urls]

pyproject.toml

@@ -17,6 +17,7 @@
 import msgpack
 import pydantic
 
+from simvue.utilities import staging_merger
 from simvue.config.user import SimvueConfiguration
 from simvue.exception import ObjectNotFoundError
 from simvue.version import __version__
@@ -164,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] = {}
 
@@ -527,7 +532,8 @@ def _cache(self) -> None:
             with self._local_staging_file.open() as in_f:
                 _local_data = json.load(in_f)
 
-        _local_data |= self._staging
+        staging_merger.merge(_local_data, self._staging)
+
         with self._local_staging_file.open("w", encoding="utf-8") as out_f:
             json.dump(_local_data, out_f, indent=2)
 

simvue/api/objects/base.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/metrics.py

@@ -214,14 +214,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/api/objects/run.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/parameters.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/config/user.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
@@ -418,7 +422,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(
@@ -464,7 +470,7 @@ def _start(self, reconnect: bool = False) -> bool:
 
         logger.debug("Starting run")
 
-        if self._sv_obj:
+        if self._sv_obj and self._sv_obj.status != "running":
             self._sv_obj.status = self._status
             self._sv_obj.commit()
 
@@ -564,6 +570,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,
@@ -585,6 +592,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.
@@ -620,13 +630,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(
@@ -637,7 +644,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"
             )
@@ -686,6 +695,7 @@ def init(
         self._sv_obj.metadata = (metadata or {}) | git_info(os.getcwd()) | environment()
         self._sv_obj.heartbeat_timeout = timeout
         self._sv_obj.alerts = []
+        self._sv_obj.notifications = notification
 
         if self._status == "running":
             self._sv_obj.system = get_system()

simvue/run.py

@@ -11,8 +11,8 @@
 import os
 import pathlib
 import typing
-
 import jwt
+from deepmerge import Merger
 
 from datetime import timezone
 from simvue.models import DATETIME_FORMAT
@@ -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
@@ -395,3 +396,18 @@ def get_mimetype_for_file(file_path: pathlib.Path) -> str:
     """Return MIME type for the given file"""
     _guess, *_ = mimetypes.guess_type(file_path)
     return _guess or "application/octet-stream"
+
+
+# Create a new Merge strategy for merging local file and staging attributes
+staging_merger = Merger(
+    # pass in a list of tuple, with the
+    # strategies you are looking to apply
+    # to each type.
+    [(list, ["override"]), (dict, ["merge"]), (set, ["union"])],
+    # next, choose the fallback strategies,
+    # applied to all other types:
+    ["override"],
+    # finally, choose the strategies in
+    # the case where the types conflict:
+    ["override"],
+)