Implement RO-crate creator

.github/workflows/workflow-test.yml

@@ -27,7 +27,7 @@ jobs:
                     import os
                     import subprocess
                     fn = [f for f in os.listdir() if f.endswith('whl') and f.startswith('tomato')][0]
-                    subprocess.run(["pip", "install", f"{fn}[testing]"])
+                    subprocess.run(["pip", "install", f"{fn}[testing,rocrate]"])
           - name: Print version
             shell: bash
             run: tomato --version

.gitignore

@@ -130,3 +130,5 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+.vscode/

docs/source/quickstart.rst

@@ -159,6 +159,10 @@ The *settings file* contains the basic information required to start the ``tomat
     [devices]
     config = '/home/kraus/.config/tomato/1.0a1/devices.yml'
 
+    [repositories]
+    default.endpoint = 'https://example.endpoint/uri-here'
+    default.api_key = 'your-api-key-here'
+
     [drivers]
     example_counter.idle_measurement_interval = 1
 
@@ -173,13 +177,40 @@ Finally, another path, *logdir*, is used to specify where logs for **tomato** ar
 - ``$env:localappdata\dgbowl\tomato\<version>`` on Windows,
 - ``$HOME/.cache/tomato/<version>/log`` on Linux.
 
-In the default *settings file* shown above, the following entries are specified:
+**jobs** section
+****************
+
+Section for configuring behaviour of **tomato** *jobs*. The following entries can be specified:
+
+- ``jobs.storage`` which is the directory where the data of **tomato** *jobs* will be stored
+- ``jobs.dbpath`` which is the location of the ``sqlite3`` database used to track *jobs*
+
+**devices** section
+*******************
+
+Section for configuring **tomato** *devices*. The following entries can be specified:
+
+- ``devices.config`` which points to a ``yaml``-formatted |devfile|_, defining the hardware configuration of the devices managed by **tomato**
+
+**repositories** section
+************************
+
+Section for settings related to the upload of completed *payloads* to pre-configured repositories using the RO-crate mechanism. By default, all *payloads* will attempt to submit to the ``default`` repository, which can be configured as follows:
+
+- ``repositories.default.endpoint`` should be the API endpoint of the repository where the RO-crates should be sent
+- ``repositories.default.api_key`` should contain any API key / token necessary for authenticating the upload
+
+More than one repository can be specified using a separate sub-section. If a ``default`` repository is not specified, RO-crate objects will **not** be created and data from completed *payloads* will **not** be uploaded.
+
+**drivers** section
+*******************
+
+Additional, *driver*-specific settings may be provided in this section. Each *driver* should be configured separately in its own sub-section. The following entries can be specified for all *drivers*:
 
-- ``jobs.storage`` which is the directory where the data of **tomato** *jobs* will be stored,
-- ``jobs.dbpath`` which is the location of the ``sqlite3`` database used to track *jobs*,
-- ``devices.config`` which points to a ``yaml``-formatted |devfile|_, defining the hardware configuration of the devices managed by **tomato**.
+- ``drivers.<driver_name>.idle_measurement_interval`` for configuring the interval between measurements performed when any registered *component* is idle,
+- ``drivers.<driver_name>.lpp_timeout`` for configuring the timeout for communication between **tomato** and the *driver* process, in seconds
 
-Additional, *driver*-specific settings may be provided in the ``[drivers]`` section, following the example of the ``drivers.example_counter.idle_measurement_interval`` entry. These *driver*-specific settings are passed to each *driver* when its process is launched and the :class:`DriverInterface` is initialised, and can therefore contain paths to various libraries or other files necessary for the *driver* to function.
+Further *driver*-specific settings, such as ``dllpath`` or ``calibration``, can be specified here. All of these *driver*-specific settings are passed to each *driver* when its process is launched and the :class:`DriverInterface` is initialised, and can therefore contain paths to various libraries or other files necessary for the *driver* to function.
 
 Devices file
 ````````````
