Add TaskMetadata.from_metadata to convert PropertySet to TaskMetadata

python/lsst/pipe/base/_task_metadata.py

@@ -27,7 +27,7 @@
 from collections.abc import Sequence
 from deprecated.sphinx import deprecated
 
-from typing import Dict, List, Union, Any, Mapping
+from typing import Dict, List, Union, Any, Mapping, Protocol, Collection
 from pydantic import BaseModel, StrictInt, StrictFloat, StrictBool, StrictStr, Field
 
 _DEPRECATION_REASON = "Will be removed after v25."
@@ -38,6 +38,20 @@
 _ALLOWED_PRIMITIVE_TYPES = (str, float, int, bool)
 
 
+class PropertySetLike(Protocol):
+    """Protocol that looks like a ``lsst.daf.base.PropertySet``
+
+    Enough of the API is specified to support conversion of a
+    ``PropertySet`` to a `TaskMetadata`.
+    """
+
+    def paramNames(self, topLevelOnly: bool = True) -> Collection[str]:
+        ...
+
+    def getArray(self, name: str) -> Any:
+        ...
+
+
 def _isListLike(v):
     return isinstance(v, Sequence) and not isinstance(v, str)
 
@@ -83,6 +97,41 @@ def from_dict(cls, d: Mapping[str, Any]) -> "TaskMetadata":
             metadata[k] = v
         return metadata
 
+    @classmethod
+    def from_metadata(cls, ps: PropertySetLike) -> "TaskMetadata":
+        """Create a TaskMetadata from a PropertySet-like object.
+
+        Parameters
+        ----------
+        ps : `lsst.daf.base.PropertySet` or `TaskMetadata`
+            A ``PropertySet``-like object to be transformed to a
+            `TaskMetadata`. A `TaskMetadata` can be copied using this
+            class method.
+
+        Returns
+        -------
+        tm : `TaskMetadata`
+            Newly-constructed metadata.
+
+        Notes
+        -----
+        Items stored in single-element arrays in the supplied object
+        will be converted to scalars in the newly-created object.
+        """
+        # Use hierarchical names to assign values from input to output.
+        # This API exists for both PropertySet and TaskMetadata.
+        # from_dict() does not work because PropertySet is not declared
+        # to be a Mapping.
+        # PropertySet.toDict() is not present in TaskMetadata so is best
+        # avoided.
+        metadata = cls()
+        for key in sorted(ps.paramNames(topLevelOnly=False)):
+            value = ps.getArray(key)
+            if len(value) == 1:
+                value = value[0]
+            metadata[key] = value
+        return metadata
+
     def add(self, name, value):
         """Store a new value, adding to a list if one already exists.
 

tests/test_taskmetadata.py

@@ -185,6 +185,22 @@ def testDict(self):
         meta2 = TaskMetadata.parse_obj(json.loads(j))
         self.assertEqual(meta2, meta)
 
+        # Round trip.
+        meta3 = TaskMetadata.from_metadata(meta)
+        self.assertEqual(meta3, meta)
+
+        # Add a new element that would be a single-element array.
+        # This will not equate as equal because from_metadata will move
+        # the item to the scalar part of the model and pydantic does not
+        # see them as equal.
+        meta3.add("e.new", 5)
+        meta4 = TaskMetadata.from_metadata(meta3)
+        self.assertNotEqual(meta4, meta3)
+        self.assertEqual(meta4["e.new"], meta3["e.new"])
+        del meta4["e.new"]
+        del meta3["e.new"]
+        self.assertEqual(meta4, meta3)
+
     def testDeprecated(self):
         """Test the deprecated interface issues warnings."""
         meta = TaskMetadata.from_dict({"a": 1, "b": 2})