diff --git a/mmif/serialize/annotation.py b/mmif/serialize/annotation.py index 1e5e905b..e0bd9072 100644 --- a/mmif/serialize/annotation.py +++ b/mmif/serialize/annotation.py @@ -6,6 +6,7 @@ of a view. For documentation on how views are represented, see :mod:`mmif.serialize.view`. """ +import functools import importlib import itertools import os @@ -22,7 +23,7 @@ from .. import DocumentTypes, AnnotationTypes import mmif_docloc_http -__all__ = ['Annotation', 'AnnotationProperties', 'Document', 'DocumentProperties', 'Text'] +__all__ = ['Annotation', 'AnnotationProperties', 'Document', 'DocumentProperties', 'Text', 'canonicalize_prop_name'] T = TypeVar('T') LIST_PRMTV = typing.List[PRMTV_TYPES] # list of values (most cases for annotation props) @@ -41,6 +42,40 @@ }) +@functools.lru_cache(maxsize=None) +def _prop_alias_map(shortname: str) -> Dict[str, str]: + """ + Maps every spelling of an aliased property of a type to the canonical + name. The table is read from the latest version of the type in the + vocabulary (``AnnotationTypes.``), which is the single source + of truth for property aliases. + + :param shortname: shortname of the annotation type (e.g. ``TimeFrame``) + :return: ``{spelling: canonical_name}``, empty for types without aliases + """ + latest = getattr(AnnotationTypes, shortname, None) + groups = getattr(latest, '_property_aliases', None) or {} + return {alias: canonical for canonical, group in groups.items() for alias in group} + + +def canonicalize_prop_name(at_type: ThingTypesBase, prop_name: str) -> str: + """ + Resolves a property name to its canonical spelling for the given type. + Property aliases were introduced in MMIF 1.0.2, when the general + ``label`` property replaced ``frameType`` (``TimeFrame``) and ``boxType`` + (``BoundingBox``). Names that are not aliases, and types without an alias + table, are returned unchanged. + + :param at_type: the ``@type`` of the annotation + :param prop_name: a property name, canonical or deprecated + :return: the canonical property name + """ + shortname = getattr(at_type, 'shortname', None) + if not shortname: + return prop_name + return _prop_alias_map(shortname).get(prop_name, prop_name) + + class Annotation(MmifObject): """ MmifObject that represents an annotation in a MMIF view. @@ -49,11 +84,11 @@ class Annotation(MmifObject): def __init__(self, anno_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: self._type: ThingTypesBase = ThingTypesBase('') # to store the parent view ID - self._props_ephemeral: AnnotationProperties = AnnotationProperties() + self._props_ephemeral: AnnotationProperties = AnnotationProperties(None, self) self._alignments = {} # to hold alignment information (Alignment anno long_id -> aligned anno long_id) self.reserved_names.update(('_props_ephemeral', '_alignments')) if not hasattr(self, 'properties'): # don't overwrite DocumentProperties on super() call - self.properties: AnnotationProperties = AnnotationProperties() + self.properties: AnnotationProperties = AnnotationProperties(None, self) self._attribute_classes = {'properties': AnnotationProperties} self.disallow_additional_properties() self._required_attributes = ["_type", "properties"] @@ -69,9 +104,7 @@ def _deserialize(self, input_dict: dict) -> None: # here is the place to parse formatted IDs and store prefixes in the parent mmif object. # (see https://github.com/clamsproject/mmif/issues/64#issuecomment-849241309 for discussion) super()._deserialize(input_dict) - for k, v in self.properties.items(): - self._add_prop_aliases(k, v) - + def _cache_alignment(self, alignment_ann: 'Annotation', alignedto_ann: 'Annotation') -> None: """ Cache alignment information. This cache will not be serialized. @@ -107,29 +140,20 @@ def get_all_aligned(self) -> Iterator['Annotation']: yield aligned - def _add_prop_aliases(self, key_to_add, val_to_add): - """ - Method to handle aliases of the same property. - Annotation property aliases were first introduced in MMIF 1.0.2, - with addition of general `label` property to all `Annotation` - subtypes, and effectively deprecated `frameType` and `boxType` - in `TimeFrame` and `BoundingBox` respectively. - """ - prop_aliases = AnnotationTypes._prop_aliases.get(self._type.shortname, {}) - for alias_rep, alias_group in prop_aliases.items(): - if key_to_add in alias_group: - for alias in alias_group: - if alias != key_to_add: - self._props_ephemeral[alias] = val_to_add - if alias in self.properties.keys(): - warning_msg = (f'Both "{key_to_add}" and "{alias}" are in the properties of "{self.id}", ' - f'however ') - if alias == alias_rep: - warning_msg += f'"{key_to_add}" is an alias of "{alias_rep}".' - else: - warning_msg += f'"{key_to_add}" and "{alias}" are both aliases of "{alias_rep}".' - warning_msg += f'Having two synonyms in the same annotation can cause unexpected behavior. ' - warnings.warn(warning_msg, UserWarning) + def _add_ephemeral_defaults(self, defaults: typing.Mapping[str, typing.Any]) -> None: + """ + Distributes view-level defaults (from the ``contains`` metadata of the + parent view) to this annotation as ephemeral properties, under their + canonical names. An annotation-level value takes precedence over a + view-level default, so a default is skipped when the annotation + carries the property itself. + + :param defaults: property defaults for this annotation's ``@type`` + """ + for name, value in defaults.items(): + name = canonicalize_prop_name(self._type, name) + if name not in self.properties: + self._props_ephemeral[name] = value def is_type(self, at_type: Union[str, ThingTypesBase]) -> bool: """ @@ -239,7 +263,9 @@ def json_primitives_list_of_list(x): def add_property(self, name: str, value: Union[PRMTV_TYPES, LIST_PRMTV, LIST_LIST_PRMTV, DICT_PRMTV, DICT_LIST_PRMTV]) -> None: """ - Adds a property to the annotation's properties. + Adds a property to the annotation's properties. A deprecated spelling + of an aliased property (e.g. ``frameType``) is stored under its + canonical name (``label``), with a :class:`DeprecationWarning`. :param name: the name of the property :param value: the property's desired value @@ -252,14 +278,14 @@ def add_property(self, name: str, # "either string, number, boolean, None, a JSON array of them, " # "or a JSON object of them keyed by strings." # f"(\"{name}\": \"{str(value)}\"") - self._add_prop_aliases(name, value) def __getitem__(self, prop_name: str): if prop_name in {'at_type', '@type'}: return str(self._type) elif prop_name == 'properties': return self.properties - elif prop_name in self.properties: + prop_name = canonicalize_prop_name(self._type, prop_name) + if prop_name in self.properties: return self.properties[prop_name] elif prop_name in self._props_ephemeral: return self._props_ephemeral[prop_name] @@ -279,7 +305,9 @@ def get(self, prop_name: str, default=None): 3. Special fields (``@type``, ``properties``) This allows convenient access to properties without explicitly - checking the ``properties`` object or view-level metadata. + checking the ``properties`` object or view-level metadata. A + deprecated spelling of an aliased property (e.g. ``frameType``) + resolves to the canonical property (``label``). :param prop_name: The name of the property to retrieve :param default: The value to return if the property is not found (default: None) @@ -318,23 +346,6 @@ def __contains__(self, item): except KeyError: return False - def _get_label(self) -> str: - """ - Another prototypical method to handle property aliases. - See :meth:`.Annotation._add_prop_aliases` for more details on - what property aliases are. - Not recommended to use this method as `_add_prop_aliases` method - is preferred. - """ - if 'label' in self: - return str(self.get('label')) - elif self._type.shortname == 'TimeFrame' and 'frameType' in self: - return str(self.get('frameType')) - elif self._type.shortname == 'BoundingBox' and 'boxType' in self: - return str(self.get('boxType')) - else: - raise KeyError("No label found in this annotation.") - def is_document(self): return isinstance(self._type, DocumentTypesBase) @@ -356,8 +367,8 @@ def __init__(self, doc_obj: Optional[Union[bytes, str, dict]] = None, *_) -> Non # see https://github.com/clamsproject/mmif-python/issues/226 for discussion # around the use of these three dictionaries # (names changed since, `existing` >> `ephemeral` and `temporary` >> `pending`) - self._props_original: DocumentProperties = DocumentProperties() - self._props_pending: AnnotationProperties = AnnotationProperties() + self._props_original: DocumentProperties = DocumentProperties(None, self) + self._props_pending: AnnotationProperties = AnnotationProperties(None, self) self.reserved_names.update(('_props_original', '_props_pending')) self._type: Union[ThingTypesBase, DocumentTypesBase] = ThingTypesBase('') @@ -578,6 +589,7 @@ class AnnotationProperties(MmifObject, MutableMapping[str, T]): """ def __delitem__(self, key: str) -> None: + key = self._canonical(key) frm = None for k in self.__iter__(): if k == key: @@ -593,36 +605,83 @@ def __delitem__(self, key: str) -> None: del frm[key] else: raise KeyError(f'Key "{key}" not found.') - + def __iter__(self) -> Iterator[str]: """ - ``__iter__`` on Mapping should basically work as ``keys()`` method + ``__iter__`` on Mapping should basically work as ``keys()`` method of vanilla dict. """ for key in itertools.chain(self._named_attributes(), self._unnamed_attributes): yield key - + def __getitem__(self, key): """ - Parent MmifObject class has a __getitem__ method that checks if - the value is empty when asked for an unnamed attribute. But for - AnnotationProperties, any arbitrary property that's added - explicitly by the user (developer) should not be ignored and - returned even the value is empty. + Parent MmifObject class has a __getitem__ method that checks if + the value is empty when asked for an unnamed attribute. But for + AnnotationProperties, any arbitrary property that's added + explicitly by the user (developer) should not be ignored and + returned even the value is empty. """ + key = self._canonical(key) if key in self._named_attributes(): return self.__dict__[key] else: return self._unnamed_attributes[key] - def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __setitem__(self, key: str, value) -> None: + """ + Sets a property. A deprecated spelling of an aliased property (e.g. + ``frameType``) is stored under its canonical name (``label``), with + a :class:`DeprecationWarning`. + """ + canonical = self._canonical(key) + if canonical != key: + warnings.warn(f'"{key}" is a deprecated alias of "{canonical}", storing the value as "{canonical}".', + DeprecationWarning) + super().__setitem__(canonical, value) + + def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, + parent: Optional['Annotation'] = None, *_) -> None: self.id: str = '' # any individual at_type (subclassing this class) can have its own set of required attributes self._required_attributes = ["id"] # allowing additional attributes for arbitrary annotation properties self._unnamed_attributes = {} + # the Annotation these properties belong to; its `@type` decides which + # property names are aliases of each other (see `canonicalize_prop_name`) + self._parent_annotation = parent + self.reserved_names.add('_parent_annotation') super().__init__(mmif_obj) + def _canonical(self, key: str) -> str: + if self._parent_annotation is None: + return key + return canonicalize_prop_name(self._parent_annotation._type, key) + + def _deserialize(self, input_dict: dict) -> None: + """ + Deserializes the properties, rewriting deprecated spellings to their + canonical names so that there is one key per property. When two + spellings of the same property are present, the canonical one wins + (or the first one seen, when neither is canonical), with a warning. + """ + canonical_dict = {} + # canonical name -> the deprecated spelling it was read from + from_alias = {} + for key, value in input_dict.items(): + canonical = self._canonical(key) + if canonical in canonical_dict and (canonical != key or canonical in from_alias): + alias = key if canonical != key else from_alias[canonical] + warnings.warn(f'Both "{alias}" and "{canonical}" are in the properties of ' + f'"{input_dict.get("id", "")}", but "{alias}" is an alias of "{canonical}". ' + f'Keeping the value of "{canonical}".', UserWarning) + if canonical != key: + continue + canonical_dict[canonical] = value + if canonical != key: + from_alias[canonical] = key + super()._deserialize(canonical_dict) + class DocumentProperties(AnnotationProperties): """ @@ -632,7 +691,8 @@ class DocumentProperties(AnnotationProperties): :param mmif_obj: the JSON data that defines the properties """ - def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, + parent: Optional['Annotation'] = None, *_) -> None: self.mime: str = '' # note the trailing underscore here. I wanted to use the name `location` # for @property in this class and `Document` class, so had to use a diff @@ -643,9 +703,9 @@ def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_) -> No self._attribute_classes = {'text': Text} # in theory, either `location` or `text` should appear in a `document` # but with current implementation, there's no easy way to set a condition - # for `oneOf` requirement - # see MmifObject::_required_attributes in model.py - super().__init__(mmif_obj) + # for `oneOf` requirement + # see MmifObject::_required_attributes in model.py + super().__init__(mmif_obj, parent) def _deserialize(self, input_dict: dict) -> None: if "location" in input_dict: diff --git a/mmif/serialize/mmif.py b/mmif/serialize/mmif.py index 802e7d77..08d1ecf1 100644 --- a/mmif/serialize/mmif.py +++ b/mmif/serialize/mmif.py @@ -276,25 +276,11 @@ def _deserialize(self, input_dict: dict) -> None: super()._deserialize(input_dict) for view in self.views: view._parent_mmif = self - # this dict will be populated with properties - # that are not encoded in individual annotations objects themselves - extrinsic_props = defaultdict(dict) - for at_type, type_lv_props in view.metadata.contains.items(): - for prop_key, prop_value in type_lv_props.items(): - extrinsic_props[at_type][prop_key] = prop_value for ann in view.get_annotations(): ## for "capital" Annotation properties - # first add all extrinsic properties to the Annotation objects - # as "ephemeral" properties. A view-level `contains` default must - # not override an annotation-level value that is already present - # in `_props_ephemeral` -- namely an alias of a property set on the - # annotation itself (put there by `_add_prop_aliases` during the - # annotation's own deserialization, which runs earlier). Per the - # spec, the annotation-level value takes precedence over the - # view-level default. - for prop_key, prop_value in extrinsic_props[ann.at_type].items(): - if prop_key not in ann._props_ephemeral: - ann._props_ephemeral[prop_key] = prop_value + # first, distribute view-level defaults from `contains` metadata + # to the Annotation objects as "ephemeral" properties + ann._add_ephemeral_defaults(view.metadata.contains.get(ann.at_type, {})) # then, do the same to associated Document objects. Note that, # in a view, it is guaranteed that all Annotation objects are not duplicates if ann.at_type == AnnotationTypes.Annotation: diff --git a/mmif/serialize/view.py b/mmif/serialize/view.py index 6a99fa49..d72a519b 100644 --- a/mmif/serialize/view.py +++ b/mmif/serialize/view.py @@ -17,7 +17,7 @@ from typing import Dict, Union, Optional, Generator, List, cast from mmif import DocumentTypes, AnnotationTypes, ThingTypesBase, ClamsTypesBase -from mmif.serialize.annotation import Annotation, Document +from mmif.serialize.annotation import Annotation, Document, canonicalize_prop_name from mmif.serialize.model import PRMTV_TYPES, MmifObject, DataList, DataDict from mmif.vocabulary.base_types import AnnotationTypesBase @@ -197,8 +197,6 @@ def new_annotation(self, at_type: Union[str, ThingTypesBase], aid: Optional[str] self._set_ann_id(new_annotation, aid) for propk, propv in properties.items(): new_annotation.add_property(propk, propv) - for propk, propv in self.metadata.contains.get(at_type, {}).items(): - new_annotation._props_ephemeral[propk] = propv return self.add_annotation(new_annotation, overwrite) def add_annotation(self, annotation: 'Annotation', overwrite=False) -> 'Annotation': @@ -222,6 +220,7 @@ def add_annotation(self, annotation: 'Annotation', overwrite=False) -> 'Annotati annotation.id = f"{self.id}{self.id_delimiter}{annotation.id}" self.annotations.append(annotation, overwrite) self.new_contain(annotation.at_type) + annotation._add_ephemeral_defaults(self.metadata.contains.get(annotation.at_type, {})) if annotation.at_type == AnnotationTypes.Alignment: self._parent_mmif._cache_alignment(annotation) return annotation @@ -280,13 +279,14 @@ def get_annotations(self, at_type: Optional[Union[str, ThingTypesBase]] = None, must match. Note that annotation type metadata are specified in the `contains` view metadata, not in individual \ annotation objects. """ - def prop_check(k, v, *props): - return any(k in prop and prop[k] == v for prop in props) + def prop_check(k, v, annotation, at_type_metadata): + k = canonicalize_prop_name(annotation.at_type, k) + return any(k in prop and prop[k] == v for prop in (annotation.properties, at_type_metadata)) for annotation in self.annotations: at_type_metadata = self.metadata.contains.get(annotation.at_type, {}) if not at_type or (at_type and annotation.at_type == at_type): - if all(map(lambda kv: prop_check(kv[0], kv[1], annotation.properties, at_type_metadata), + if all(map(lambda kv: prop_check(kv[0], kv[1], annotation, at_type_metadata), properties.items())): yield annotation @@ -485,7 +485,7 @@ def new_contain(self, at_type: Union[str, ThingTypesBase], **contains_metadata) at_type = ThingTypesBase.from_str(at_type) if at_type not in self.contains: - new_contain = Contain(contains_metadata) + new_contain = Contain._for_type(at_type, contains_metadata) self.add_contain(new_contain, at_type) return new_contain @@ -563,6 +563,15 @@ class Contain(DataDict[str, str]): annotation type in the ``contains`` metadata of a MMIF view. """ + @classmethod + def _for_type(cls, at_type: ThingTypesBase, contains_metadata: dict) -> 'Contain': + """ + Builds a ``Contain`` for the given ``@type``, with property names + rewritten to their canonical spellings (see + :func:`mmif.serialize.annotation.canonicalize_prop_name`). + """ + return cls({canonicalize_prop_name(at_type, k): v for k, v in contains_metadata.items()}) + class AnnotationsList(DataList[Union[Annotation, Document]]): """ @@ -608,7 +617,7 @@ def _deserialize(self, input_dict: dict) -> None: for key, value in input_dict.items(): if isinstance(key, str): key = ThingTypesBase.from_str(key) - self._items[key] = Contain(value) + self._items[key] = Contain._for_type(key, value) def update(self, other: Union[dict, 'ContainsDict'], overwrite=False): for k, v in other.items(): diff --git a/mmif/utils/summarizer/nodes.py b/mmif/utils/summarizer/nodes.py index 53201022..fa2e467c 100644 --- a/mmif/utils/summarizer/nodes.py +++ b/mmif/utils/summarizer/nodes.py @@ -1,10 +1,64 @@ import json +from collections.abc import MutableMapping from typing import Any +from mmif.serialize.annotation import canonicalize_prop_name +from mmif.serialize.model import MmifObject from mmif.utils.summarizer import config +class NodeProperties(MutableMapping): + + """Read-through view of an annotation's properties for use in a Node. Reads + resolve, in order, local writes made by the summarizer, the annotation's own + properties (deprecated spellings included, e.g. `word` for `text`), and the + view-level defaults for the annotation's type in the view's `contains` + metadata. Values are plain JSON types, as they would be after a serialization + round trip. Writes never reach the annotation; they are kept locally so that + summarizing leaves the loaded MMIF unchanged. + + Note on the alternative: the SDK could instead copy `contains` defaults into + each annotation's `.properties` at deserialization and rely on + `Mmif.factor_out_shared_properties` to lift them back out on serialization. + That round trip is not shape-preserving when a type has a single annotation + in the view, when the key is non-factorable (`start`, `source`, `target`, ...), + or when another annotation of the type overrides the default, so the SDK + keeps defaults out of `.properties` and this class resolves them here.""" + + def __init__(self, annotation, view): + self.annotation = annotation + contains = {} if view is None else view.metadata.contains.get(annotation.at_type, {}) + self.defaults = {canonicalize_prop_name(annotation.at_type, k): v for k, v in contains.items()} + self.local = {} + + @staticmethod + def _as_json(value): + return json.loads(str(value)) if isinstance(value, MmifObject) else value + + def __getitem__(self, key): + if key in self.local: + return self.local[key] + if key in self.annotation.properties: + return self._as_json(self.annotation.properties[key]) + return self.defaults[canonicalize_prop_name(self.annotation.at_type, key)] + + def __setitem__(self, key, value): + self.local[key] = value + + def __delitem__(self, key): + del self.local[key] + + def __iter__(self): + seen = set() + for key in (*self.annotation.properties, *self.defaults, *self.local): + if key not in seen: + seen.add(key) + yield key + + def __len__(self): + return sum(1 for _ in self) + class Node(object): @@ -16,7 +70,7 @@ def __init__(self, graph, view, annotation): # copy some information from the Annotation self.at_type = annotation.at_type self.identifier = annotation.id - self.properties = json.loads(str(annotation.properties)) + self.properties = NodeProperties(annotation, view) # get the document from the view or the properties self.document = self._get_document() # The targets property contains a list of annotations or documents that @@ -151,23 +205,10 @@ def add_anchors_from_alignment(self, target: Any, debug=False): def _get_document(self): """Return the document or annotation node that the annotation/document in - the node refers to via the document property. This could be a local property - or a metadata property if there is no such local property. Return None - if neither of those exist.""" - # try the local property + the node refers to via the document property (the annotation's own, or the + view-level default). Return None if there is none.""" docid = self.properties.get('document') - if docid is not None: - # print('>>>', docid, self.graph.get_node(docid)) - return self.graph.get_node(docid) - # try the metadata property - if self.view is not None: - try: - metadata = self.view.metadata.contains[self.at_type] - docid = metadata['document'] - return self.graph.get_node(docid) - except KeyError: - return None - return None + return None if docid is None else self.graph.get_node(docid) def summary(self): """The default summary is just the identfier, this should typically be @@ -197,9 +238,9 @@ def pp(self, close=True): class TimeFrameNode(Node): def __str__(self): - frame_type = ' ' + self.frame_type() if self.has_label() else '' + label = f' {self.label()}' if self.has_label() else '' return ('' - % (self.identifier, self.start(), self.end(), frame_type)) + % (self.identifier, self.start(), self.end(), label)) def start(self): return self.properties.get('start', -1) @@ -207,13 +248,11 @@ def start(self): def end(self): return self.properties.get('end', -1) - def frame_type(self): - # TODO: rename this, uses old property since replaced by "label"" - # NOTE: this is still aloowing for the old property though - return self.properties.get('label') or self.properties.get('frameType') + def label(self): + return self.properties.get('label') def has_label(self): - return self.frame_type() is not None + return self.label() is not None def representatives(self) -> list: """Return a list of the representative TimePoints.""" @@ -228,7 +267,7 @@ def summary(self): return { 'id': self.identifier, 'start': self.properties['start'], 'end': self.properties['end'], - 'frameType': self.properties.get('frameType') } + 'frameType': self.label() } class EntityNode(Node): diff --git a/mmif/utils/summarizer/summary.py b/mmif/utils/summarizer/summary.py index 28339fad..0db14e66 100644 --- a/mmif/utils/summarizer/summary.py +++ b/mmif/utils/summarizer/summary.py @@ -147,7 +147,7 @@ def __init__(self, summary): self.data = defaultdict(list) # summary.graph.get_nodes(config.ANNOTATION, view_id=view.id) for anno in summary.graph.get_nodes(config.ANNOTATION): - self.data[anno.view.id].append(anno.properties) + self.data[anno.view.id].append(dict(anno.properties)) def get(self, item): return self.data.get(item, []) @@ -300,7 +300,7 @@ def __init__(self, identifier: str, sentence: list, transcript: CharacterList): # this adds the current token to the transcript start = t.properties['start'] end = t.properties['end'] - word = t.properties['word'] + word = t.properties['text'] transcript.set_chars(word, start, end) self.id = identifier self.start = sentence[0].anchors['time-offsets'][0] @@ -378,7 +378,7 @@ def __init__(self, summary): def _collect_timeframe_summaries(self): for tf in self.nodes: - label = tf.frame_type() + label = tf.label() try: start, end = tf.anchors['time-offsets'] except KeyError: diff --git a/tests/test_serialize.py b/tests/test_serialize.py index 3873678f..0ebe50c0 100644 --- a/tests/test_serialize.py +++ b/tests/test_serialize.py @@ -594,20 +594,6 @@ def test___contains__(self): self.assertTrue('v5' in mmif_obj) self.assertFalse('v432402' in mmif_obj) - def test_get_label(self): - v = View() - a = v.new_annotation(AnnotationTypes.TimeFrame, label="speech") - self.assertEqual(a._get_label(), "speech") - a = v.new_annotation(AnnotationTypes.TimeFrame, frameType="speech") - self.assertEqual(a._get_label(), "speech") - a = v.new_annotation(AnnotationTypes.BoundingBox, label="text") - self.assertEqual(a._get_label(), "text") - a = v.new_annotation(AnnotationTypes.BoundingBox, boxType="text") - self.assertEqual(a._get_label(), "text") - with self.assertRaises(KeyError): - a = v.new_annotation(AnnotationTypes.BoundingBox) - _ = a._get_label() - def test_timestamp_uses_utc_with_z_suffix(self): """Test that timestamps are in UTC with 'Z' suffix to avoid ambiguity""" from datetime import timezone @@ -883,8 +869,10 @@ def test_new_contain(self): self.view_obj.new_contain("http://vocab.lappsgrid.org/Token") # can add by obj at_type self.view_obj.new_contain(AnnotationTypes.TimePoint) - # can add details - self.view_obj.new_contain(AnnotationTypes.TimeFrame, **{"frameType": "speech"}) + # can add details (to a type not yet in the view); a deprecated + # property spelling is stored under the canonical name + contain = View().new_contain(AnnotationTypes.TimeFrame, **{"frameType": "speech"}) + self.assertEqual({"label": "speech"}, dict(contain.items())) with pytest.raises(ValueError): # empty at_type is not allowed self.view_obj.new_contain("") @@ -895,6 +883,19 @@ def test_add_annotation(self): self.view_obj.add_annotation(anno_obj) # raise exception if this fails self.assertEqual(old_len+1, len(self.view_obj.annotations)) self.assertIn('http://vocab.lappsgrid.org/NamedEntity', self.view_obj.metadata.contains) + # an added annotation picks up the view-level defaults of its type + # (as ephemeral properties, not as its own) ... + tf = Annotation({'@type': str(AnnotationTypes.TimeFrame), 'properties': {'id': 'tf_x', 'start': 0, 'end': 1}}) + self.view_obj.add_annotation(tf) + self.assertEqual('m1', tf.get_property('document')) + self.assertNotIn('document', list(tf.properties)) + # ... under canonical names, however the view spelled them + v = View() + v.new_contain(AnnotationTypes.TimeFrame, frameType='speech') + tf = v.add_annotation(Annotation({'@type': str(AnnotationTypes.TimeFrame), + 'properties': {'id': 'tf_1', 'start': 0, 'end': 1}})) + self.assertEqual('speech', tf.get_property('label')) + self.assertEqual('speech', tf.get_property('frameType')) _ = self.view_obj.serialize() # raise exception if this fails self.view_obj.new_annotation(AnnotationTypes.TimePoint) roundtrip = View(json.loads(self.view_obj.serialize())) @@ -907,7 +908,7 @@ def test_new_annotation(self): a2 = self.view_obj.new_annotation('TimeFrame') self.assertNotEqual(a1.id, a2.id) self.assertEqual(a1.id.rsplit('_', 1)[0], a2.id.rsplit('_', 1)[0]) - a3 = self.view_obj.new_annotation('TimeFrame', frameType='speech', start=100, end=500) + a3 = self.view_obj.new_annotation('TimeFrame', label='speech', start=100, end=500) self.assertEqual(4, len(a3.properties)) def test_new_textdocument(self): @@ -965,6 +966,9 @@ def test_get_annotations(self): # just property annotations = list(mmif_obj['v3'].get_annotations(frameType='speech')) self.assertEqual(len(annotations), 1) + # the same property, asked for by its canonical name + annotations = list(mmif_obj['v3'].get_annotations(AnnotationTypes.TimeFrame, label='speech')) + self.assertEqual(len(annotations), 1) # at_type + annotation metadata annotations = list(mmif_obj['v3'].get_annotations(AnnotationTypes.TimeFrame, timeUnit='milliseconds')) self.assertEqual(len(annotations), 2) @@ -1057,8 +1061,11 @@ def setUp(self) -> None: def test_annotation_properties_wrapper(self): ann_obj = Annotation(self.data['everything']['annotations'][0]) props_json = self.data['everything']['annotations'][0]['properties'] - props_obj = AnnotationProperties(props_json) - self.assertEqual(ann_obj.properties, props_obj) + # this annotation carries the deprecated `frameType` spelling; the + # wrapper holds the same properties, under canonical names + self.assertIn('frameType', props_json) + self.assertEqual(ann_obj.properties, AnnotationProperties(props_json, ann_obj)) + self.assertNotIn('frameType', list(ann_obj.properties)) ann_obj.add_property('new_prop', 'new_prop_value') self.assertEqual(ann_obj.properties['new_prop'], 'new_prop_value') for k in ann_obj.properties.keys(): @@ -1178,50 +1185,95 @@ def test_get_property(self): def test_get_property_with_alias(self): v = View() tf1 = v.new_annotation(AnnotationTypes.TimeFrame, start=0, end=10, label="speech") - # tf3 = v.new_annotation(AnnotationTypes.TimeFrame, start=20, end=30, frameType="speech") self.assertEqual(tf1.get_property('label'), "speech") self.assertEqual(tf1.get_property('frameType'), "speech") - with warnings.catch_warnings(record=True) as caught_warnings: - # should throw a warning for alias collision - tf2 = v.new_annotation(AnnotationTypes.BoundingBox, start=10, end=20, - label="nonspeech", boxType="nonspeech") - # again, should throw a warning for alias collision for frameType and label, - # but not for frameLabel - tf3 = v.new_annotation(AnnotationTypes.TimeFrame, start=10, end=20, - frameType="nonspeech", label="nonspeech", frameLabel="speech") - self.assertEqual(2, len(caught_warnings)) - self.assertTrue("nonspeech", tf3.get_property('frameType')) - self.assertTrue("nonspeech", tf3.get_property('label')) - self.assertTrue("speech", tf3.get_property('frameLabel')) + # for every alias group the vocabulary defines on these types: a + # deprecated spelling is stored under the canonical name (with a + # deprecation warning), and either spelling reads, tests membership, + # and deletes against that single key + for at_type in (AnnotationTypes.TimeFrame, AnnotationTypes.BoundingBox, AnnotationTypes.Token): + for canonical, group in at_type._property_aliases.items(): + for alias in group - {canonical}: + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter("always") + ann = v.new_annotation(at_type, **{alias: "x"}) + self.assertEqual(1, len([w for w in caught_warnings + if issubclass(w.category, DeprecationWarning)])) + self.assertEqual([canonical], [k for k in ann.properties if k in (canonical, alias)]) + self.assertIn(alias, ann.properties) + self.assertEqual("x", ann.properties[alias]) + self.assertEqual("x", ann.get_property(canonical)) + self.assertEqual("x", ann.get_property(alias)) + del ann.properties[alias] + self.assertNotIn(canonical, ann.properties) + # setting one property twice under different spellings: the last + # write wins, as for any property; frameLabel is not an alias + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + tf3 = v.new_annotation(AnnotationTypes.TimeFrame, start=10, end=20, + label="speech", frameType="nonspeech", frameLabel="speech") + self.assertEqual("nonspeech", tf3.get_property('label')) + self.assertEqual("speech", tf3.get_property('frameLabel')) def test_alias_precedes_view_contains_default(self): - # An annotation that sets a property via a deprecated alias (frameType - # aliases label) provides an annotation-level value. A view-level - # `contains` default for the canonical name must NOT override it -- the - # annotation-level value takes precedence over the view-level default. + # An annotation-level value always wins over a view-level `contains` + # default for the same canonical property, and its alias name must + # resolve to that same annotation-level value, never to the + # `contains` default, regardless of whether the annotation supplied + # its value under the canonical name or one of its aliases. When + # there is no annotation-level value at all, a genuine view-level + # default applies and is queryable through both names. tf = "http://mmif.clams.ai/vocabulary/TimeFrame/v6" - raw = { - "metadata": {"mmif": "http://mmif.clams.ai/1.2.0"}, - "documents": [{"@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1", - "properties": {"id": "m1", "mime": "video/mp4", - "location": "file:///v.mp4"}}], - "views": [{"id": "v1", - "metadata": {"app": "http://x/1", "contains": {tf: {"label": "X"}}}, - "annotations": [{"@type": tf, - "properties": {"id": "v1:a1", "start": 0, - "end": 5, "frameType": "bars"}}]}] - } - mmif_obj = Mmif(json.dumps(raw), validate=False) - ann = mmif_obj['v1:a1'] - # the annotation's own frameType (alias of label) wins over contains + + def get_property_resolution(ann_props, contains_props): + raw = { + "metadata": {"mmif": "http://mmif.clams.ai/1.2.0"}, + "documents": [{"@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1", + "properties": {"id": "m1", "mime": "video/mp4", + "location": "file:///v.mp4"}}], + "views": [{"id": "v1", + "metadata": {"app": "http://x/1", "contains": {tf: contains_props}}, + "annotations": [{"@type": tf, + "properties": {"id": "v1:a1", "start": 0, + "end": 5, **ann_props}}]}] + } + return Mmif(json.dumps(raw), validate=False)['v1:a1'] + + # own value set via the deprecated alias (frameType); contains gives a + # default under the canonical name (label). The alias-set value wins. + ann = get_property_resolution({"frameType": "bars"}, {"label": "X"}) + self.assertEqual('bars', ann.get_property('label')) + self.assertEqual('bars', ann.get_property('frameType')) + + # own value set via the canonical name (label) itself; contains gives + # a default for that same canonical name. The annotation's own value + # still wins, and the alias (frameType) must resolve to it too, not + # to the contains default. + ann = get_property_resolution({"label": "bars"}, {"label": "credits"}) self.assertEqual('bars', ann.get_property('label')) self.assertEqual('bars', ann.get_property('frameType')) - # a genuine view-level default (not shadowed by any annotation value) - # still applies - raw['views'][0]['metadata']['contains'][tf] = {"timeUnit": "milliseconds"} - raw['views'][0]['annotations'][0]['properties'] = {"id": "v1:a1", "start": 0, "end": 5} - ann = Mmif(json.dumps(raw), validate=False)['v1:a1'] - self.assertEqual('milliseconds', ann.get_property('timeUnit')) + + # no own value at all: the view-level default applies, and is + # queryable through both the canonical name and its alias. + ann = get_property_resolution({}, {"label": "bars"}) + self.assertEqual('bars', ann.get_property('label')) + self.assertEqual('bars', ann.get_property('frameType')) + + # the view-level default may itself use the deprecated spelling + ann = get_property_resolution({}, {"frameType": "bars"}) + self.assertEqual('bars', ann.get_property('label')) + self.assertEqual('bars', ann.get_property('frameType')) + ann = get_property_resolution({"label": "own"}, {"frameType": "X"}) + self.assertEqual('own', ann.get_property('label')) + self.assertEqual('own', ann.get_property('frameType')) + + # both spellings on the annotation itself: the canonical one wins, + # with a warning, and only the canonical key remains + with self.assertWarns(UserWarning): + ann = get_property_resolution({"frameType": "B", "label": "A"}, {}) + self.assertEqual('A', ann.get_property('label')) + self.assertEqual('A', ann.get_property('frameType')) + self.assertNotIn('frameType', list(ann.properties)) def test_change_id(self): anno_obj: Annotation = self.data['everything']['mmif']['v5:bb1'] @@ -1274,6 +1326,13 @@ def test_get_property(self): self.assertEqual(d.id, d.get_property('id')) self.assertEqual(d.location, d.get_property('location')) self.assertEqual('value1', d.get_property('prop1')) + # document types have no property aliases; a name that is an alias + # for annotation types (word) is stored as spelled, without a warning + with warnings.catch_warnings(): + warnings.simplefilter("error") + d.properties['word'] = 'w' + self.assertIn('word', list(d.properties)) + self.assertEqual('w', d.get_property('word')) d.id = 'v1:d1' self.assertEqual('v1:d1', d.long_id) self.assertEqual('v1:d1', d._short_id) @@ -1532,35 +1591,45 @@ def test_capital_annotation_no_collision_across_apps(self): self.assertEqual({1709977.0, 1709975}, durations) def test_factor_out_shared_properties(self): - # A view-level metadata property (document/timeUnit/labelset) shared + # A view-level metadata property (document/timeUnit/label) shared # across all annotations of a type is factored up into the view's - # `contains` on serialize, while instance-level properties (label) and - # anchors stay on the annotations. + # `contains` on serialize, while an instance-level property that + # differs (note) stays on the annotations. The shared `label` is + # spelled by its alias (frameType) on one annotation, so this also + # covers factoring across spellings and round-tripping a factored-out + # value through its alias, not just its canonical name. m = Mmif(validate=False) v = m.new_view(); v.metadata.app = tester_appname v.new_annotation(AnnotationTypes.TimeFrame, document='m1', timeUnit='ms', - start=0, end=5, label='bars') - v.new_annotation(AnnotationTypes.TimeFrame, document='m1', timeUnit='ms', - start=5, end=9, label='slate') + start=0, end=5, label='bars', note='first') + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + v.new_annotation(AnnotationTypes.TimeFrame, document='m1', timeUnit='ms', + start=5, end=9, frameType='bars', note='second') out = json.loads(m.serialize()) vout = out['views'][0] tf_key = next(k for k in vout['metadata']['contains'] if 'TimeFrame' in k) contains = vout['metadata']['contains'][tf_key] - # shared `document`/`timeUnit` are factored up ... + # shared `document`/`timeUnit`/`label` are factored up ... self.assertEqual('m1', contains.get('document')) self.assertEqual('ms', contains.get('timeUnit')) + self.assertEqual('bars', contains.get('label')) for a in vout['annotations']: # ... and removed from each annotation ... self.assertNotIn('document', a['properties']) self.assertNotIn('timeUnit', a['properties']) - # ... while anchors and the differing label stay per-annotation + self.assertNotIn('label', a['properties']) + # ... while anchors and the differing `note` stay per-annotation self.assertIn('start', a['properties']) - self.assertIn('label', a['properties']) - # round-trip: the factored values still resolve per annotation + self.assertIn('note', a['properties']) + # round-trip: the factored values still resolve per annotation, + # including through the `label` alias (`frameType`) m2 = Mmif(m.serialize()) for a in m2.views[0].get_annotations(AnnotationTypes.TimeFrame): self.assertEqual('m1', a.get_property('document')) self.assertEqual('ms', a.get_property('timeUnit')) + self.assertEqual('bars', a.get_property('label')) + self.assertEqual('bars', a.get_property('frameType')) def test_factor_out_protects_listed_props_and_skips_singletons(self): # Blocklisted props (anchors/links) are never factored even when diff --git a/tests/test_utils_cli.py b/tests/test_utils_cli.py index dd33fec2..fa286e22 100644 --- a/tests/test_utils_cli.py +++ b/tests/test_utils_cli.py @@ -363,6 +363,29 @@ def test_summarize_validates_content(self): finally: os.unlink(tmp_file) + def test_summarize_timeframes_with_lifted_label(self): + # a `label` shared by all TimeFrames in a view is lifted into the + # view's `contains` metadata at serialization; the summarizer must + # still report it for each time frame + mmif_obj = Mmif(BASIC_MMIF_STRING) + view = mmif_obj.new_view() + view.metadata.app = 'http://apps.clams.ai/test/v1' + for start, end in ((0, 5), (5, 9)): + view.new_annotation(AnnotationTypes.TimeFrame, document='d1', timeUnit='milliseconds', + start=start, end=end, label='slate') + serialized = json.loads(mmif_obj.serialize()) + tf_key = next(k for k in serialized['views'][0]['metadata']['contains'] if 'TimeFrame' in k) + self.assertEqual('slate', serialized['views'][0]['metadata']['contains'][tf_key]['label']) + tmp_file = self.create_temp_mmif_file(mmif_obj) + try: + output = self.run_cli_capture_stdout( + argparse.Namespace(MMIF_FILE=tmp_file, output=None, pretty=False) + ) + timeframes = output['timeframes'][view.metadata.app] + self.assertEqual(['slate', 'slate'], [tf['label'] for tf in timeframes]) + finally: + os.unlink(tmp_file) + if __name__ == '__main__': unittest.main()