diff --git a/documentation/target-versions.csv b/documentation/target-versions.csv
index d6d11880..13953245 100644
--- a/documentation/target-versions.csv
+++ b/documentation/target-versions.csv
@@ -1,4 +1,5 @@
"``mmif-python`` version","Target MMIF Specification"
+`1.5.3 `__,`1.2.0 `__
`1.5.2 `__,`1.2.0 `__
`1.5.1 `__,`1.2.0 `__
`1.5.0 `__,`1.2.0 `__
diff --git a/mmif/serialize/mmif.py b/mmif/serialize/mmif.py
index 245c96aa..802e7d77 100644
--- a/mmif/serialize/mmif.py
+++ b/mmif/serialize/mmif.py
@@ -175,6 +175,14 @@ class Mmif(MmifObject):
all_views = mmif.views[1:4] # Slice of views
"""
+ #: Hardcoded list of property names that must stay at the individual
+ #: annotation level, and never "factored" into the view's ``contains``
+ #: metadata, even when redundant across annotations.
+ _NON_FACTORABLE_PROPS = frozenset({
+ 'start', 'end', 'targets', 'source', 'target', 'representatives',
+ 'mime', 'location', 'location_', 'text',
+ })
+
def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *, validate: bool = True) -> None:
self.metadata: MmifMetadata = MmifMetadata()
self.documents: DocumentsList = DocumentsList()
@@ -211,26 +219,42 @@ def validate(json_str: Union[bytes, str, dict]) -> None:
json_str = json.loads(json_str)
jsonschema.validators.validate(json_str, schema)
- def serialize(self, sanitize: bool = False, autogenerate_capital_annotations: bool = True, **kwargs) -> str:
+ def serialize(self,
+ pretty: bool = False,
+ include_context: bool = True,
+ *,
+ sanitize: bool = False,
+ autogenerate_capital_annotations: bool = True,
+ factor_out_shared_properties: bool = True) -> str:
"""
Serializes the MMIF object to a JSON string.
+ :param pretty: If True, returns the string with indentation.
+ :param include_context: If False, excludes contextual attributes (e.g.
+ runtime timestamps) from serialization.
:param sanitize: If True, performs some sanitization of before returning
the JSON string. See :meth:`sanitize` for details.
:param autogenerate_capital_annotations: If True, automatically convert
any "pending" temporary properties from `Document` objects to
`Annotation` objects. See :meth:`generate_capital_annotations` for
details.
- :param kwargs: Keyword arguments to pass to the parent's ``serialize``
- method (e.g., ``pretty=True``, ``include_context=False``).
+ :param factor_out_shared_properties: If True, lift properties shared
+ (same value) across all annotations of a type in a view into the
+ view's ``contains`` metadata. See
+ :meth:`factor_out_shared_properties` for details. Independent of
+ ``autogenerate_capital_annotations``, but runs after it (so
+ freshly-materialized pending properties are included) when both are
+ enabled.
:return: JSON string of the MMIF object.
"""
if autogenerate_capital_annotations:
self.generate_capital_annotations()
+ if factor_out_shared_properties:
+ self.factor_out_shared_properties()
# sanitization should be done after `Annotation` annotations are generated
if sanitize:
self.sanitize()
- return super().serialize(**kwargs)
+ return super().serialize(pretty=pretty, include_context=include_context)
def _deserialize(self, input_dict: dict) -> None:
"""
@@ -261,9 +285,16 @@ def _deserialize(self, input_dict: dict) -> None:
for ann in view.get_annotations():
## for "capital" Annotation properties
# first add all extrinsic properties to the Annotation objects
- # as "ephemeral" properties
+ # 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():
- ann._props_ephemeral[prop_key] = prop_value
+ if prop_key not in ann._props_ephemeral:
+ ann._props_ephemeral[prop_key] = prop_value
# 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:
@@ -297,7 +328,7 @@ def _when_failed():
f"Alignment {alignment_ann.id} has `source` and `target` properties that do not point to Annotation objects.",
RuntimeWarning)
## caching alignments
- if all(map(lambda x: x in alignment_ann.properties, ('source', 'target'))):
+ if all(map(lambda x: x in alignment_ann, ('source', 'target'))):
try:
source_ann = self.__getitem__(alignment_ann.get('source'))
target_ann = self.__getitem__(alignment_ann.get('target'))
@@ -383,15 +414,51 @@ def generate_capital_annotations(self):
if view_to_write.metadata.app == current_app and doc_id in view_to_write.annotations:
view_to_write[doc_id].properties.update(props)
else:
- if len(anns_to_write) == 1:
- # if there's only one document, we can record the doc_id in the contains metadata
- view_to_write.metadata.new_contain(AnnotationTypes.Annotation, document=doc_id)
- props.pop('document', None)
- else:
- # otherwise, doc_id needs to be recorded in the annotation property
- props['document'] = doc_id
+ props['document'] = doc_id
view_to_write.new_annotation(AnnotationTypes.Annotation, **props)
+
+ def factor_out_shared_properties(self):
+ """
+ Factor properties that are shared (identical value) across all
+ annotations of a type within a view up into that view's ``contains``
+ metadata, per the MMIF spec recommendation to prefer view-level
+ defaults for properties shared among annotations of a type.
+
+ A property is factored when it is present, with the same value, on
+ **every** annotation of a given ``@type`` in the view, there are at
+ least two such annotations (so the value is genuinely shared), and the
+ key is not in the :attr:`_NON_FACTORABLE_PROPS` protection list.
+ Factored values are mirrored into each annotation's ephemeral props at
+ deserialization time, so in-memory access is unchanged, while the
+ serialized output carries the value once, in ``contains``.
+ """
+ for view in self.views:
+ by_type = defaultdict(list)
+ for ann in view.get_annotations():
+ by_type[ann.at_type].append(ann)
+ for at_type, anns in by_type.items():
+ if len(anns) < 2:
+ continue
+ shared = {}
+ for key in list(anns[0].properties.keys()):
+ if key in self._NON_FACTORABLE_PROPS:
+ continue
+ value = anns[0].properties[key]
+ if all(key in a.properties and a.properties[key] == value
+ for a in anns):
+ shared[key] = value
+ if not shared:
+ continue
+ if at_type not in view.metadata.contains:
+ view.metadata.new_contain(at_type)
+ contain = view.metadata.contains[at_type]
+ for key, value in shared.items():
+ contain[key] = value
+ for a in anns:
+ del a.properties[key]
+ a._props_ephemeral[key] = value
+
def sanitize(self):
"""
Sanitizes a Mmif object by running some safeguards.
diff --git a/tests/test_serialize.py b/tests/test_serialize.py
index 271cf9c1..3873678f 100644
--- a/tests/test_serialize.py
+++ b/tests/test_serialize.py
@@ -76,7 +76,7 @@ def test_str_mmif_deserialize(self):
mmif_obj = Mmif(example)
except ValidationError:
self.fail(f"example {i}")
- self.assertEqual(mmif_obj.serialize(True), Mmif(mmif_obj.serialize()).serialize(True), f'Failed on {i}')
+ self.assertEqual(mmif_obj.serialize(pretty=True), Mmif(mmif_obj.serialize()).serialize(pretty=True), f'Failed on {i}')
def test_json_mmif_deserialize(self):
for i, example in MMIF_EXAMPLES.items():
@@ -91,7 +91,7 @@ def test_json_mmif_deserialize(self):
for annotation in view.annotations:
self.assertIn('_type', annotation.__dict__)
self.assertTrue('id' in list(mmif_obj.views._items.values())[0].__dict__)
- self.assertEqual(mmif_obj.serialize(True), Mmif(json.loads(mmif_obj.serialize())).serialize(True), f'Failed on {i}')
+ self.assertEqual(mmif_obj.serialize(pretty=True), Mmif(json.loads(mmif_obj.serialize())).serialize(pretty=True), f'Failed on {i}')
def test_str_vs_json_deserialize(self):
for i, example in MMIF_EXAMPLES.items():
@@ -99,7 +99,7 @@ def test_str_vs_json_deserialize(self):
continue
str_mmif_obj = Mmif(example)
json_mmif_obj = Mmif(json.loads(example))
- self.assertEqual(str_mmif_obj.serialize(True), json_mmif_obj.serialize(True), f'Failed on {i}')
+ self.assertEqual(str_mmif_obj.serialize(pretty=True), json_mmif_obj.serialize(pretty=True), f'Failed on {i}')
def test_bad_mmif_deserialize_no_metadata(self):
self.mmif_examples_json['everything'].pop('metadata')
@@ -1194,6 +1194,35 @@ def test_get_property_with_alias(self):
self.assertTrue("nonspeech", tf3.get_property('label'))
self.assertTrue("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.
+ 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
+ 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'))
+
def test_change_id(self):
anno_obj: Annotation = self.data['everything']['mmif']['v5:bb1']
@@ -1502,6 +1531,100 @@ def test_capital_annotation_no_collision_across_apps(self):
self.assertEqual(2, len(caps))
self.assertEqual({1709977.0, 1709975}, durations)
+ def test_factor_out_shared_properties(self):
+ # A view-level metadata property (document/timeUnit/labelset) 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.
+ 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')
+ 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 ...
+ self.assertEqual('m1', contains.get('document'))
+ self.assertEqual('ms', contains.get('timeUnit'))
+ 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.assertIn('start', a['properties'])
+ self.assertIn('label', a['properties'])
+ # round-trip: the factored values still resolve per annotation
+ 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'))
+
+ def test_factor_out_protects_listed_props_and_skips_singletons(self):
+ # Blocklisted props (anchors/links) are never factored even when
+ # identical across annotations; a type with a single annotation is
+ # skipped (needs > 1 to be genuinely shared).
+ m = Mmif(validate=False)
+ v = m.new_view(); v.metadata.app = tester_appname
+ tf = v.new_annotation(AnnotationTypes.TimeFrame, start=0, end=5)
+ td1 = v.new_textdocument(text='a'); td2 = v.new_textdocument(text='b')
+ # two Alignments share the same `source` -- must NOT be factored out
+ v.new_annotation(AnnotationTypes.Alignment, source=tf.id, target=td1.id)
+ v.new_annotation(AnnotationTypes.Alignment, source=tf.id, target=td2.id)
+ out = json.loads(m.serialize())
+ aligns = [a for a in out['views'][0]['annotations'] if 'Alignment' in a['@type']]
+ for a in aligns:
+ self.assertIn('source', a['properties'])
+ # a single-annotation type is not factored (threshold)
+ m2 = Mmif(validate=False); v2 = m2.new_view(); v2.metadata.app = tester_appname
+ v2.new_annotation(AnnotationTypes.TimeFrame, document='m1', start=0, end=5)
+ out2 = json.loads(m2.serialize())
+ self.assertIn('document', out2['views'][0]['annotations'][0]['properties'])
+
+ def test_alignment_cached_from_view_level_source_target(self):
+ # regression for the `_cache_alignment` fix: `source`/`target` supplied
+ # as view-level `contains` defaults (distributed to ephemeral, absent
+ # from the annotation's own properties) must still be honored.
+ al = "http://mmif.clams.ai/vocabulary/Alignment/v1"
+ tf = "http://mmif.clams.ai/vocabulary/TimeFrame/v6"
+ td = "http://mmif.clams.ai/vocabulary/TextDocument/v2"
+ raw = {
+ "metadata": {"mmif": "http://mmif.clams.ai/1.2.0"},
+ "documents": [],
+ "views": [{"id": "v1",
+ "metadata": {"app": "http://x/1", "contains": {
+ al: {"source": "v1:tf1", "target": "v1:td1"}}},
+ "annotations": [
+ {"@type": tf, "properties": {"id": "v1:tf1", "start": 0, "end": 5}},
+ {"@type": td, "properties": {"id": "v1:td1", "text": {"@value": "x"}}},
+ {"@type": al, "properties": {"id": "v1:a1"}}]}]
+ }
+ m = Mmif(json.dumps(raw), validate=False)
+ self.assertTrue(list(m['v1:tf1'].get_all_aligned()))
+
+ def test_factor_out_shared_properties_flag(self):
+ # `factor_out_shared_properties` gates factoring independently of
+ # `autogenerate_capital_annotations`.
+ m = Mmif(validate=False)
+ v = m.new_view(); v.metadata.app = tester_appname
+ v.new_annotation(AnnotationTypes.TimeFrame, document='m1', start=0, end=5)
+ v.new_annotation(AnnotationTypes.TimeFrame, document='m1', start=5, end=9)
+ # disabled -> shared `document` stays on each annotation
+ off = json.loads(m.serialize(factor_out_shared_properties=False))
+ for a in off['views'][0]['annotations']:
+ self.assertIn('document', a['properties'])
+ # independent of generation: a reloaded MMIF (props already in
+ # `.properties`) is still factored with capital-annotation generation OFF
+ reloaded = Mmif(m.serialize(factor_out_shared_properties=False))
+ on = json.loads(reloaded.serialize(autogenerate_capital_annotations=False,
+ factor_out_shared_properties=True))
+ tf_key = next(k for k in on['views'][0]['metadata']['contains'] if 'TimeFrame' in k)
+ self.assertEqual('m1', on['views'][0]['metadata']['contains'][tf_key].get('document'))
+ for a in on['views'][0]['annotations']:
+ self.assertNotIn('document', a['properties'])
+
def test_capital_annotation_generation_viewfinder(self):
mmif = Mmif(validate=False)
for i in range(1, 3):