Add Measurement class

src/scippneutron/meta/__init__.py

@@ -1,3 +1,5 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
 """Metadata utilities.
 
 .. rubric:: Models
@@ -7,6 +9,7 @@
   :template: model-template.rst
 
   Beamline
+  Measurement
   Person
   Software
   Source
@@ -43,6 +46,7 @@
     Person,
     Software,
     Source,
+    Measurement,
     PulseDuration,
     SourceFrequency,
     SourcePeriod,
@@ -54,6 +58,7 @@
 
 __all__ = [
     'Beamline',
+    'Measurement',
     'Person',
     'ORCIDiD',
     'Software',

src/scippneutron/meta/_model.py

@@ -1,14 +1,155 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
 from __future__ import annotations
 
 import enum
+from datetime import datetime
 from typing import NewType
 
 import scipp as sc
+import scippnexus as snx
+from dateutil.parser import parse as parse_datetime
 from pydantic import BaseModel, ConfigDict, EmailStr
 
 from ._orcid import ORCIDiD
 
 
+class Beamline(BaseModel):
+    """A beamline / instrument.
+
+    ``name`` should be the canonical spelling of the beamline name.
+    The location of the beamline is split into ``facility`` and ``site``, where
+    a 'facility' is located at a 'site'. For example:
+
+    >>> Beamline(
+    ...     name='Amor',
+    ...     facility='SINQ',
+    ...     site='PSI',
+    ... )
+
+    If there is no separate facility and site, omit ``site``:
+
+    >>> Beamline(
+    ...     name='ESTIA',
+    ...     facility='ESS',
+    ...     site=None,  # can be omitted
+    ... )
+
+    If the beamline has been upgraded, provide a revision to indicate
+    which version of the beamline was used.
+    """
+
+    name: str
+    """Name of the beamline."""
+    facility: str | None = None
+    """Facility where the beamline is located."""
+    site: str | None = None
+    """Site where the facility is located."""
+    revision: str | None = None
+    """Revision of the beamline in case of upgrades."""
+
+    @classmethod
+    def from_nexus_entry(
+        cls, entry: snx.Group, *, instrument_name: str | None = None
+    ) -> Beamline:
+        """Construct a Beamline object from a Nexus entry.
+
+        NeXus does not have a standard method for specifying the facility, site, or
+        revision. This function only sets those fields for known instruments.
+
+        Parameters
+        ----------
+        entry:
+            ScippNexus group for a NeXus entry.
+            The entry needs to contain an ``NXinstrument`` with a 'name' field
+            to identify the instrument.
+        instrument_name:
+            If the entry contains more than one ``NXinstrument`` group, this parameter
+            must be the name of one of these groups.
+
+        Returns
+        -------
+        :
+            A Beamline object constructed from the given Nexus entry.
+        """
+        instrument = _get_unique_nexus_child(entry, snx.NXinstrument, instrument_name)
+        instrument_name = _read_optional_nexus_string(instrument, 'name')
+        if instrument_name is None:
+            raise ValueError("No instrument name found in Nexus entry")
+
+        facility, site = _guess_facility_and_site(instrument_name)
+
+        return cls(
+            name=instrument_name,
+            facility=facility,
+            site=site,
+        )
+
+
+class Measurement(BaseModel):
+    """A single measurement.
+
+    Terminology:
+
+        - An "experiment" is the collection of all related measurements,
+          typically done during one beamtime.
+        - A "measurement" is a single step of data collection.
+          It typically corresponds to a single NeXus file or a single entry
+          in a NeXus file.
+
+    The ``Measurement`` class represents a single measurement but also includes some
+    information about the experiment that this measurement is part of.
+    In particular, ``experiment_id`` and ``experiment_doi`` encode
+    information about the experiment.
+    *All* other fields encode information about a measurement; this includes
+    ``start_time`` and ``end_time``.
+    """
+
+    title: str | None
+    """The title of the measurement."""
+    run_number: str | None = None
+    """Run number of the measurement."""
+    experiment_id: str | None = None
+    """An ID for the experiment that this measurement is part of, e.g., proposal ID."""
+    experiment_doi: str | None = None
+    """A DOI for the experiment that this measurement is part of."""
+    start_time: datetime | None = None
+    """Date and time when the measurement started."""
+    end_time: datetime | None = None
+    """Date and time when the measurement ended."""
+
+    @classmethod
+    def from_nexus_entry(cls, entry: snx.Group) -> Measurement:
+        """Construct a Measurement object from a Nexus entry.
+
+        Parameters
+        ----------
+        entry:
+            ScippNexus group for a NeXus entry.
+
+        Returns
+        -------
+        :
+            An Measurement object constructed from the given Nexus entry.
+        """
+        return cls(
+            title=_read_optional_nexus_string(entry, 'title'),
+            run_number=_read_optional_nexus_string(entry, 'entry_identifier'),
+            experiment_id=_read_optional_nexus_string(entry, 'experiment_identifier'),
+            start_time=_read_optional_nexus_datetime(entry, 'start_time'),
+            end_time=_read_optional_nexus_datetime(entry, 'end_time'),
+            experiment_doi=None,
+        )
+
+    @property
+    def run_number_maybe_int(self) -> int | str | None:
+        """Return the run number as an int if possible."""
+        try:
+            return int(self.run_number)
+        except ValueError:
+            return self.run_number
+
+
 class Person(BaseModel):
     """A person.
 
@@ -49,41 +190,6 @@ class Person(BaseModel):
     """Affiliation of the person."""
 
 
