diff --git a/documentation/target-versions.csv b/documentation/target-versions.csv
index 13953245..05eaa918 100644
--- a/documentation/target-versions.csv
+++ b/documentation/target-versions.csv
@@ -1,4 +1,5 @@
"``mmif-python`` version","Target MMIF Specification"
+`1.5.4 `__,`1.2.0 `__
`1.5.3 `__,`1.2.0 `__
`1.5.2 `__,`1.2.0 `__
`1.5.1 `__,`1.2.0 `__
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/mmif/utils/video_document_helper.py b/mmif/utils/video_document_helper.py
index bb1f5b1a..65edc93d 100644
--- a/mmif/utils/video_document_helper.py
+++ b/mmif/utils/video_document_helper.py
@@ -1,6 +1,8 @@
import contextvars
import importlib
+import shlex
import sys
+from contextlib import contextmanager
from enum import Enum
import math
@@ -27,6 +29,70 @@
'offset spans (see issue #379).'
)
+_MEDIA_CHECK_HINT = (
+ 'The file may be truncated or corrupted; container metadata does not '
+ 'reveal this, so to verify the file run '
+ '`ffmpeg -v error -i {path} -map 0:v:0 -f null -` and treat any stderr '
+ 'output as a problem (ffmpeg exits with 0 either way).'
+)
+
+
+def _media_check_hint(path: str) -> str:
+ """
+ Formats the "verify this file" hint for a resolved document path.
+
+ :param path: resolved local path of the video document
+ :returns: hint text, with the path quoted for shell use
+ :rtype: str
+ """
+ return _MEDIA_CHECK_HINT.format(path=shlex.quote(path))
+
+
+def _stream_rate(stream):
+ """
+ Resolves the frame rate of a video stream, as an exact fraction.
+
+ Prefers ffmpeg's guessed rate (``av_guess_frame_rate``), which falls back
+ to the stream's base rate (``r_frame_rate``) when the average rate
+ (``avg_frame_rate``) is unset. Matroska/webm commonly leave
+ ``average_rate`` as ``None`` (which would crash ``float()``) or 0, so it
+ cannot be used directly; both are falsy, so the chain yields the first
+ usable one. The result stays a ``Fraction`` for exact frame-count math.
+
+ :param stream: PyAV video stream
+ :returns: frame rate, or a falsy value when the stream reports none
+ """
+ return stream.guessed_rate or stream.average_rate or stream.base_rate
+
+
+@contextmanager
+def _ffmpeg_op(action: str, video_document: Document, path: str):
+ """
+ Turns a failed FFmpeg operation into an actionable ``RuntimeError``.
+
+ Any :py:class:`av.error.FFmpegError` (and the ``IndexError`` of a file
+ that opens but holds no video stream) is re-raised as a ``RuntimeError``
+ naming the document, the media check hint, and the original message. The
+ original message is repeated in the new one on purpose, because
+ downstream error reporting (e.g. ``clams-python``'s error view) records
+ only the outermost exception, so a chained cause would be dropped.
+
+ :param action: what was attempted, as a verb phrase (e.g. ``'open'``)
+ :param video_document: :py:class:`~mmif.serialize.annotation.Document`
+ holding the video document being read
+ :param path: resolved local path of the video document
+ :raises RuntimeError: if the wrapped operation fails
+ """
+ av = _check_cv_dep('av')
+ try:
+ yield
+ except (av.error.FFmpegError, IndexError) as e:
+ raise RuntimeError(
+ f'Cannot {action} video {video_document.id} ({path}). '
+ f'{_media_check_hint(path)} '
+ f'(original error: {type(e).__name__}: {e})'
+ ) from e
+
def _check_cv_dep(dep):
"""Import a CV dependency, raising ImportError with a helpful message."""
@@ -88,21 +154,28 @@ def open_container(video_document: Document):
:returns: open PyAV :py:class:`av.container.InputContainer`
:rtype: av.container.InputContainer
:raises ValueError: if ``video_document`` is missing or of the wrong type
+ :raises RuntimeError: if the file cannot be opened and read as a video,
+ for instance an mp4 whose ``moov`` atom is cut off, or a prefix too
+ short to hold a video stream
"""
av = _check_cv_dep('av')
if video_document is None or video_document.at_type != DocumentTypes.VideoDocument:
raise ValueError(f'The document does not exist.')
- container = av.open(video_document.location_path(nonexist_ok=False))
- stream = container.streams.video[0]
+ path = video_document.location_path(nonexist_ok=False)
+ container = None
+ with _ffmpeg_op('open', video_document, path):
+ try:
+ container = av.open(path)
+ stream = container.streams.video[0]
+ except BaseException:
+ # a file can open and still hold no video stream, in which case
+ # the container must not be left dangling
+ if container is not None:
+ container.close()
+ raise
time_base = float(stream.time_base)
- # Frame rate: prefer ffmpeg's guessed rate (av_guess_frame_rate), which
- # falls back to the stream's base rate (r_frame_rate) when the average
- # rate (avg_frame_rate) is unset. Matroska/webm commonly leave
- # `average_rate` as None (which would crash float()) or 0, so it cannot be
- # used directly. `rate` stays a Fraction for exact frame-count math; both
- # None and a zero rate are falsy, so the chain yields the first usable one.
- rate = stream.guessed_rate or stream.average_rate or stream.base_rate
+ rate = _stream_rate(stream)
fps = round(float(rate), 2) if rate else 0.0
# Resolve duration (in seconds) from the most reliable source available.
# `stream.duration` is exact for CFR H.264/MP4, but Matroska/webm leave it
@@ -121,13 +194,13 @@ def open_container(video_document: Document):
if stream.frames > 0:
frame_count = stream.frames
elif duration_s is not None and rate:
- frame_count = int(round(duration_s * float(rate)))
+ frame_count = round(duration_s * float(rate))
else:
frame_count = 0
if duration_s is not None:
- duration_ms = int(round(duration_s * 1000))
+ duration_ms = round(duration_s * 1000)
elif frame_count > 0 and fps > 0:
- duration_ms = int(round(frame_count / fps * 1000))
+ duration_ms = round(frame_count / fps * 1000)
else:
duration_ms = 0
video_document.add_property(FPS_DOCPROP_KEY, fps)
@@ -161,6 +234,78 @@ def get_framerate(video_document: Document) -> float:
container.close()
+def _decode_frames(container, stream, decode_errors: List[str]):
+ """
+ Iterates decoded frames of a video stream, stopping at the first error.
+
+ Decoding stops at the first :py:class:`av.error.FFmpegError`, which a
+ truncated or corrupted file can raise at any point, but the decoder is
+ flushed first so frames that are held back for reordering are not lost.
+ The error text is appended to ``decode_errors`` for the caller to report.
+
+ :param container: open PyAV :py:class:`av.container.InputContainer`
+ :param stream: video stream to decode
+ :param decode_errors: list that collects decoding error texts
+ :returns: generator of :py:class:`av.video.frame.VideoFrame`
+ """
+ av = _check_cv_dep('av')
+ try:
+ for frame in container.decode(stream):
+ yield frame
+ except av.error.FFmpegError as e:
+ decode_errors.append(f'{type(e).__name__}: {e}')
+ try:
+ for frame in stream.codec_context.decode(None):
+ yield frame
+ except av.error.FFmpegError as flush_e:
+ decode_errors.append(f'{type(flush_e).__name__}: {flush_e}')
+
+
+_MAX_LISTED_TIMEPOINTS = 5
+
+
+def _warn_missing_images(
+ video_document: Document,
+ path: str,
+ missing_ms: List[int],
+ n_missing_positions: int,
+ decode_errors: List[str],
+) -> None:
+ """
+ Issues a single warning for timepoints that have no image.
+
+ A missing timepoint that falls before the document's ``duration``, or a
+ decoding error, means the file holds fewer images than its container
+ metadata promises, hence the media check hint. Timepoints past a known
+ duration are simply out of range, and get no hint.
+
+ :param video_document: :py:class:`~mmif.serialize.annotation.Document`
+ holding a video document
+ :param path: resolved local path of the video document
+ :param missing_ms: sorted distinct timepoints (in ms) that have no image
+ :param n_missing_positions: number of positions in the returned list that
+ hold ``None``, which exceeds ``len(missing_ms)`` when timepoints repeat
+ :param decode_errors: decoding error texts collected while decoding
+ """
+ if len(missing_ms) <= _MAX_LISTED_TIMEPOINTS:
+ where = ', '.join(f'{ms}ms' for ms in missing_ms)
+ else:
+ where = (f'{len(missing_ms)} distinct timepoints between '
+ f'{missing_ms[0]}ms and {missing_ms[-1]}ms')
+ msg = (f'No image is available for {n_missing_positions} timepoint(s) '
+ f'({where}) of video {video_document.id}; the returned list holds '
+ f'`None` at those positions.')
+ duration_ms = video_document.get_property(DURATION_DOCPROP_KEY)
+ # a zero duration means the container reports none at all, so the
+ # out-of-range test cannot be applied and the file stays suspect
+ if decode_errors or not duration_ms or missing_ms[0] < duration_ms:
+ msg += f' {_media_check_hint(path)}'
+ if decode_errors:
+ msg += f' (first decoding error: {decode_errors[0]})'
+ # helper -> public extraction function -> the caller to be pointed at
+ warnings.warn(msg, stacklevel=3)
+
+
def extract_images_from_timepoints(
video_document: Document,
timepoints_ms: Iterable[int],
@@ -173,13 +318,20 @@ def extract_images_from_timepoints(
presentation timestamp (PTS) is closest to it. Duplicate timepoints
produce duplicate images at the same list positions as the input.
+ The returned list always has one entry per requested timepoint. An entry
+ is ``None`` when no image is available, namely when the timepoint is
+ more than one frame duration past the last decodable frame. This happens
+ for timepoints past the end of the video, and also within a truncated or
+ corrupted file, where decoding stops early. Such a call issues a single
+ warning naming the missing timepoints.
+
:param video_document: :py:class:`~mmif.serialize.annotation.Document`
holding a video document (``"@type": ".../VideoDocument/..."``)
:param timepoints_ms: iterable of timepoint values in milliseconds
:param as_PIL: return :py:class:`PIL.Image.Image` (RGB) instead of
:py:class:`~numpy.ndarray` (BGR)
:returns: images in the same order (and with the same multiplicity) as
- ``timepoints_ms``
+ ``timepoints_ms``, with ``None`` where no image is available
:rtype: list
"""
original_timepoints = list(timepoints_ms)
@@ -190,17 +342,33 @@ def extract_images_from_timepoints(
Image = _check_cv_dep('PIL.Image') if as_PIL else None
container = open_container(video_document)
+ path = video_document.location_path()
result_map = {}
+ decode_errors: List[str] = []
try:
stream = container.streams.video[0]
time_base = float(stream.time_base)
+ rate = _stream_rate(stream)
+ # Tolerance for a target that falls outside the decoded frames. A
+ # frame covers the interval up to the next one, so a target within
+ # that span still belongs to it; this is the normal case for the last
+ # frame of a healthy video, whose header duration usually runs a
+ # frame longer than the frame's own PTS. `frame.duration` is used when
+ # the container provides it, with the stream rate as the fallback.
+ default_tolerance_ticks = (round(1 / float(rate) / time_base)
+ if rate else 0)
+
+ def _tolerance_ticks(frame):
+ return frame.duration or default_tolerance_ticks
+
# convert each target ms to stream ticks (PTS units)
- target_ticks = [int(round(t_ms / 1000.0 / time_base))
+ target_ticks = [round(t_ms / 1000.0 / time_base)
for t_ms in unique_sorted_ms]
# seek to the nearest keyframe at or before the earliest target
- container.seek(target_ticks[0], backward=True, any_frame=False,
- stream=stream)
+ with _ffmpeg_op('seek in', video_document, path):
+ container.seek(target_ticks[0], backward=True, any_frame=False,
+ stream=stream)
targets = iter(zip(unique_sorted_ms, target_ticks))
cur_ms, cur_pts = next(targets, (None, None))
@@ -211,13 +379,19 @@ def _emit(frame, t_ms):
result_map[t_ms] = (frame.to_image() if as_PIL
else frame.to_ndarray(format='bgr24'))
- for frame in container.decode(stream):
+ for frame in _decode_frames(container, stream, decode_errors):
if frame.pts is None:
continue
pts = frame.pts
while cur_ms is not None and pts >= cur_pts:
- # pick whichever of (prev, current) is closer to target
- if prev_pts is None or (pts - cur_pts) <= (cur_pts - prev_pts):
+ if prev_pts is None:
+ # nothing decoded before the target, which happens when
+ # the seek lands after it or the start of the stream is
+ # missing; the same distance limit applies as at the end
+ if pts - cur_pts <= _tolerance_ticks(frame):
+ _emit(frame, cur_ms)
+ # otherwise pick whichever of (prev, current) is closer
+ elif (pts - cur_pts) <= (cur_pts - prev_pts):
_emit(frame, cur_ms)
else:
_emit(prev_frame, cur_ms)
@@ -227,24 +401,24 @@ def _emit(frame, t_ms):
if cur_ms is None:
break
- # targets past the last decoded frame: fall back to the last frame
+ # Targets past the last decoded frame belong to that frame only while
+ # they fall within its duration; anything further has no image, and is
+ # left out of `result_map` to become a `None` entry below.
while cur_ms is not None:
- if prev_frame is not None:
- warnings.warn(
- f'Timepoint {cur_ms}ms is beyond the video duration; '
- f'returning the last decoded frame for {video_document.id}.'
- )
+ if (prev_frame is not None
+ and cur_pts - prev_pts <= _tolerance_ticks(prev_frame)):
_emit(prev_frame, cur_ms)
- else:
- warnings.warn(
- f'No frames decoded for timepoint {cur_ms}ms from '
- f'video {video_document.id}.'
- )
cur_ms, cur_pts = next(targets, (None, None))
finally:
container.close()
- return [result_map[t] for t in original_timepoints if t in result_map]
+ missing_ms = [t for t in unique_sorted_ms if t not in result_map]
+ if missing_ms:
+ n_missing_positions = sum(1 for t in original_timepoints
+ if t not in result_map)
+ _warn_missing_images(video_document, path, missing_ms,
+ n_missing_positions, decode_errors)
+ return [result_map.get(t) for t in original_timepoints]
def _tp_ids_to_timepoints_ms(mmif: Mmif, tp_ids: List[str]) -> List[int]:
@@ -407,7 +581,9 @@ def extract_images_by_count_with_sources(
:param as_PIL: return :py:class:`~PIL.Image.Image` instead of
:py:class:`~numpy.ndarray`
:return: tuple of (list of images, list of selected target TP IDs);
- the two lists are parallel
+ the two lists are parallel, and an image is ``None`` when none is
+ available for that timepoint (see
+ :py:func:`extract_images_from_timepoints`)
:rtype: tuple
"""
if 'targets' not in annotation.properties:
@@ -460,7 +636,8 @@ def extract_images_by_count(
:param fraction: fraction of targets to include (ideally)
:param as_PIL: return :py:class:`~PIL.Image.Image` instead of
:py:class:`~numpy.ndarray`
- :return: list of images
+ :return: list of images, with ``None`` where none is available (see
+ :py:func:`extract_images_from_timepoints`)
:rtype: list
"""
images, _ = extract_images_by_count_with_sources(
@@ -496,8 +673,10 @@ def extract_images_by_mode_with_sources(
:param as_PIL: return :py:class:`PIL.Image.Image` instead of
:py:class:`~numpy.ndarray`
:return: tuple of (list of images, list of sources); the two lists
- are parallel. May be ``([], [])`` for ``REPRESENTATIVES`` mode
- when no representatives exist.
+ are parallel, and an image is ``None`` when none is available for
+ that timepoint (see :py:func:`extract_images_from_timepoints`).
+ May be ``([], [])`` for ``REPRESENTATIVES`` mode when no
+ representatives exist.
:rtype: tuple
"""
if mode is None:
@@ -540,8 +719,9 @@ def extract_images_by_mode(
default
:param as_PIL: return :py:class:`PIL.Image.Image` instead of
:py:class:`~numpy.ndarray`
- :return: list of images (may be empty for ``REPRESENTATIVES`` mode
- when no representatives exist)
+ :return: list of images, with ``None`` where none is available (see
+ :py:func:`extract_images_from_timepoints`); may be empty for
+ ``REPRESENTATIVES`` mode when no representatives exist
:rtype: list
"""
images, _ = extract_images_by_mode_with_sources(
diff --git a/pyproject.toml b/pyproject.toml
index 32985ec9..97611807 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,7 +39,7 @@ source = "https://github.com/clamsproject/mmif-python"
mmif-spec = "https://mmif.clams.ai/1.2.0"
[project.optional-dependencies]
-cv = ["pillow", "opencv-python", "ffmpeg-python", "wurlitzer", "av"]
+cv = ["pillow", "opencv-python", "ffmpeg-python", "wurlitzer", "av>=15"]
seq = ["numpy"]
docs = ["sphinx>=7.0,<8.0", "furo", "m2r2", "autodoc-pydantic"]
test = [
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.py b/tests/test_utils.py
index 2bae378f..f4f6e185 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -3,6 +3,7 @@
import pathlib
import tempfile
import unittest
+import warnings
from pathlib import Path
from unittest import mock
@@ -13,6 +14,7 @@
from mmif import (
AnnotationTypes,
Document,
+ DocumentTypes,
Mmif
)
from mmif.utils import sequence_helper as sqh
@@ -45,13 +47,14 @@ def test_convert(self, value):
self.assertAlmostEqual(value, out_t, delta=1000/self.FPS) # up to 1 frame error
+@pytest.mark.filterwarnings('ignore::DeprecationWarning')
class TestVideoDocumentHelper(unittest.TestCase):
def setUp(self):
self.fps = 29.97
self.mmif_obj = Mmif(validate=False)
self.a_view = self.mmif_obj.new_view()
self.video_doc = Document({
- "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1",
+ "@type": DocumentTypes.VideoDocument,
"properties": {
"mime": "video",
"id": "d1",
@@ -61,7 +64,7 @@ def setUp(self):
self.video_doc.add_property('fps', self.fps)
self.mmif_obj.add_document(self.video_doc)
- def test_extract_mid_frame(self):
+ def test_get_mid_framenum(self):
tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=100, end=200, timeUnit='frame', document=self.video_doc.id)
self.assertEqual(150, vdh.get_mid_framenum(self.mmif_obj, tf))
tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=0, end=200, timeUnit='frame', document=self.video_doc.id)
@@ -69,8 +72,7 @@ def test_extract_mid_frame(self):
tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=0, end=3, timeUnit='seconds', document=self.video_doc.id)
self.assertEqual(vdh.convert(1.5, 's', 'f', self.fps), vdh.get_mid_framenum(self.mmif_obj, tf))
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
- def test_extract_representative_frame(self):
+ def test_get_representative_framenum(self):
tp = self.a_view.new_annotation(AnnotationTypes.TimePoint, timePoint=1500, timeUnit='milliseconds', document=self.video_doc.id)
tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=1000, end=2000, timeUnit='milliseconds', document=self.video_doc.id)
tf.add_property('representatives', [tp.id])
@@ -88,23 +90,26 @@ def test_extract_representative_frame(self):
def test_get_framerate(self):
self.assertAlmostEqual(29.97, vdh.get_framerate(self.video_doc), places=0)
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
- def test_frames_to_seconds(self):
- self.assertAlmostEqual(3.337, vdh.framenum_to_second(self.video_doc, 100), places=0)
-
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
- def test_frames_to_milliseconds(self):
- self.assertAlmostEqual(3337.0, vdh.framenum_to_millisecond(self.video_doc, 100), places=0)
-
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
- def test_seconds_to_frames(self):
- self.assertAlmostEqual(100, vdh.second_to_framenum(self.video_doc, 3.337), places=0)
-
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
- def test_milliseconds_to_frames(self):
- self.assertAlmostEqual(100, vdh.millisecond_to_framenum(self.video_doc, 3337.0), places=0)
+ def test_get_framerate_without_cached_property(self):
+ # without an fps-like property, the rate comes from the container
+ fixture = pathlib.Path(__file__).parent / 'black-2997fps.mp4'
+ vd = Document({
+ "@type": DocumentTypes.VideoDocument,
+ "properties": {"mime": "video", "id": "r1",
+ "location": f"file://{fixture}"},
+ })
+ self.assertAlmostEqual(29.97, vdh.get_framerate(vd), places=1)
+ # any of the accepted spellings is honored without opening the file
+ for key in ('frameRate', 'frames_per_second', 'frame-per-second'):
+ with self.subTest(key=key):
+ aliased = Document({
+ "@type": DocumentTypes.VideoDocument,
+ "properties": {"mime": "video", "id": "r2",
+ "location": "file:///nonexistent.mp4",
+ key: 12.5},
+ })
+ self.assertEqual(12.5, vdh.get_framerate(aliased))
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
def test_convert_roundtrip(self):
# ms for 1 frame
tolerance = 1000 / self.video_doc.get_property('fps')
@@ -113,7 +118,6 @@ def test_convert_roundtrip(self):
m2f2m = vdh.framenum_to_millisecond(self.video_doc, m2f)
self.assertAlmostEqual(ms, m2f2m, delta=tolerance)
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
def test_sample_frames(self):
s_frame = vdh.second_to_framenum(self.video_doc, 3)
e_frame = vdh.second_to_framenum(self.video_doc, 5.5)
@@ -147,7 +151,6 @@ def test_convert_timeframe(self):
for times in zip((3.337, 6.674), vdh.convert_timeframe(self.mmif_obj, timeframe_ann, 's')):
self.assertAlmostEqual(*times, places=0)
- @pytest.mark.filterwarnings('ignore::DeprecationWarning')
def test_extract_frames_as_images(self):
frame_list = [5, 10, 15]
target_images = vdh.extract_frames_as_images(self.video_doc, frame_list, as_PIL=False)
@@ -163,45 +166,79 @@ def test_extract_frames_as_images(self):
frame_list.append(tot_fcount + 1)
new_target_images = vdh.extract_frames_as_images(self.video_doc, frame_list, as_PIL=False)
self.assertEqual(4, len(frame_list))
+ # the deprecated cv2 path keeps dropping unavailable frames, so its
+ # list shortens where the current API would hold a `None` instead
self.assertEqual(3, len(new_target_images))
- def test_open_container(self):
- # open_container sets fps/frameCount/duration as informational props
+ def test_deprecated_capture(self):
+ fixture = pathlib.Path(__file__).parent / 'black-2997fps.mp4'
vd = Document({
- "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1",
- "properties": {
- "mime": "video",
- "id": "o1",
- "location": f"file://{pathlib.Path(__file__).parent}/black-2997fps.mp4",
- }
+ "@type": DocumentTypes.VideoDocument,
+ "properties": {"mime": "video", "id": "c1",
+ "location": f"file://{fixture}"},
})
- c = vdh.open_container(vd)
+ with pytest.warns(DeprecationWarning, match='open_container'):
+ capture = vdh.capture(vd)
try:
self.assertAlmostEqual(29.97, vd.get_property('fps'), places=1)
- self.assertGreater(vd.get_property('frameCount'), 0)
self.assertGreater(vd.get_property('duration'), 0)
finally:
- c.close()
+ capture.release()
+
+ def test_deprecated_frame_extractors(self):
+ tp = self.a_view.new_annotation(
+ AnnotationTypes.TimePoint, timePoint=1500,
+ timeUnit='milliseconds', document=self.video_doc.id)
+ tf = self.a_view.new_annotation(
+ AnnotationTypes.TimeFrame, start=1000, end=2000,
+ timeUnit='milliseconds', document=self.video_doc.id)
+ tf.add_property('representatives', [tp.id])
+ self.assertIsNotNone(vdh.extract_mid_frame(self.mmif_obj, tf))
+ self.assertIsNotNone(
+ vdh.extract_representative_frame(self.mmif_obj, tf))
+ self.assertEqual(
+ [vdh.millisecond_to_framenum(self.video_doc, 1500)],
+ vdh.get_representative_framenums(self.mmif_obj, tf))
- def test_open_container_matroska(self):
- # regression for issue #390: Matroska/webm leave the per-stream
- # `frames`/`duration` fields unset, so open_container must fall back to
- # the container-level duration instead of caching frameCount/duration 0.
+ def test_open_container(self):
+ # open_container sets fps/frameCount/duration as informational props.
+ # The webm case is a regression for issue #390: Matroska leaves the
+ # per-stream `frames`/`duration` fields unset, so open_container must
+ # fall back to the container-level duration instead of caching 0.
+ for fixture in ('black-2997fps.mp4', 'black-2997fps.webm'):
+ with self.subTest(fixture=fixture):
+ path = pathlib.Path(__file__).parent / fixture
+ vd = Document({
+ "@type": DocumentTypes.VideoDocument,
+ "properties": {"mime": "video", "id": "o1",
+ "location": f"file://{path}"},
+ })
+ c = vdh.open_container(vd)
+ try:
+ self.assertAlmostEqual(29.97, vd.get_property('fps'),
+ places=1)
+ self.assertGreater(vd.get_property('frameCount'), 0)
+ self.assertGreater(vd.get_property('duration'), 0)
+ finally:
+ c.close()
+
+ def test_open_container_rejects_non_video(self):
+ for bad_doc in (None, Document({
+ "@type": "http://mmif.clams.ai/vocabulary/TextDocument/v1",
+ "properties": {"mime": "text", "id": "t1"}})):
+ with self.subTest(doc=bad_doc):
+ with pytest.raises(ValueError):
+ vdh.open_container(bad_doc)
+
+ def test_open_container_missing_file(self):
+ # a missing file is not a broken file, and must stay distinguishable
vd = Document({
- "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1",
- "properties": {
- "mime": "video",
- "id": "o1",
- "location": f"file://{pathlib.Path(__file__).parent}/black-2997fps.webm",
- }
+ "@type": DocumentTypes.VideoDocument,
+ "properties": {"mime": "video", "id": "m1",
+ "location": "file:///nonexistent-video.mp4"},
})
- c = vdh.open_container(vd)
- try:
- self.assertAlmostEqual(29.97, vd.get_property('fps'), places=1)
- self.assertGreater(vd.get_property('frameCount'), 0)
- self.assertGreater(vd.get_property('duration'), 0)
- finally:
- c.close()
+ with pytest.raises(FileNotFoundError):
+ vdh.open_container(vd)
def test_sample_timepoints(self):
# half-open interval; step in ms
@@ -216,18 +253,31 @@ def test_sample_timepoints(self):
vdh.sample_timepoints(0, 100, -10)
def test_extract_images_from_timepoints(self):
- # basic: three distinct timepoints
+ # in-range timepoints all yield an image, and nothing is reported
ms_list = [1000, 2000, 3000]
- imgs = vdh.extract_images_from_timepoints(
- self.video_doc, ms_list, as_PIL=False)
- self.assertEqual(3, len(imgs))
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter('always')
+ imgs = vdh.extract_images_from_timepoints(
+ self.video_doc, ms_list, as_PIL=False)
+ self.assertEqual(len(ms_list), len(imgs))
+ self.assertTrue(all(img is not None for img in imgs))
+ self.assertEqual([], [w for w in caught
+ if 'No image is available' in str(w.message)])
# empty input
self.assertEqual(
[], vdh.extract_images_from_timepoints(self.video_doc, []))
- # duplicates preserved in input order
+ # duplicates yield the same image object at each repeated position
dup_ms = [500, 250, 500, 750, 250]
dup_imgs = vdh.extract_images_from_timepoints(self.video_doc, dup_ms)
- self.assertEqual(5, len(dup_imgs))
+ self.assertEqual(len(dup_ms), len(dup_imgs))
+ self.assertIs(dup_imgs[0], dup_imgs[2])
+ self.assertIs(dup_imgs[1], dup_imgs[4])
+
+ def test_extract_images_as_PIL(self):
+ Image = pytest.importorskip('PIL.Image')
+ imgs = vdh.extract_images_from_timepoints(
+ self.video_doc, [1000], as_PIL=True)
+ self.assertIsInstance(imgs[0], Image.Image)
def _make_timepoints(self, count):
# Explicit aid avoids a pre-existing clams-vocabulary / mmif-python
@@ -418,7 +468,7 @@ def test_pts_offset_regression(self):
import av
fixture = pathlib.Path(__file__).parent / 'testsrc-2997fps-ptsoffset.mp4'
vd = Document({
- "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1",
+ "@type": DocumentTypes.VideoDocument,
"properties": {
"mime": "video",
"id": "p1",
@@ -463,6 +513,324 @@ def test_pts_offset_regression(self):
'PTS-offset videos')
+class TestBrokenVideoDocument(unittest.TestCase):
+ """
+ Covers videos that hold fewer images than their container metadata
+ promises (issue #400). A byte-truncated file surfaces differently per
+ container and per cut position, so each path gets its own fixture.
+ """
+
+ fixtures_dir = pathlib.Path(__file__).parent
+
+ @classmethod
+ def setUpClass(cls):
+ # fixtures are byte-identical for every test, so they are built once
+ cls.tmpdir = tempfile.TemporaryDirectory()
+ cls.addClassCleanup(cls.tmpdir.cleanup)
+ tmp = pathlib.Path(cls.tmpdir.name)
+ healthy_mp4 = cls.fixtures_dir / 'black-2997fps.mp4'
+ faststart = tmp / 'faststart.mp4'
+ cls._remux_faststart(healthy_mp4, faststart)
+ cls.cuts = {
+ # cut mid-packet: the mp4 demuxer hands the partial packet to the
+ # decoder, which raises
+ 'unclean.mp4': (cls._truncate(faststart, tmp / 'unclean.mp4'),
+ True),
+ # cut at a packet boundary: a clean EOF, indistinguishable from
+ # the end of a healthy video
+ 'clean.mp4': (cls._truncate(faststart, tmp / 'clean.mp4',
+ cls._clean_cut_offset(faststart)),
+ False),
+ # Matroska drops the incomplete block, so any cut is clean
+ 'cut.webm': (cls._truncate(cls.fixtures_dir / 'black-2997fps.webm',
+ tmp / 'cut.webm'),
+ False),
+ }
+ # `moov` sits at the end of the shipped fixtures, so a cut copy of the
+ # original cannot be opened at all
+ cls.trailing_moov = cls._truncate(healthy_mp4, tmp / 'moovend.mp4')
+ # opens fine, holds no video stream; the realistic case is an audio
+ # file registered as a VideoDocument, and a tiny prefix behaves alike
+ cls.audio_only = cls._write_audio_only(tmp / 'audio.wav')
+ cls.tiny = cls._truncate(faststart, tmp / 'tiny.mp4', 200)
+ cls.healthy = healthy_mp4
+
+ @staticmethod
+ def _doc(path, aid='d1'):
+ return Document({
+ "@type": DocumentTypes.VideoDocument,
+ "properties": {"mime": "video", "id": aid,
+ "location": f"file://{path}"}
+ })
+
+ @staticmethod
+ def _remux_faststart(src, dst):
+ # a cut copy of an mp4 is only decodable at all when the `moov` atom
+ # sits at the front; the shipped fixtures carry it at the end
+ import av
+ faststart = {'movflags': 'faststart'}
+ with av.open(str(src)) as inp, \
+ av.open(str(dst), 'w', options=faststart) as out:
+ in_stream = inp.streams.video[0]
+ try:
+ out_stream = out.add_stream_from_template(in_stream)
+ except AttributeError: # PyAV < 14
+ out_stream = out.add_stream(template=in_stream)
+ for packet in inp.demux(in_stream):
+ if packet.dts is None:
+ continue
+ packet.stream = out_stream
+ out.mux(packet)
+
+ @staticmethod
+ def _write_audio_only(dst):
+ import av
+ import numpy as np
+ rate = 8000
+ with av.open(str(dst), 'w') as out:
+ stream = out.add_stream('pcm_s16le', rate=rate, layout='mono')
+ frame = av.AudioFrame.from_ndarray(
+ np.zeros((1, rate), dtype='int16'), format='s16',
+ layout='mono')
+ frame.sample_rate = rate
+ for packet in stream.encode(frame):
+ out.mux(packet)
+ for packet in stream.encode(None):
+ out.mux(packet)
+ return dst
+
+ @staticmethod
+ def _truncate(src, dst, nbytes=None):
+ data = pathlib.Path(src).read_bytes()
+ dst.write_bytes(data[:nbytes if nbytes else len(data) // 2])
+ return dst
+
+ @staticmethod
+ def _clean_cut_offset(path):
+ # end of the last complete packet before the halfway point
+ import av
+ half = pathlib.Path(path).stat().st_size // 2
+ with av.open(str(path)) as container:
+ stream = container.streams.video[0]
+ return next(p.pos for p in container.demux(stream)
+ if p.pos is not None and p.pos >= half)
+
+ @staticmethod
+ def _last_decodable_ms(path):
+ # ground truth for the assertions below, computed without vdh; the
+ # decoder is flushed after an error, as vdh does, so that frames held
+ # back for reordering count here too
+ import av
+ last = None
+
+ def frames(container, stream):
+ try:
+ for packet in container.demux(stream):
+ yield from packet.decode()
+ except av.error.FFmpegError:
+ try:
+ yield from stream.codec_context.decode(None)
+ except av.error.FFmpegError:
+ pass
+
+ with av.open(str(path)) as container:
+ stream = container.streams.video[0]
+ for frame in frames(container, stream):
+ if frame.pts is not None:
+ pts_ms = round(frame.pts * stream.time_base * 1000)
+ last = max(last or 0, pts_ms)
+ return last
+
+ @staticmethod
+ def _missing_image_warnings(caught):
+ # the module also emits unrelated warnings (e.g. vocabulary version
+ # mismatches), and a single call must report missing images once
+ return [str(w.message) for w in caught
+ if 'No image is available' in str(w.message)]
+
+ def _extract(self, doc, timepoints, **kwargs):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter('always')
+ images = vdh.extract_images_from_timepoints(
+ doc, timepoints, **kwargs)
+ return images, self._missing_image_warnings(caught)
+
+ def _frame_ms(self, doc):
+ return int(1000 / vdh.get_framerate(doc))
+
+ def test_truncated_file_yields_none_past_last_frame(self):
+ import av
+ for name, (path, expect_decode_error) in self.cuts.items():
+ with self.subTest(fixture=name):
+ doc = self._doc(path)
+ last_ok = self._last_decodable_ms(path)
+ self.assertIsNotNone(last_ok,
+ 'fixture should decode at least once')
+ frame_ms = self._frame_ms(doc)
+ # the last decodable frame covers one frame duration, and
+ # nothing beyond it; 2ms of slack absorbs ms/tick rounding
+ timepoints = [0, last_ok, last_ok + frame_ms - 2,
+ last_ok + frame_ms + 2, 59000]
+ images, messages = self._extract(doc, timepoints)
+ self.assertEqual(len(timepoints), len(images))
+ for i in (0, 1, 2):
+ self.assertIsNotNone(images[i], f'position {i}')
+ self.assertIsNone(images[3])
+ self.assertIsNone(images[4])
+ # within the tolerance, the last frame itself is reused
+ self.assertEqual(images[1].tobytes(), images[2].tobytes())
+ self.assertEqual(1, len(messages))
+ self.assertIn(vdh._media_check_hint(str(path)), messages[0])
+ self.assertEqual(
+ expect_decode_error,
+ av.error.InvalidDataError.__name__ in messages[0])
+ self.assertEqual(expect_decode_error,
+ 'first decoding error:' in messages[0])
+
+ def test_missing_only_past_duration_is_not_flagged(self):
+ # a file that merely ends is not suspect, even after a clean cut
+ path, _ = self.cuts['cut.webm']
+ doc = self._doc(path)
+ vdh.open_container(doc).close()
+ duration = doc.get_property('duration')
+ images, messages = self._extract(doc, [duration + 10000])
+ self.assertEqual([None], images)
+ self.assertEqual(1, len(messages))
+ self.assertNotIn(vdh._media_check_hint(str(path)), messages[0])
+
+ def test_missing_before_and_past_duration_is_flagged(self):
+ # the earliest missing timepoint decides, and both are listed
+ path, _ = self.cuts['cut.webm']
+ doc = self._doc(path)
+ vdh.open_container(doc).close()
+ duration = doc.get_property('duration')
+ last_ok = self._last_decodable_ms(path)
+ timepoints = [last_ok + 5000, duration + 10000]
+ images, messages = self._extract(doc, timepoints)
+ self.assertEqual([None, None], images)
+ self.assertEqual(1, len(messages))
+ self.assertIn(vdh._media_check_hint(str(path)), messages[0])
+ self.assertIn(f'{timepoints[0]}ms, {timepoints[1]}ms', messages[0])
+
+ def test_repeated_missing_timepoint_counts_positions(self):
+ path, _ = self.cuts['cut.webm']
+ images, messages = self._extract(self._doc(path), [59000] * 3)
+ self.assertEqual([None, None, None], images)
+ self.assertIn('for 3 timepoint(s)', messages[0])
+
+ def test_missing_entry_with_as_PIL(self):
+ Image = pytest.importorskip('PIL.Image')
+ path, _ = self.cuts['cut.webm']
+ images, _ = self._extract(self._doc(path), [0, 59000], as_PIL=True)
+ self.assertIsInstance(images[0], Image.Image)
+ self.assertIsNone(images[1])
+
+ def test_tolerance_falls_back_to_stream_rate(self):
+ # a frame may carry no duration of its own, in which case the same
+ # tolerance must come from the stream frame rate
+ class _NoDuration:
+ def __init__(self, frame):
+ self._frame = frame
+ self.duration = None
+
+ def __getattr__(self, name):
+ return getattr(self._frame, name)
+
+ real_decode_frames = vdh._decode_frames
+
+ def stripped(container, stream, decode_errors):
+ for frame in real_decode_frames(container, stream, decode_errors):
+ yield _NoDuration(frame)
+
+ doc = self._doc(self.healthy)
+ vdh.open_container(doc).close()
+ duration = doc.get_property('duration')
+ with mock.patch.object(vdh, '_decode_frames', stripped):
+ images, messages = self._extract(doc, [duration - 1])
+ self.assertIsNotNone(images[0])
+ self.assertEqual([], messages)
+
+ def test_unopenable_files_raise_with_hint(self):
+ cases = {'trailing moov': self.trailing_moov,
+ 'audio only': self.audio_only,
+ 'tiny prefix': self.tiny}
+ for label, path in cases.items():
+ with self.subTest(case=label):
+ with pytest.raises(RuntimeError) as excinfo:
+ vdh.open_container(self._doc(path))
+ self.assertIn(vdh._media_check_hint(str(path)),
+ str(excinfo.value))
+ # get_framerate goes through open_container, so it reports the same way
+ with pytest.raises(RuntimeError):
+ vdh.get_framerate(self._doc(self.trailing_moov))
+
+ def test_failed_seek_raises_with_hint(self):
+ # PyAV containers are immutable extension types, so the failure is
+ # injected through a proxy around the real container
+ import av
+ path, _ = self.cuts['cut.webm']
+ doc = self._doc(path)
+
+ class _SeekFails:
+ def __init__(self, container):
+ self._container = container
+
+ def __getattr__(self, name):
+ return getattr(self._container, name)
+
+ def seek(self, *args, **kwargs):
+ raise av.error.InvalidDataError(1094995529, 'seek failed')
+
+ real_open_container = vdh.open_container
+
+ def wrapped(video_document):
+ return _SeekFails(real_open_container(video_document))
+
+ with mock.patch.object(vdh, 'open_container', wrapped):
+ with pytest.raises(RuntimeError) as excinfo:
+ vdh.extract_images_from_timepoints(doc, [1000])
+ self.assertIn(vdh._media_check_hint(str(path)), str(excinfo.value))
+ self.assertIn('seek failed', str(excinfo.value))
+
+ def test_empty_request_never_opens_the_file(self):
+ doc = self._doc(self.trailing_moov)
+ self.assertEqual([], vdh.extract_images_from_timepoints(doc, []))
+
+ def test_none_stays_parallel_with_sources(self):
+ # the breakage that motivated issue #400: apps pair images with
+ # timepoints by list index
+ mmif_obj = Mmif(validate=False)
+ doc = self._doc(self.healthy)
+ mmif_obj.add_document(doc)
+ vdh.open_container(doc).close()
+ duration = doc.get_property('duration')
+ view = mmif_obj.new_view()
+ tp_ids = []
+ for i, ms in enumerate([1000, duration + 10000, duration + 20000]):
+ tp = view.new_annotation(
+ AnnotationTypes.TimePoint, aid=f'tp_{i}', timePoint=ms,
+ timeUnit='milliseconds', document=doc.id)
+ tp_ids.append(tp.id)
+ time_frame = view.new_annotation(
+ AnnotationTypes.TimeFrame, aid='tf_0', document=doc.id,
+ timeUnit='milliseconds', start=1000, end=duration + 20000,
+ targets=tp_ids, representatives=tp_ids)
+ with warnings.catch_warnings():
+ warnings.simplefilter('ignore')
+ by_mode = vdh.extract_images_by_mode_with_sources(
+ mmif_obj, time_frame, mode=vdh.SamplingMode.REPRESENTATIVES)
+ by_count = vdh.extract_images_by_count_with_sources(
+ mmif_obj, time_frame, fraction=1.0)
+ for label, (images, sources) in {'by_mode': by_mode,
+ 'by_count': by_count}.items():
+ with self.subTest(entrypoint=label):
+ self.assertEqual(tp_ids, list(sources))
+ self.assertEqual(len(sources), len(images))
+ self.assertIsNotNone(images[0])
+ self.assertIsNone(images[1])
+ self.assertIsNone(images[2])
+
+
class TestSequenceHelper(unittest.TestCase):
def test_validate_labelset(self):
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()