Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions documentation/target-versions.csv
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"``mmif-python`` version","Target MMIF Specification"
`1.5.3 <https://pypi.org/project/mmif-python/1.5.3/>`__,`1.2.0 <https://mmif.clams.ai/1.2.0/>`__
`1.5.2 <https://pypi.org/project/mmif-python/1.5.2/>`__,`1.2.0 <https://mmif.clams.ai/1.2.0/>`__
`1.5.1 <https://pypi.org/project/mmif-python/1.5.1/>`__,`1.2.0 <https://mmif.clams.ai/1.2.0/>`__
`1.5.0 <https://pypi.org/project/mmif-python/1.5.0/>`__,`1.2.0 <https://mmif.clams.ai/1.2.0/>`__
Expand Down
95 changes: 81 additions & 14 deletions mmif/serialize/mmif.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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:
"""
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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'))
Expand Down Expand Up @@ -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.
Expand Down
129 changes: 126 additions & 3 deletions tests/test_serialize.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand All @@ -91,15 +91,15 @@ 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():
if not i.startswith('mmif_'):
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')
Expand Down Expand Up @@ -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']

Expand Down Expand Up @@ -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):
Expand Down
Loading