@@ -306,8 +337,10 @@ As of ``tomato-2.1``, execution of a certain :class:`Task` can be postponed and
 As of ``tomato-2.0``, the :obj:`task_params` specified in the *payload* are validated by the device *driver*. In particular, the values of all entries within :obj:`task_params` are checked for compatibility with the :obj:`Attr.type` and :obj:`Attr.options` (if supplied), as well as against the provided :obj:`Attr.minimum` and :obj:`Attr.maximum`, as specified in the matching :class:`~tomato.driverinterface_2_1.Attr`.
 
 .. autopydantic_model:: dgbowl_schemas.tomato.payload.Payload
+  :no-index:
 
 .. autopydantic_model:: dgbowl_schemas.tomato.payload.Task
+  :no-index:
 
 .. |devfile| replace:: *devices file*
 

docs/source/version.rst

@@ -18,8 +18,14 @@ Changes from ``tomato-2.1`` include:
 
 - Fixes many bugs due to the :func:`cmp_measure` function race condition with running tasks.
 - Introduces the "lazy pirate" pattern in ``tomato-job`` processes, which should make jobs more reliable.
+- Added functionality for automatically creating RO-crates from completed jobs.
+
+- A new ``Payload-2.2``:
+
+  - The ``settings.snapshot.snapshot_interval`` replaces ``settings.snapshot.frequency``. The ``snapshot_interval`` can be provided as :class:`str`, which will be converted to the number of seconds using :mod:`pint`.
+  - The ``sample.identifier`` replaces ``sample.name``. A new required ``user.identifier`` section
+  - The ``settings.output.repositories`` entry is added, which allows users to select which repository configured in the ``settings.toml`` file will be used to upload the job data.
 
-- A new ``Payload-2.2``, where the ``settings.snapshot.snapshot_interval`` replaces ``settings.snapshot.frequency``. As a consequence, the ``snapshot_interval`` can be provided as :class:`str`, which will be converted to the number of seconds using :mod:`pint`.
 
 .. codeauthor::
     Peter Kraus

pyproject.toml

