ditching three-dicts implementation for annotation prop "magics"

mmif/serialize/__init__.py

@@ -6,7 +6,9 @@
 from .annotation import __all__ as anno_all
 from .mmif import *
 from .mmif import __all__ as mmif_all
+from .producer import *
+from .producer import __all__ as producer_all
 from .view import *
 from .view import __all__ as view_all
 
-__all__ = anno_all + mmif_all + view_all
+__all__ = anno_all + mmif_all + producer_all + view_all

mmif/serialize/model.py

@@ -23,19 +23,19 @@
 
 import json
 import warnings
-from datetime import datetime
-from typing import Union, Any, Dict, Optional, TypeVar, Generic, Generator, Iterator, Type, Set, ClassVar, List
+from collections.abc import Generator, Iterator
+from typing import Any, ClassVar, Generic, Optional, TypeVar, Union
 
 T = TypeVar('T')
 S = TypeVar('S')
-PRMTV_TYPES: Type = Union[str, int, float, bool, None]
+PRMTV_TYPES: type = Union[str, int, float, bool, None]  # noqa: UP007
 
 __all__ = [
-    'MmifObject',
-    'MmifObjectEncoder',
-    'DataList',
+    'PRMTV_TYPES',
     'DataDict',
-    'PRMTV_TYPES'
+    'DataList',
+    'MmifObject',
+    'MmifObjectEncoder'
 ]
 
 
@@ -96,7 +96,7 @@ class MmifObject(object):
 
     # these are the reserved names that cannot be used as attribute names, and 
     # they won't be serialized
