first draft

refactor

fix typo

refactor

wip

update  github action to trigger on pull requests

refactor wip

remove trace_info

add recipy

cleanup

refactor

refactor

clean

tasks

raise when missing config

added entry cli

wip

wip

first /

rename queues to tasks

refactor

wip

wip

secret config with no "."

fix

fix

fix

use tmp to download

wip

wip

wip

wip

added heartbeat

fix

up

up

up

.github/workflows/python-publish.yml

@@ -6,7 +6,7 @@ name: Upload Python Package
 on:
 
   push: {}
-
+  pull_request:
   release:
     types: [created]
 

README.md

@@ -5,7 +5,7 @@ This project is **BETA** and will be **Experimental** for the foreseeable future
 Interfaces and functionality are likely to change, and the project itself may be scrapped.
 **DO NOT** use this software in any project/software that is operational.
 
-A package to manahe a registry or data-driven forecasts.
+A package to manage a registry or data-driven forecasts.
 
 ## Documentation
 

pyproject.toml

@@ -51,15 +51,19 @@ dynamic = [
 ]
 dependencies = [
   "anemoi-datasets",
+  "jsonpatch",
+  "requests",
 ]
 
 optional-dependencies.all = [
+  "boto3",
 ]
+
 optional-dependencies.dev = [
+  "boto3",
   "nbsphinx",
   "pandoc",
   "pytest",
-  "requests",
   "sphinx",
   "sphinx-argparse",
   "sphinx-rtd-theme",
@@ -70,14 +74,17 @@ optional-dependencies.dev = [
 optional-dependencies.docs = [
   "nbsphinx",
   "pandoc",
-  "requests",
   "sphinx",
   "sphinx-argparse",
   "sphinx-rtd-theme",
   "termcolor",
   "tomli",
 ]
 
+optional-dependencies.s3 = [
+  "boto3",
+]
+
 optional-dependencies.tests = [
   "pytest",
 ]

src/anemoi/registry/__init__.py

@@ -14,10 +14,13 @@
 
 
 def config():
-    from anemoi.utils.config import DotDict
     from anemoi.utils.config import load_config
 
-    config = load_config().get("registry")
+    config = load_config(secrets=["api_token"])
     if not config:
-        LOG.warning(f"No 'registry' section in config. Config is {load_config()}. Limited functionalities.")
-    return DotDict(config)
+        raise ValueError("Anemoi config is required.")
+
+    config = config.get("registry")
+    if not config:
+        raise ValueError("Section 'registry' is missing in config.")
+    return config

src/anemoi/registry/commands/__init__.py

@@ -19,6 +19,6 @@
 COMMANDS = register_commands(
     os.path.dirname(__file__),
     __name__,
-    lambda x: x.command(),
+    lambda x: x.command() if hasattr(x, "command") else None,
     lambda name, error: Failed(name, error),
 )

src/anemoi/registry/commands/base.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+# (C) Copyright 2024 ECMWF.
+#
+# This software is licensed under the terms of the Apache Licence Version 2.0
+# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+# In applying this licence, ECMWF does not waive the privileges and immunities
+# granted to it by virtue of its status as an intergovernmental organisation
+# nor does it submit to any jurisdiction.
+#
+
+"""Command place holder. Delete when we have real commands.
+
+"""
+
+import logging
+import os
+
+from ..entry import CatalogueEntryNotFound
+from . import Command
+
+LOG = logging.getLogger(__name__)
+
+
+class BaseCommand(Command):
+    internal = True
+    timestamp = True
+
+    def check_arguments(self, args):
+        pass
+
+    def is_path(self, name_or_path):
+        return os.path.exists(name_or_path)
+
+    def is_identifier(self, name_or_path):
+        try:
+            self.entry_class(key=name_or_path)
+            return True
+        except CatalogueEntryNotFound:
+            return False
+
+    def process_task(self, entry, args, k, func_name=None, /, **kwargs):
+        """
+        Call the method `k` on the entry object.
+        The args/kwargs given to the method are extracted from from the argument `k` in the `args` object.
+
+        Additionally the argument `k` is casted to the correct type,
+        depending on if this is a string, int, float, list or dict, or a boolean.
+
+        The provided **kwargs are also passed to the method.
+        The method name can be changed by providing the `func_name` argument.
+        """
+
+        assert isinstance(k, str), k
+        if func_name is None:
+            func_name = k
+
+        v = getattr(args, k)
+
+        if v is None:
+            return
+        if v is True:
+            LOG.debug(f"{entry.key} : Processing task {k}")
+            return getattr(entry, func_name)(**kwargs)
+        if v is False:
+            return
+        if isinstance(v, (str, int, float)):
+            LOG.debug(f"{entry.key} : Processing task {k} with {v}")
+            return getattr(entry, func_name)(v, **kwargs)
+        if isinstance(v, list):
+            v_str = ", ".join(str(x) for x in v)
+            LOG.debug(f"{entry.key} : Processing task {k} with {v_str}")
+            return getattr(entry, func_name)(*v, **kwargs)
+        if isinstance(v, dict):
+            v_str = ", ".join(f"{k_}={v_}" for k_, v_ in v.items())
+            LOG.debug(f"{entry.key} : Processing task {k} with {v_str}")
+            return getattr(entry, func_name)(**v, **kwargs)
+        raise ValueError(f"Invalid task {k}={v}. type(v)= {type(v)}")
+
+    def run(self, args):
+        LOG.debug(f"anemoi-registry args: {args}")
+        name_or_path = args.NAME_OR_PATH
+        entry = self.get_entry(name_or_path)
+        self._run(entry, args)
+
+    def get_entry(self, name_or_path):
+        if self.is_path(name_or_path):
+            LOG.info(f"Found local {self.kind} at {name_or_path}")
+            return self.entry_class(path=name_or_path)
+
+        if self.is_identifier(name_or_path):
+            LOG.info(f"Processing {self.kind} with identifier '{name_or_path}'")
+            return self.entry_class(key=name_or_path)
+
+    def run_from_identifier(self, *args, **kwargs):
+        raise NotImplementedError()
+
+    def run_from_path(self, *args, **kwargs):
+        raise NotImplementedError()

src/anemoi/registry/commands/datasets.py

@@ -14,13 +14,15 @@
 
 import logging
 
-from ..entry import DatasetCatalogueEntry
-from ._base import BaseCommand
+from ..entry.dataset import DatasetCatalogueEntry
+from .base import BaseCommand
 
 LOG = logging.getLogger(__name__)
 
 
 class Datasets(BaseCommand):
+    """Manage datasets in the catalogue. Register, add locations, set status, etc."""
+
     internal = True
     timestamp = True
     entry_class = DatasetCatalogueEntry
@@ -35,72 +37,26 @@ def add_arguments(self, command_parser):
             action="store_true",
         )
         # command_parser.add_argument("--delete", help=f"Delete the {self.kind} from the catalogue and from any other location", action="store_true")
-        command_parser.add_argument("--json", help="Output json record", action="store_true")
-
         command_parser.add_argument("--set-status", help="Set the status to the dataset")
-        command_parser.add_argument("--add-location", nargs="+", help="Add a location to the dataset")
+        command_parser.add_argument("--add-recipe", help="Add a recipe file")
+        command_parser.add_argument(
+            "--add-location",
+            nargs="+",
+            help="Path to add a location to the dataset. Implies --platform",
+        )
+        command_parser.add_argument("--platform", help="Platform to add the location to.")
 
     def check_arguments(self, args):
         pass
 
-    def run_from_identifier(
-        self,
-        identifier,
-        add_location,
-        set_status,
-        unregister,
-        json,
-        remove_location=False,
-        **kwargs,
-    ):
-        self.warn_unused_arguments(kwargs)
-
-        entry = self.entry_class(key=identifier)
-
-        if unregister:
-            entry.unregister()
-        if add_location:
-            entry.add_location(**add_location)
-        if remove_location:
-            entry.remove_location(**remove_location)
-        if set_status:
-            entry.set_status(set_status)
-
-        if json:
-            print(entry.as_json())
-
-    def run_from_path(
-        self,
-        path,
-        register,
-        unregister,
-        add_location,
-        json,
-        set_status,
-        # remove_location,
-        # upload,
-        # upload_uri_pattern,
-        **kwargs,
-    ):
-        self.warn_unused_arguments(kwargs)
-
-        entry = self.entry_class(path=path)
-
-        if register:
-            entry.register()
-        if unregister:
-            entry.unregister()
-        if add_location:
-            entry.add_location(**add_location)
-        # if remove_location:
-        #    entry.remove_location(**remove_location)
-        if set_status:
-            entry.set_status(set_status)
-        # if delete:
-        #    entry.delete()
-
-        if json:
-            print(entry.as_json())
+    def _run(self, entry, args):
+        # order matters
+        self.process_task(entry, args, "unregister")
+        self.process_task(entry, args, "register")
+        # self.process_task(entry, args, "remove_location")
+        self.process_task(entry, args, "add_location", platform=args.platform)
+        self.process_task(entry, args, "add_recipe")
+        self.process_task(entry, args, "set_status")
 
 
 command = Datasets