-class Beamline(BaseModel):
-    """A beamline / instrument.
-
-    ``name`` should be the canonical spelling of the beamline name.
-    The location of the beamline is split into ``facility`` and ``site``, where
-    a 'facility' is located at a 'site'. For example:
-
-    >>> Beamline(
-    ...     name='Amor',
-    ...     facility='SINQ',
-    ...     site='PSI',
-    ... )
-
-    If there is no separate facility and site, omit ``site``:
-
-    >>> Beamline(
-    ...     name='ESTIA',
-    ...     facility='ESS',
-    ...     site=None,  # can be omitted
-    ... )
-
-    If the beamline has been upgraded, provide a revision to indicate
-    which version of the beamline was used.
-    """
-
-    name: str
-    """Name of the beamline."""
-    facility: str | None = None
-    """Facility where the beamline is located."""
-    site: str | None = None
-    """Site where the facility is located."""
-    revision: str | None = None
-    """Revision of the beamline in case of upgrades."""
-
-
 class Software(BaseModel):
     """A piece of software.
 
@@ -117,14 +223,46 @@ class Software(BaseModel):
     version: str
     """Complete version of the piece of software."""
     url: str | None = None
-    """URL to the concrete version of the software."""
+    """URL to the concrete version of the software.
+
+    If no URL for a concrete version is available,
+    a URL of the project or source code may be used.
+    """
     doi: str | None = None
     """DOI of the concrete version of the software.
 
     If there is no DOI for the concrete version,
     a general DOI for the software may be used.
     """
 
+    @classmethod
+    def from_package_metadata(cls, package_name: str) -> Software:
+        """Construct a Software instance from the metadata of an installed package.
+
+        This function attempts to deduce all information it can from package metadata.
+        But it only has access to the information that is encoded in the package.
+        It therefore returns the base project URL instead of a concrete release URL,
+        and it does not return a DOI.
+
+        Parameters
+        ----------
+        package_name:
+            The name of the Python package.
+
+        Returns
+        -------
+        :
+            A Software instance.
+        """
+        from importlib.metadata import version
+
+        return cls(
+            name=package_name,
+            version=version(package_name),
+            url=_deduce_package_source_url(package_name),
+            doi=None,
+        )
+
     @property
     def name_version(self) -> str:
         """The name and version of the software, separated by a space."""
@@ -138,6 +276,20 @@ def compact_repr(self) -> str:
         return self.name_version
 
 
+def _deduce_package_source_url(package_name: str) -> str | None:
+    from importlib.metadata import metadata
+
+    if not (urls := metadata(package_name).get_all("project-url")):
+        return None
+
+    try:
+        return next(
+            url.split(',')[-1].strip() for url in urls if url.startswith("Source")
+        )
+    except StopIteration:
+        return None
+
+
 PulseDuration = NewType('PulseDuration', sc.Variable)
 PulseDuration.__doc__ = """Duration of a source pulse."""
 SourceFrequency = NewType('SourceFrequency', sc.Variable)
@@ -207,3 +359,74 @@ def to_pipeline_params(self) -> dict[type, object]:
     probe=RadiationProbe.Neutron,
 )
 ESS_SOURCE.__doc__ = """Default parameters of the ESS source."""
+
+
+def _read_optional_nexus_string(group: snx.Group | None, key: str) -> str | None:
+    if group is None:
+        return None
+    if (ds := group.get(key)) is not None:
+        return ds[()]
+    return None
+
+
+def _read_optional_nexus_datetime(group: snx.Group | None, key: str) -> datetime | None:
+    if (s := _read_optional_nexus_string(group, key)) is not None:
+        return parse_datetime(s)
+    return None
+
+
+def _get_unique_nexus_child(
+    entry: snx.Group, nx_class: type, name: str | None
+) -> snx.Group | None:
+    if name is not None:
+        return entry.get(name)
+    children = entry[nx_class]
+    if len(children) > 1:
+        raise RuntimeError(
+            f"Got multiple {nx_class.__name__} in NeXus entry '{entry.name}'"
+        )
+    if len(children) == 0:
+        return None
+    return next(iter(children.values()))
+
+
+# More instruments may be added as needed.
+# All instrument names are lowercase.
+_FACILITY_PER_INSTRUMENT: dict[str, str | tuple[str, str]] = {
+    # ESS
+    'beer': 'ESS',
+    'bifrost': 'ESS',
+    'cspec': 'ESS',
+    'dream': 'ESS',
+    'estia': 'ESS',
+    'freia': 'ESS',
+    'heimdal': 'ESS',
+    'loki': 'ESS',
+    'magic': 'ESS',
+    'miracles': 'ESS',
+    'nmx': 'ESS',
+    'odin': 'ESS',
+    'skadi': 'ESS',
+    'tbl': 'ESS',
+    'trex': 'ESS',
+    'vespa': 'ESS',
+    # SINQ
+    'amor': ('SINQ', 'PSI'),
+}
+
+
+# NeXus provides no way to specify the facility.
+# But we can usually guess it based on the instrument name.
+def _guess_facility_and_site(
+    instrument_name: str | None,
+) -> tuple[str | None, str | None]:
+    if instrument_name is None:
+        return None, None
+
+    match _FACILITY_PER_INSTRUMENT.get(instrument_name.lower()):
+        case None:
+            return None, None
+        case (facility, site):
+            return facility, site
+        case facility:
+            return facility, None

src/scippneutron/meta/_orcid.py

@@ -1,3 +1,5 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
 from __future__ import annotations
 
 import itertools