-    reserved_names: Set[str] = {
+    reserved_names: set[str] = {
         'reserved_names',
         '_unnamed_attributes',
         '_attribute_classes',
@@ -105,9 +105,9 @@ class MmifObject(object):
         '_contextual_attributes'
     }
     _unnamed_attributes: Optional[dict]
-    _exclude_from_diff: Set[str]
-    _contextual_attributes: Set[str]
-    _attribute_classes: Dict[str, Type] = {}  # Mapping: str -> Type
+    _exclude_from_diff: set[str]
+    _contextual_attributes: set[str]
+    _attribute_classes: dict[str, type] = {}  # Mapping: str -> Type
 
     def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None:
         if isinstance(mmif_obj, bytes):
@@ -183,7 +183,7 @@ def serialize(self, pretty: bool = False, include_context: bool = True) -> str:
         """
         return json.dumps(self._serialize(include_context=include_context), indent=2 if pretty else None, cls=MmifObjectEncoder)
 
-    def _serialize(self, alt_container: Optional[Dict] = None, include_context: bool = True) -> dict:
+    def _serialize(self, alt_container: Optional[dict] = None, include_context: bool = True) -> dict:
         """
         Maps a MMIF object to a plain python dict object,
         rewriting internal keys that start with '_' to
@@ -259,7 +259,7 @@ def _load_json(json_obj: Union[dict, str]) -> dict:
         :param json_str: the JSON data to load and process
         :return: the mapped data as a dict
         """
-        def from_atsign(d: Dict[str, Any]) -> dict:
+        def from_atsign(d: dict[str, Any]) -> dict:
             for k in list(d.keys()):
                 if k.startswith('@'):
                     d[f'_{k[1:]}'] = d.pop(k)
@@ -298,9 +298,9 @@ def _deserialize(self, input_dict: dict) -> None:
         If a subclass needs a special treatment during the mapping, it needs to
         override this method.
 
-        This default method won't work for generic types (e.g. List[X], Dict[X, Y]).
+        This default method won't work for generic types (e.g. list[X], dict[X, Y]).
         For now, lists are abstracted as DataList and dicts are abstracted as XXXMetadata classes.
-        However, if an attribute uses a generic type (e.g. view_metadata.contains: Dict[str, Contain])
+        However, if an attribute uses a generic type (e.g. view_metadata.contains: dict[str, Contain])
         that class should override _deserialize of its own.
 
         :param input_dict: the prepared JSON data that defines the object
@@ -452,7 +452,7 @@ class DataList(MmifObject, Generic[T]):
     """
     def __init__(self, mmif_obj: Optional[Union[bytes, str, list]] = None, *_):
         self.reserved_names.add('_items')
-        self._items: Dict[str, T] = dict()
+        self._items: dict[str, T] = dict()
         self.disallow_additional_properties()
         if mmif_obj is None:
             mmif_obj = []
@@ -484,8 +484,8 @@ def _deserialize(self, input_list: list) -> None:
     def get(self, key: str, default=None) -> Optional[T]:
         """
         .. deprecated:: 1.1.3
-           Do not use in new code. Will be removed in 2.0.0.
-           Use container-level access or positional indexing instead.
+           Will be removed in 2.0.0. Use container-level access or positional
+           indexing instead.
 
         Deprecated method for retrieving list elements by string ID.
 
@@ -517,13 +517,11 @@ def get(self, key: str, default=None) -> Optional[T]:
         --------
         __getitem__ : List-style positional access with integers
         """
-        warnings.warn(
-            "The 'get' method on list-like collections is deprecated and "
-            "will be removed in 2.0.0. Use container-level access "
-            "(e.g., mmif['v1']) or positional indexing (e.g., views[0]).",
-            DeprecationWarning,
-            stacklevel=2
-        )
+        warnings.warn("get() on list-like collections is deprecated and will "
+                      "be removed in 2.0.0. Use container-level access (e.g.,"
+                      " mmif['v1']) or positional indexing (e.g., views[0]) "
+                      "instead.",
+                      DeprecationWarning, stacklevel=2)
         return self._items.get(key, default)
 
     def _append_with_key(self, key: str, value: T, overwrite=False) -> None:
@@ -549,7 +547,7 @@ def _append_with_key(self, key: str, value: T, overwrite=False) -> None:
     def append(self, value, overwrite):
         raise NotImplementedError()
 
-    def __getitem__(self, key: Union[int, slice]) -> Union[T, List[T]]:
+    def __getitem__(self, key: Union[int, slice]) -> Union[T, list[T]]:
         """
         List-style positional access using integers or slices.
 
@@ -615,7 +613,7 @@ def empty(self):
 class DataDict(MmifObject, Generic[T, S]):
     def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_):
         self.reserved_names.add('_items')
-        self._items: Dict[T, S] = dict()
+        self._items: dict[T, S] = dict()
         self.disallow_additional_properties()
         if mmif_obj is None:
             mmif_obj = {}

mmif/serialize/producer.py

@@ -0,0 +1,237 @@
+"""
+The :mod:`producer` module contains the producer context, the write mode of a
+MMIF. See :meth:`mmif.serialize.mmif.Mmif.set_producer`.
+"""
+
+import copy
+import warnings
+from collections import defaultdict
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Any, Optional, Union, cast
+
+from mmif.serialize.view import View
+from mmif.vocabulary import AnnotationTypes
+
+if TYPE_CHECKING:
+    from mmif.serialize.mmif import Mmif
+
+__all__ = ['EmptyOutputWarning', 'ProducerContext']
+
+
+class EmptyOutputWarning(UserWarning):
+    """
+    Issued when a run leaves no view: it produced no annotations, no error,
+    and no warnings.
+    """
+
+
+class ProducerContext:
+    """
+    The write mode of a MMIF, for one run of a producer. Holds what the run
+    needs to know: a snapshot taken when the producer is set, the signature,
+    and the warnings raised during the run.
+
+    While a producer is set, views that existed when it was set are serialized
+    as they are in memory, and :meth:`mmif.serialize.mmif.Mmif.serialize`
+    finalizes the views of the run in its output: capital ``Annotation``
+    records, factored properties, ``contains`` pruning, removal of empty
+    views, and one run ID. The ``Mmif`` object itself does not change, so
+    repeated serialization gives the same output.
+
+    Leaving the ``with`` block turns the mode off. An exception propagates
+    unchanged.
+
+    :param mmif: the MMIF the run works on
+    :param signature: a view whose metadata is signed by the producer, copied
+        to the views this context writes
+    :raises RuntimeError: if a producer is already set on the MMIF
+    """
+
+    def __init__(self, mmif: 'Mmif', signature: View) -> None:
+        if mmif._producer is not None:
+            raise RuntimeError('A producer is already set on this MMIF.')
+        self.mmif = mmif
+        self.signature = signature
+        self._warnings: list[Warning] = []
+        self._view_metadata: dict[str, Any] = {}
+        self._run_id: Optional[datetime] = None
+        self._existing_view_ids = {view.id for view in mmif.views}
+        self._entry_pending = {doc.id: dict(doc._props_pending.items())
+                               for doc in self._documents()}
+        mmif._producer = self
+
+    def __enter__(self) -> 'ProducerContext':
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_tb) -> bool:
+        self.mmif._producer = None
+        return False
+
+    def add_warnings(self,
+                     *issued: Union[Warning, warnings.WarningMessage]) -> None:
+        """
+        Records warnings of this run, written to one signed warnings view.
+
+        :param issued: warnings, as raised or as caught by
+            ``warnings.catch_warnings(record=True)``
+        """
+        for warning in issued:
+            if isinstance(warning, warnings.WarningMessage):
+                # `warnings.warn()` turns a str message into a Warning
+                warning = cast(Warning, warning.message)
+            self._warnings.append(warning)
+
+    def set_view_metadata(self, key: str, value: Any) -> None:
+        """
+        Sets an additional metadata property on every view of this run,
+        including the views this context writes (e.g., ``appProfiling``).
+
+        :param key: the metadata key
+        :param value: the value
+        """
+        self._view_metadata[key] = value
+
+    def _finalize_output(self, views_out: list[dict],
+                         include_context: bool) -> None:
+        """
+        Finalizes the views of this run in a serialized MMIF. Earlier views
+        are not touched.
+
+        :param views_out: serialized views, changed in place
+        :param include_context: See :meth:`mmif.serialize.mmif.Mmif.serialize`.
+        """
+        if self._run_id is None:
+            self._run_id = self._find_run_id()
+        plan = self._plan_capital_annotations(views_out, include_context)
+        self.mmif._write_capital_annotations(plan, views_out, include_context)
+        run_views_out = [view_out for view_out in views_out
+                         if view_out['id'] not in self._existing_view_ids]
+        self.mmif._factor_out_in_output(run_views_out)
+        self.mmif._prune_contains_in_output(run_views_out)
+
+        empty_ids = {view_out['id'] for view_out in run_views_out
+                     if self._is_empty(view_out)}
+        views_out[:] = [view_out for view_out in views_out
+                        if view_out['id'] not in empty_ids]
+        run_views_out = [view_out for view_out in run_views_out
+                         if view_out['id'] not in empty_ids]
+
+        run_warnings = list(self._warnings)
+        if not run_views_out:
+            run_warnings.append(EmptyOutputWarning(
+                f'App "{self.signature.metadata.app}" produced no annotations '
+                f'for this input; the input may lack expected content, or '
+                f'there may genuinely be nothing to annotate.'))
+        if run_warnings:
+            warnings_view = self._new_signed_view(views_out)
+            warnings_view.metadata.add_warnings(*run_warnings)
+            warnings_out = warnings_view._serialize(
+                include_context=include_context)
+            views_out.append(warnings_out)
+            run_views_out.append(warnings_out)
+
+        for view_out in run_views_out:
+            view_out['metadata'].update(self._view_metadata)
+            if include_context:
+                view_out['metadata']['timestamp'] = self._run_id
+
+    def _find_run_id(self) -> datetime:
+        """
+        Finds the run ID: the timestamp of the first view of this run, which
+        ``Mmif.new_view()`` assigned. A run that created no view gets the
+        current time.
+
+        :return: the run ID
+        """
+        return next((view.metadata.timestamp for view in self.mmif.views
+                     if view.id not in self._existing_view_ids
+                     and view.metadata.timestamp is not None),
+                    datetime.now(timezone.utc))
+
+    def _plan_capital_annotations(
+            self, views_out: list[dict], include_context: bool,
+    ) -> list[tuple[View, str, dict, bool]]:
+        """
+        Decides where the pending properties of this run are recorded: those
+        that weren't pending on entry and that no view already records. A
+        document created in this run takes them itself, and the others get
+        records in the last view of this run. A signed view is added to the
+        output when the run created no view to hold the records.
+
+        :param views_out: serialized views, changed when a view is added
+        :param include_context: See :meth:`mmif.serialize.mmif.Mmif.serialize`.
+        :return: the plan, in the form that
+            :meth:`mmif.serialize.mmif.Mmif._write_capital_annotations` takes
+        """
+        recorded = defaultdict(dict)
+        for view in self.mmif.views:
+            for ann in view.get_annotations(AnnotationTypes.Annotation):
+                recorded[ann.get('document')].update(
+                    ann._resolved_properties())
+
+        run_views = [view for view in self.mmif.views
+                     if view.id not in self._existing_view_ids]
+        doc_views = {doc.id: view for view in run_views
+                     for doc in view.get_documents()}
+        target = next((view for view in reversed(run_views)
+                       if not view.has_error() and not view.has_warnings()),
+                      None)
+
+        plan = []
+        for doc in self._documents():
+            on_entry = self._entry_pending.get(doc.id, {})
+            props = {key: value for key, value in doc._props_pending.items()
+                     if key != 'id'
+                     and not (key in on_entry and on_entry[key] == value)
+                     and not (key in recorded[doc.id]
+                              and recorded[doc.id][key] == value)}
+            if not props:
+                continue
+            if doc.id in doc_views:
+                plan.append((doc_views[doc.id], doc.id, props, True))
+                continue
+            if target is None:
+                target = self._new_signed_view(views_out)
+                views_out.append(
+                    target._serialize(include_context=include_context))
+            plan.append((target, doc.id, props, False))
+        return plan
+
+    def _documents(self):
+        """
+        Iterates over all documents, in the top-level list and in the views.
+        """
+        yield from self.mmif.documents
+        for view in self.mmif.views:
+            yield from view.get_documents()
+
+    def _new_signed_view(self, views_out: list[dict]) -> View:
+        """
+        Creates a view for the output only, with an ID that is free in the
+        output and in the object, and with the metadata of the signature.
+
+        :param views_out: serialized views
+        :return: the new view, not added to the MMIF
+        """
+        used_ids = {view_out['id'] for view_out in views_out}
+        used_ids.update(view.id for view in self.mmif.views)
+        index = len(used_ids)
+        while f'{self.mmif.view_prefix}{index}' in used_ids:
+            index += 1
+        view = View()
+        view.id = f'{self.mmif.view_prefix}{index}'
+        view.metadata = copy.deepcopy(self.signature.metadata)
+        return view
+
+    @staticmethod
+    def _is_empty(view_out: dict) -> bool:
+        """
+        Checks that a serialized view has no annotations, no error, and no
+        warnings.
+
+        :param view_out: a serialized view
+        :return: True for an empty view
+        """
+        metadata = view_out['metadata']
+        return not (view_out['annotations'] or metadata.get('error')
+                    or metadata.get('warnings'))