@@ -29,8 +29,7 @@ dependencies = [
     "toml >= 0.10",
     "pyyaml >= 6.0",
     "psutil >= 5.9",
-    # "dgbowl_schemas >= 124",
-    "dgbowl_schemas @ git+https://github.com/dgbowl/dgbowl-schemas.git@payload_2.2",
+    "dgbowl_schemas @ git+https://github.com/dgbowl/dgbowl-schemas.git",
     "pyzmq >= 25.1",
     "netcdf4 >= 1.7",
     "xarray >= 2024.10.0",
@@ -50,6 +49,9 @@ docs = [
     "autodoc-pydantic ~= 2.1",
     "sphinxcontrib-mermaid ~= 0.9.2",
 ]
+rocrate = [
+    "rocrate >= 0.14"
+]
 
 [project.urls]
 Documentation = "https://dgbowl.github.io/tomato/main"

src/tomato/__init__.py

@@ -31,6 +31,7 @@ def parse_args(parser, verbose, is_tomato=False):
     set_loglevel(verbosity)
 
     context = zmq.Context()
+    kwargs = {}
     if not is_tomato:
         status = tomato.status(**vars(args), context=context)
         if not status.success:
@@ -39,9 +40,10 @@ def parse_args(parser, verbose, is_tomato=False):
             else:
                 print(f"Failure: {status.msg}")
             return
+        kwargs["daemon"] = status.data
 
     if "func" in args:
-        ret = args.func(**vars(args), verbosity=verbosity, context=context)
+        ret = args.func(**vars(args), verbosity=verbosity, context=context, **kwargs)
         if args.yaml:
             print(yaml.dump(ret.dict()))
         else:

src/tomato/daemon/crates.py

@@ -0,0 +1,65 @@
+import logging
+from typing import Union
+
+try:
+    from rocrate.rocrate import ROCrate
+    from rocrate.model import Person, ContextEntity
+
+    _has_rocrate = True
+except ImportError:
+    _has_rocrate = False
+
+logger = logging.getLogger(__name__)
+
+
+PROFILE_URI = "https://github.com/MADICES/MADICES-2025/discussions/25"
+PROFILE_VER = "0.2"
+
+
+def RepositoryObject(
+    crate: "ROCrate", identifier: str = None, properties: dict = None
+) -> "ContextEntity":
+    if properties is None:
+        properties = {}
+    properties["@type"] = "RepositoryObject"
+    return ContextEntity(crate, identifier=identifier, properties=properties)
+
+
+def Profile(
+    crate: "ROCrate", identifier: str = PROFILE_URI, properties: dict = None
+) -> "ContextEntity":
+    if properties is None:
+        properties = {}
+    properties["@type"] = "Profile"
+    properties["version"] = PROFILE_VER
+    return ContextEntity(crate, identifier=identifier, properties=properties)
+
+
+def to_rocrate(
+    datapath: str, userid: str, sampleid: str, make_child: bool = True
+) -> Union["ROCrate", None]:
+    if _has_rocrate is False:
+        return None
+    crate = ROCrate()
+    profile = crate.add(Profile(crate))
+    author = crate.add(Person(crate, userid))
+    parent = crate.add(RepositoryObject(crate, sampleid))
+    target = parent
+    if make_child:
+        child = crate.add(RepositoryObject(crate, None))
+        parent["hasPart"] = child
+        target = child
+    crate.add_file(
+        datapath,
+        properties={
+            "encodingFormat": "application/netcdf",
+            "author": author,
+        },
+    )
+    dataset = crate.get("./")
+    dataset["conformsTo"] = profile
+    target["hasPart"] = dataset
+    logger.debug("RO-crate created, writing a zip file")
+    cratepath = f"{datapath[:-3]}.zip"
+    crate.write_zip(cratepath)
+    logger.debug("RO-crate written into '%s'", cratepath)

src/tomato/daemon/io.py

@@ -43,7 +43,7 @@ def load(daemon: Daemon):
     daemon.status = "running"
 
 
-def merge_netcdfs(job: Job, snapshot=False):
+def merge_netcdfs(job: Job, snapshot=False) -> str:
     """
     Merges the individual pickled :class:`xr.Datasets` of each Component found in :obj:`job.jobpath`
     into a single :class:`xr.DataTree`, which is then stored in the NetCDF file,
@@ -69,6 +69,7 @@ def merge_netcdfs(job: Job, snapshot=False):
     logger.debug("saving DataTree into a NetCDF file at '%s'", outpath)
     dt.to_netcdf(outpath)
     dt.close()
+    return outpath
 
 
 def data_to_pickle(ds: xr.Dataset, path: Path, role: str):

src/tomato/daemon/job.py

@@ -12,26 +12,34 @@
 
 """
 
+import argparse
+import json
+import logging
 import os
+import psutil
 import subprocess
-import logging
-import json
+import sys
 import time
-import argparse
-from importlib import metadata
+import xarray as xr
+import zmq
+
 from datetime import datetime, timezone, timedelta
+from importlib import metadata
 from pathlib import Path
 from threading import current_thread, Thread
-import zmq
-import psutil
-import sys
-import xarray as xr
-
+from tomato.daemon.crates import to_rocrate
 from tomato.daemon.io import merge_netcdfs, data_to_pickle
 from tomato.daemon import jobdb, lpp
-from tomato.models import Pipeline, Daemon, Component, Device, Driver, Job
-from dgbowl_schemas.tomato import to_payload
-from dgbowl_schemas.tomato.payload import Task
+from tomato.models import (
+    Pipeline,
+    Daemon,
+    Component,
+    Device,
+    Driver,
+    Job,
+    Task,
+    to_payload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -263,7 +271,7 @@ def action_queued_jobs(daemon, matched, req, dbpath):
         for pip in matched[job.id]:
             if not pip.ready:
                 continue
-            elif pip.sampleid != job.payload.sample.name:
+            elif pip.sampleid != job.payload.sample.identifier:
                 continue
             logger.info("job %d: found a matched & ready pip '%s'", job.id, pip.name)
 
@@ -273,10 +281,15 @@ def action_queued_jobs(daemon, matched, req, dbpath):
 
             logger.debug("job %d: storing jobdata.json", job.id)
             jpath = root / "jobdata.json"
+            repositories = {}
+            for repo, repoparams in daemon.settings["repositories"].items():
+                if repo in job.payload.settings.output.repositories:
+                    repositories[repo] = repoparams
             jobargs = {
                 "pipeline": pip.model_dump(),
                 "payload": job.payload.model_dump(),
                 "devices": {dn: dev.model_dump() for dn, dev in daemon.devs.items()},
+                "repositories": repositories,
                 "job": dict(id=job.id, path=str(root)),
             }
             with jpath.open("w", encoding="UTF-8") as of:
@@ -408,7 +421,9 @@ def tomato_job() -> None:
 
     with args.jobfile.open() as infile:
         jsdata = json.load(infile)
+
     payload = to_payload(**jsdata["payload"])
+    repositories = jsdata["repositories"]
 
     pip = jsdata["pipeline"]["name"]
     jobid = jsdata["job"]["id"]
@@ -421,7 +436,7 @@ def tomato_job() -> None:
         handlers=[logging.FileHandler(logpath, mode="a")],
     )
     logger = logging.getLogger(__name__)
-
+    logger.debug(f"{jsdata=}")
     logger.debug(f"{payload=}")
 
     verbosity = payload.settings.verbosity
@@ -472,7 +487,18 @@ def tomato_job() -> None:
     else:
         job.status = "ce"
     logger.info("writing final data to a NetCDF file")
-    merge_netcdfs(job)
+    outpath = merge_netcdfs(job)
+    if len(jsdata["repositories"]) > 0:
+        logger.debug(
+            "job configured with repositories: '%s'", list(repositories.keys())
+        )
+        logger.info("writing final RO-crate")
+        to_rocrate(
+            datapath=outpath,
+            userid=job.payload.user.identifier,
+            sampleid=job.payload.sample.identifier,
+            make_child=job.payload.sample.sample_is_parent,
+        )
     logger.info("job finished with status '%s', updating job db", job.status)
     params = dict(status=job.status, completed_at=job.completed_at)
     job = jobdb.update_job_id(job.id, params, args.dbpath)
@@ -765,10 +791,10 @@ def job_main_loop(
     logger.debug("polling threads until completion")
     while True:
         tN = time.perf_counter()
-        if snapshot is not None and tN - tS > snapshot.snapshot_interval:
+        if snapshot is not None and tN - tS > snapshot.interval:
             logger.debug("creating snapshot")
             merge_netcdfs(job, snapshot=True)
-            tS += snapshot.snapshot_interval
+            tS += snapshot.interval
 
         # Collect and push task names
         for t in threads.values():

src/tomato/driverinterface_1_0/__init__.py

@@ -18,8 +18,7 @@
 from pydantic import BaseModel
 from threading import Thread, current_thread, RLock
 from queue import Queue
-from tomato.models import Reply
-from dgbowl_schemas.tomato.payload import Task
+from tomato.models import Reply, Task
 import logging
 from functools import wraps
 from xarray import Dataset

src/tomato/driverinterface_2_0/__init__.py

@@ -12,8 +12,7 @@
 from pydantic import BaseModel, Field
 from threading import Thread, current_thread, RLock
 from queue import Queue
-from tomato.models import Reply
-from dgbowl_schemas.tomato.payload import Task
+from tomato.models import Reply, Task
 import logging
 from functools import wraps
 import xarray as xr

src/tomato/driverinterface_2_1/__init__.py

@@ -13,10 +13,9 @@
 from threading import Thread, current_thread, RLock
 from collections import defaultdict
 import queue
-from tomato.models import Reply
+from tomato.models import Reply, Task
 from tomato.driverinterface_2_1.decorators import in_devmap, to_reply, log_errors
 from tomato.driverinterface_2_1.types import Type, Val, Key
-from dgbowl_schemas.tomato.payload import Task
 import logging
 
 import xarray as xr