diff --git a/.gitignore b/.gitignore index aac88e97..b5524d05 100644 --- a/.gitignore +++ b/.gitignore @@ -56,7 +56,7 @@ hs_err_pid* # virtual machine crash logs, see http://www.java.com/en/download/he **/__pycache__ **/dist **/*.egg-info -**/.pytype +**/.mypy_cache build/** .coverage coverage.xml diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index f6704270..086438b4 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -77,7 +77,7 @@ This means adding a properly structured module within the CLI package is all tha pip install -e ".[dev]" ``` -An editable install (`pip install -e .`) is required before running tests or building docs. The package uses `importlib.metadata` for version resolution at runtime, which only works when the package is registered in the environment. You can no longer run `pytest` or `pytype` directly against the source tree without installing first. If you want to avoid pulling in all dependencies, `pip install -e . --no-deps` is sufficient to register the package metadata. +An editable install (`pip install -e .`) is required before running tests or building docs. The package uses `importlib.metadata` for version resolution at runtime, which only works when the package is registered in the environment. You can no longer run `pytest` or `mypy` directly against the source tree without installing first. If you want to avoid pulling in all dependencies, `pip install -e . --no-deps` is sufficient to register the package metadata. ## Local Development diff --git a/build-tools/clean.py b/build-tools/clean.py index 9a35c493..5a3f270b 100644 --- a/build-tools/clean.py +++ b/build-tools/clean.py @@ -13,7 +13,7 @@ # Directories to remove CLEAN_DIRS = [ "build", "dist", "*.egg-info", "mmif_python-*", - ".pytest_cache", ".pytype", ".hypothesis", + ".pytest_cache", ".mypy_cache", ".hypothesis", "tests/.hypothesis", "htmlcov", "docs-test", ] diff --git a/build-tools/test.py b/build-tools/test.py index fe5455e4..48ae390f 100644 --- a/build-tools/test.py +++ b/build-tools/test.py @@ -4,7 +4,7 @@ This script is equivalent to:: pip install -e ".[test]" - pytype mmif + python -m mypy python -m pytest --cov=mmif --cov-report=xml -m "not slow" """ import argparse @@ -51,10 +51,10 @@ def main(): cwd=project_root, ) - # Run pytype static analysis - print("\n--- Running pytype ---") + # Run mypy static analysis (configured in pyproject.toml) + print("\n--- Running mypy ---") run_command( - ["pytype", "mmif"], + [sys.executable, "-m", "mypy"], cwd=project_root, ) diff --git a/mmif/serialize/__init__.py b/mmif/serialize/__init__.py index 7aebbc94..dca6aa24 100644 --- a/mmif/serialize/__init__.py +++ b/mmif/serialize/__init__.py @@ -3,12 +3,13 @@ """ from .annotation import * -from .annotation import __all__ as anno_all from .mmif import * -from .mmif import __all__ as mmif_all from .producer import * -from .producer import __all__ as producer_all from .view import * -from .view import __all__ as view_all -__all__ = anno_all + mmif_all + producer_all + view_all +# a literal, so that type checkers can follow the star import of this package +__all__ = [ + 'Annotation', 'AnnotationProperties', 'Contain', 'Document', + 'DocumentProperties', 'EmptyOutputWarning', 'Mmif', 'ProducerContext', + 'Text', 'View', 'ViewMetadata', 'canonicalize_prop_name', +] diff --git a/mmif/serialize/annotation.py b/mmif/serialize/annotation.py index a3f47244..a3f8d91d 100644 --- a/mmif/serialize/annotation.py +++ b/mmif/serialize/annotation.py @@ -16,7 +16,7 @@ import typing import warnings from collections.abc import Iterator, MutableMapping -from typing import Optional, TypeVar, Union +from typing import TYPE_CHECKING, Any, TypeVar, cast from urllib.parse import urlparse import mmif_docloc_http @@ -26,6 +26,9 @@ from .. import AnnotationTypes, DocumentTypes from .model import PRMTV_TYPES, MmifObject +if TYPE_CHECKING: + from .view import View + __all__ = ['Annotation', 'AnnotationProperties', 'Document', 'DocumentProperties', 'Text', 'canonicalize_prop_name'] T = TypeVar('T') @@ -84,7 +87,7 @@ class Annotation(MmifObject): MmifObject that represents an annotation in a MMIF view. """ - def __init__(self, anno_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, anno_obj: bytes | str | dict | None = None, *_) -> None: self._type: ThingTypesBase = ThingTypesBase('') # values that are not stored on the annotation itself but cached at # load time: properties recorded by capital `Annotation`s (on @@ -92,8 +95,8 @@ def __init__(self, anno_obj: Optional[Union[bytes, str, dict]] = None, *_) -> No self._props_ephemeral: AnnotationProperties = AnnotationProperties(None, self) # the view holding this annotation; backs `parent` and the lookup of # view-level `contains` defaults - self._parent_view = None - self._alignments = {} # to hold alignment information (Alignment anno long_id -> aligned anno long_id) + self._parent_view: 'View | None' = None + self._alignments: dict = {} # to hold alignment information (Alignment anno long_id -> aligned anno long_id) self.reserved_names.update(('_props_ephemeral', '_parent_view', '_alignments')) if not hasattr(self, 'properties'): # don't overwrite DocumentProperties on super() call self.properties: AnnotationProperties = AnnotationProperties(None, self) @@ -123,7 +126,7 @@ def _cache_alignment(self, alignment_ann: 'Annotation', alignedto_ann: 'Annotati """ self._alignments[alignment_ann] = alignedto_ann - def aligned_to_by(self, alignment: 'Annotation') -> 'Optional[Annotation]': + def aligned_to_by(self, alignment: 'Annotation') -> 'Annotation | None': """ Retrieves the other side of ``Alignment`` annotation that has this annotation on one side. @@ -175,7 +178,7 @@ def _resolved_properties(self) -> dict[str, typing.Any]: :return: a new dict of canonical property names to values """ - resolved = {} + resolved: dict = {} if self._parent_view is not None: defaults = self._parent_view.metadata.contains.get(self._type) if defaults: @@ -184,7 +187,7 @@ def _resolved_properties(self) -> dict[str, typing.Any]: resolved.update(self.properties.items()) return resolved - def is_type(self, at_type: Union[str, ThingTypesBase]) -> bool: + def is_type(self, at_type: str | ThingTypesBase) -> bool: """ Check if the @type of this object matches. """ @@ -197,7 +200,7 @@ def at_type(self) -> ThingTypesBase: return self._type @at_type.setter - def at_type(self, at_type: Union[str, ThingTypesBase]) -> None: + def at_type(self, at_type: str | ThingTypesBase) -> None: if isinstance(at_type, str): self._type = ThingTypesBase.from_str(at_type) elif isinstance(at_type, type) and issubclass(at_type, ThingTypesBase): @@ -208,8 +211,9 @@ def at_type(self, at_type: Union[str, ThingTypesBase]) -> None: # Use ``model_construct()`` to skip Pydantic field validation — # direct ``__init__`` rejects types with required fields # (e.g., ``Alignment`` requires ``source``/``target``). - instance = at_type.model_construct() - instance.initialized_from = at_type.uri + vocab_type: Any = at_type + instance = vocab_type.model_construct() + instance.initialized_from = vocab_type.uri self._type = instance else: self._type = at_type @@ -282,7 +286,7 @@ def _short_id(self) -> str: @staticmethod def check_prop_value_is_simple_enough( - value: Union[PRMTV_TYPES, LIST_PRMTV, LIST_LIST_PRMTV, DICT_PRMTV, DICT_LIST_PRMTV]) -> bool: + value: PRMTV_TYPES | LIST_PRMTV | LIST_LIST_PRMTV | DICT_PRMTV | DICT_LIST_PRMTV) -> bool: def json_primitives(x): return isinstance(x, typing.get_args(PRMTV_TYPES)) @@ -299,7 +303,7 @@ def json_primitives_list_of_list(x): or (isinstance(value, dict) and all(map(lambda x: isinstance(x[0], str) and (json_primitives(x[1]) or json_primitives_list(x[1])), value.items()))) def add_property(self, name: str, - value: Union[PRMTV_TYPES, LIST_PRMTV, LIST_LIST_PRMTV, DICT_PRMTV, DICT_LIST_PRMTV]) -> None: + value: PRMTV_TYPES | LIST_PRMTV | LIST_LIST_PRMTV | DICT_PRMTV | DICT_LIST_PRMTV) -> None: """ Adds a property to the annotation's properties. A deprecated spelling of an aliased property (e.g. ``frameType``) is stored under its @@ -410,7 +414,7 @@ class Document(Annotation): #: the spellings of the ``text`` property in :meth:`add_property` _TEXT_PROP_NAMES = frozenset({'text', '@text', '_text'}) - def __init__(self, doc_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, doc_obj: bytes | str | dict | None = None, *_) -> None: # see https://github.com/clamsproject/mmif-python/issues/226 for discussion # around the property stores of a document # (names changed since, `existing` >> `ephemeral` and `temporary` >> `pending`) @@ -424,14 +428,14 @@ def __init__(self, doc_obj: Optional[Union[bytes, str, dict]] = None, *_) -> Non self._props_pending: AnnotationProperties = AnnotationProperties(None, self) self.reserved_names.add('_props_pending') - self._type: Union[ThingTypesBase, DocumentTypesBase] = ThingTypesBase('') + self._type: ThingTypesBase | DocumentTypesBase = ThingTypesBase('') self.properties: DocumentProperties = DocumentProperties(None, self) self.disallow_additional_properties() self._attribute_classes = {'properties': DocumentProperties} super().__init__(doc_obj) def add_property(self, name: str, - value: Union[PRMTV_TYPES, LIST_PRMTV] + value: PRMTV_TYPES | LIST_PRMTV | LIST_LIST_PRMTV | DICT_PRMTV | DICT_LIST_PRMTV ) -> None: """ Adds a property to the document's properties. @@ -470,7 +474,7 @@ def add_property(self, name: str, # that already existed refuse `add_property`; "already existed" is # defined by the producer context if name in self._TEXT_PROP_NAMES: - text = Text(value) + text = Text(cast(str | dict, value)) current = self.properties.text if (self.location is not None or current.value) and text != current: raise self._read_only_error('text') @@ -481,11 +485,12 @@ def add_property(self, name: str, raise self._read_only_error(name) self.properties.mime = mime elif name == "location": + location = cast(str, value) requested = DocumentProperties() - requested.location = value + requested.location = location if self.location is not None and self.location != requested.location: raise self._read_only_error(name) - self.location = value + self.location = location elif self.check_prop_value_is_simple_enough(value): self._props_pending[name] = value else: @@ -603,7 +608,7 @@ def text_language(self, lang_code: str) -> None: def text_value(self) -> str: if self._type == DocumentTypes.TextDocument: if self.location: - f = open(self.location_path(nonexist_ok=False), 'r', encoding='utf8') + f = open(cast(str, self.location_path(nonexist_ok=False)), 'r', encoding='utf8') textvalue = f.read() f.close() return textvalue @@ -620,7 +625,7 @@ def text_value(self, text_value: str) -> None: raise ValueError("Only TextDocument can have `text` field.") @property - def location(self) -> Optional[str]: + def location(self) -> str | None: """ ``location`` property must be a legitimate URI. That is, should the document be a local file then the file:// scheme must be used. @@ -632,21 +637,21 @@ def location(self) -> Optional[str]: def location(self, location: str) -> None: self.properties.location = location - def location_scheme(self) -> Optional[str]: + def location_scheme(self) -> str | None: """ Retrieves URI scheme of the document location. Returns None when no location is set. """ return self.properties.location_scheme() - def location_address(self) -> Optional[str]: + def location_address(self) -> str | None: """ Retrieves the full address from the document location URI. Returns None when no location is set. """ return self.properties.location_address() - def location_path(self, nonexist_ok=True) -> Optional[str]: + def location_path(self, nonexist_ok=True) -> str | None: """ Retrieves a path that's resolved to a pathname in the local file system. To obtain the original value of the "path" part in the location string @@ -665,6 +670,7 @@ class AnnotationProperties(MmifObject, MutableMapping[str, T]): :param mmif_obj: the JSON data that defines the properties """ + _unnamed_attributes: dict def __delitem__(self, key: str) -> None: key = self._canonical(key) @@ -718,8 +724,8 @@ def __setitem__(self, key: str, value) -> None: DeprecationWarning) super().__setitem__(canonical, value) - def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, - parent: 'Optional[Annotation]' = None, *_) -> None: + def __init__(self, mmif_obj: bytes | str | dict | None = None, + parent: 'Annotation | None' = None, *_) -> None: self.id: str = '' # any individual at_type (subclassing this class) can have its own set of required attributes self._required_attributes = ["id"] @@ -745,7 +751,7 @@ def _deserialize(self, input_dict: dict) -> None: """ canonical_dict = {} # canonical name -> the deprecated spelling it was read from - from_alias = {} + from_alias: dict = {} for key, value in input_dict.items(): canonical = self._canonical(key) if canonical in canonical_dict and (canonical != key or canonical in from_alias): @@ -769,8 +775,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, - parent: 'Optional[Annotation]' = None, *_) -> None: + def __init__(self, mmif_obj: bytes | str | dict | None = None, + parent: 'Annotation | None' = 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 @@ -813,7 +819,7 @@ def text_value(self, s: str) -> None: self.text.value = s @property - def location(self) -> Optional[str]: + def location(self) -> str | None: """ ``location`` property must be a legitimate URI. That is, should the document be a local file then the file:// scheme must be used. @@ -829,7 +835,7 @@ def location(self, location: str) -> None: else: self.location_ = location - def location_scheme(self) -> Optional[str]: + def location_scheme(self) -> str | None: """ Retrieves URI scheme of the document location. Returns None when no location is set. @@ -838,7 +844,7 @@ def location_scheme(self) -> Optional[str]: return None return urlparse(self.location).scheme - def location_address(self) -> Optional[str]: + def location_address(self) -> str | None: """ Retrieves the full address from the document location URI. Returns None when no location is set. @@ -851,7 +857,7 @@ def location_address(self) -> Optional[str]: else: return "".join((parsed_location.netloc, parsed_location.path)) - def location_path(self) -> Optional[str]: + def location_path(self) -> str | None: """ .. deprecated:: 1.0.2 Will be removed in 2.0.0. Use :meth:`location_path_resolved` @@ -863,7 +869,7 @@ def location_path(self) -> Optional[str]: DeprecationWarning) return self.location_path_resolved() - def location_path_resolved(self, nonexist_ok=True) -> Optional[str]: + def location_path_resolved(self, nonexist_ok=True) -> str | None: """ Retrieves only path name of the document location (hostname is ignored), and then try to resolve the path name in the local file system. @@ -887,7 +893,7 @@ def location_path_resolved(self, nonexist_ok=True) -> Optional[str]: else: return p - def location_path_literal(self) -> Optional[str]: + def location_path_literal(self) -> str | None: """ Retrieves only path name of the document location (hostname is ignored). Returns None when no location is set. @@ -899,7 +905,7 @@ def location_path_literal(self) -> Optional[str]: class Text(MmifObject): - def __init__(self, text_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, text_obj: bytes | str | dict | None = None, *_) -> None: self._value: str = '' self._language: str = '' self.disallow_additional_properties() diff --git a/mmif/serialize/mmif.py b/mmif/serialize/mmif.py index 5994974c..632e1c61 100644 --- a/mmif/serialize/mmif.py +++ b/mmif/serialize/mmif.py @@ -16,7 +16,7 @@ from collections import defaultdict from collections.abc import Iterable, Iterator from datetime import datetime, timezone -from typing import Optional, Union, cast +from typing import cast from urllib.parse import urlparse import jsonschema.validators @@ -43,7 +43,7 @@ class MmifMetadata(MmifObject): :param metadata_obj: the JSON data """ - def __init__(self, metadata_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, metadata_obj: bytes | str | dict | None = None, *_) -> None: # TODO (krim @ 10/7/20): there could be a better name and a better way to give a value to this self.mmif: str = f"http://mmif.clams.ai/{mmif.__specver__}" self._required_attributes = ["mmif"] @@ -57,7 +57,7 @@ class DocumentsList(DataList[Document]): """ _items: dict[str, Document] - def _deserialize(self, input_list: list) -> None: # pytype: disable=signature-mismatch + def _deserialize(self, input_list: list) -> None: # type: ignore[override] """ Extends base ``_deserialize`` method to initialize ``items`` as a dict from document IDs to :class:`mmif.serialize.document.Document` objects. @@ -93,12 +93,12 @@ class ViewsList(DataList[View]): """ _items: dict[str, View] - def __init__(self, mmif_obj: Optional[Union[bytes, str, list]] = None, parent_mmif=None, *_): + def __init__(self, mmif_obj: bytes | str | list | None = None, parent_mmif=None, *_): self._parent_mmif = parent_mmif self.reserved_names.update(("_parent_mmif", "_id_counts")) super().__init__(mmif_obj) - def _deserialize(self, input_list: list) -> None: # pytype: disable=signature-mismatch + def _deserialize(self, input_list: list) -> None: # type: ignore[override] """ Extends base ``_deserialize`` method to initialize ``items`` as a dict from view IDs to :class:`mmif.serialize.view.View` objects. @@ -127,7 +127,7 @@ def append(self, value: View, overwrite=False) -> None: """ super()._append_with_key(value.id, value, overwrite) - def get_last_contentful_view(self) -> Optional[View]: + def get_last_contentful_view(self) -> View | None: """ Returns the last view that is contentful, i.e., has at least one annotation (see :meth:`~mmif.serialize.view.View.has_content`). @@ -139,14 +139,15 @@ def get_last_contentful_view(self) -> Optional[View]: return view return None - def get_last_view(self) -> Optional[View]: + def get_last_view(self) -> View | None: """ Returns the last view appended. """ if self._items: return self._items[list(self._items.keys())[-1]] - - def get_last(self) -> Optional[View]: + return None + + def get_last(self) -> View | None: """ .. deprecated:: 1.0.18 Will be removed in 2.0.0. Use :meth:`get_last_contentful_view` instead. @@ -200,12 +201,12 @@ class Mmif(MmifObject): 'mime', 'location', 'location_', 'text', }) - def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *, validate: bool = True) -> None: + def __init__(self, mmif_obj: bytes | str | dict | None = None, *, validate: bool = True) -> None: self.metadata: MmifMetadata = MmifMetadata() self.documents: DocumentsList = DocumentsList() self.views: ViewsList = ViewsList() # the write mode, see `set_producer()` - self._producer: Optional[ProducerContext] = None + self._producer: ProducerContext | None = None self.reserved_names.add('_producer') if validate and mmif_obj is not None: self.validate(mmif_obj) @@ -219,7 +220,7 @@ def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *, valida super().__init__(mmif_obj) @staticmethod - def validate(json_str: Union[bytes, str, dict]) -> None: + def validate(json_str: bytes | str | dict) -> None: """ Validates a MMIF JSON object against the MMIF Schema. Note that this method operates before processing by MmifObject._load_str, @@ -244,8 +245,8 @@ def serialize(self, include_context: bool = True, *, sanitize: bool = False, - autogenerate_capital_annotations: Optional[bool] = None, - factor_out_shared_properties: Optional[bool] = None) -> str: + autogenerate_capital_annotations: bool | None = None, + factor_out_shared_properties: bool | None = None) -> str: """ Serializes the MMIF object to a JSON string. @@ -547,12 +548,13 @@ def _plan_capital_annotations(self) -> list[tuple[View, str, dict, bool]]: # to avoid duplicate property recording, this will be populated with # existing Annotation objects from all existing views - existing_anns = defaultdict(lambda: defaultdict(dict)) + existing_anns: defaultdict[str, defaultdict[str, dict]] = defaultdict( + lambda: defaultdict(dict)) # ideally, if we can "de-duplicate" props at `add_property()` time, that'd be more efficient, # but that is impossible without looking for the target `document` across other views and top documents list # new properties to record in the current serialization call - anns_to_write = defaultdict(dict) + anns_to_write: defaultdict[str, dict] = defaultdict(dict) for view in self.views: for ann in view.get_annotations(AnnotationTypes.Annotation): # values recorded at view level (`contains`) count too; @@ -732,14 +734,14 @@ def add_document(self, document: Document, overwrite=False) -> None: raise KeyError(f"{document.id} already exists in the views list. ") self.documents.append(document, overwrite) - def get_documents_in_view(self, vid: Optional[str] = None) -> list[Document]: + def get_documents_in_view(self, vid: str | None = None) -> list[Document]: """ Method to get all documents object queries by a view id. :param vid: the source view ID to search for :return: a list of documents matching the requested source view ID, or an empty list if the view not found """ - view = self.views.get(vid) + view = None if vid is None else self.views.get(vid) if view is not None: return view.get_documents() else: @@ -788,7 +790,7 @@ def get_documents_by_property(self, prop_key: str, prop_value: str) -> list[Docu docs.extend([document for document in self.documents if document[prop_key] == prop_value]) return docs - def get_documents_locations(self, m_type: Union[DocumentTypesBase, str], path_only=False) -> list[Optional[str]]: + def get_documents_locations(self, m_type: DocumentTypesBase | str, path_only=False) -> list[str | None]: """ This method returns the file paths of documents of given type. Only top-level documents have locations, so we only check them. @@ -803,7 +805,7 @@ def get_documents_locations(self, m_type: Union[DocumentTypesBase, str], path_on else: return [doc.location for doc in docs] - def get_document_location(self, m_type: Union[DocumentTypesBase, str], path_only=False) -> Optional[str]: + def get_document_location(self, m_type: DocumentTypesBase | str, path_only=False) -> str | None: """ Method to get the location of *first* document of given type. @@ -853,7 +855,7 @@ def get_view_by_id(self, view_id: str) -> View: raise KeyError(f"View with ID {view_id} not found in the MMIF object.") return cast(View, view_found) - def get_alignments(self, at_type1: Union[str, ThingTypesBase], at_type2: Union[str, ThingTypesBase]) -> dict[str, list[Annotation]]: + def get_alignments(self, at_type1: str | ThingTypesBase, at_type2: str | ThingTypesBase) -> dict[str, list[Annotation]]: """ Finds views where alignments between two given annotation types occurred. @@ -865,7 +867,7 @@ def get_alignments(self, at_type1: Union[str, ThingTypesBase], at_type2: Union[s at_type1, at_type2 = [ThingTypesBase.from_str(x) if isinstance(x, str) else x for x in (at_type1, at_type2)] assert at_type1 != at_type2, f"Alignment must be between two different types, given only one: {at_type1}" for alignment_view in self.get_all_views_contain(AnnotationTypes.Alignment): - alignments = [] + alignments: list[Annotation] = [] contains_meta = alignment_view.metadata.contains[AnnotationTypes.Alignment] if 'sourceType' in contains_meta and 'targetType' in contains_meta: aligned_types = [ThingTypesBase.from_str(x) @@ -874,12 +876,12 @@ def get_alignments(self, at_type1: Union[str, ThingTypesBase], at_type2: Union[s alignments.extend(alignment_view.annotations) else: for alignment in alignment_view.get_annotations(AnnotationTypes.Alignment): - aligned_types = set() + unique_types = set() for ann_id in [alignment['target'], alignment['source']]: ann_id = cast(str, ann_id) aligned_type = cast(Annotation, self.__getitem__(ann_id)).at_type - aligned_types.add(aligned_type) - aligned_types = list(aligned_types) # because membership check for sets also checks hash() values + unique_types.add(aligned_type) + aligned_types = list(unique_types) # because membership check for sets also checks hash() values if len(aligned_types) == 2 and at_type1 in aligned_types and at_type2 in aligned_types: alignments.append(alignment) if len(alignments) > 0: @@ -912,7 +914,7 @@ def get_all_views_with_error(self) -> list[View]: get_views_with_error = get_all_views_with_error - def get_all_views_contain(self, *at_types: Union[ThingTypesBase, str]) -> list[View]: + def get_all_views_contain(self, *at_types: ThingTypesBase | str) -> list[View]: """ Returns the list of all views in the MMIF if given types are present in that view's 'contains' metadata. @@ -925,7 +927,7 @@ def get_all_views_contain(self, *at_types: Union[ThingTypesBase, str]) -> list[V get_views_contain = get_all_views_contain - def get_view_with_error(self) -> Optional[View]: + def get_view_with_error(self) -> View | None: """ Returns the last view appended that contains an error. @@ -936,7 +938,7 @@ def get_view_with_error(self) -> Optional[View]: return view return None - def get_last_error(self) -> Optional[str]: + def get_last_error(self) -> str | None: """ Returns the last error message found in the views. @@ -945,7 +947,7 @@ def get_last_error(self) -> Optional[str]: v = self.get_view_with_error() return v.get_error() if v is not None else None - def get_view_contains(self, at_types: Union[ThingTypesBase, str, list[Union[str, ThingTypesBase]]]) -> Optional[View]: + def get_view_contains(self, at_types: ThingTypesBase | str | list[str | ThingTypesBase]) -> View | None: """ Returns the last view appended that contains the given types in its 'contains' metadata. @@ -996,7 +998,7 @@ def get_all_views_by_app(self, *app_ids: str) -> list[View]: get_views_by_app = get_all_views_by_app - def get_view_by_app(self, *app_ids: str) -> Optional[View]: + def get_view_by_app(self, *app_ids: str) -> View | None: """ Returns the last view appended that was produced by any of the given apps. App IDs are matched as in :meth:`get_all_views_by_app`. @@ -1021,7 +1023,7 @@ def get_all_views_by_apptag(self, *tags: str) -> list[View]: get_views_by_apptag = get_all_views_by_apptag - def get_view_by_apptag(self, *tags: str) -> Optional[View]: + def get_view_by_apptag(self, *tags: str) -> View | None: """ Returns the last view appended whose ``appTags`` metadata contains any of the given tags. @@ -1033,7 +1035,7 @@ def get_view_by_apptag(self, *tags: str) -> Optional[View]: return matches[-1] if matches else None def get_annotations_of_labels( - self, at_type: Union[ThingTypesBase, str], + self, at_type: ThingTypesBase | str, include_labels: Iterable[str] = (), exclude_labels: Iterable[str] = ()) -> Iterator[Annotation]: """ @@ -1069,7 +1071,7 @@ def get_annotations_of_labels( continue yield ann - def _is_in_time_range(self, ann: Annotation, range_s: Union[int, float], range_e: Union[int, float]) -> bool: + def _is_in_time_range(self, ann: Annotation, range_s: int | float, range_e: int | float) -> bool: """ Checks if the annotation is anchored within the given time range. Any overlap is considered included. @@ -1082,8 +1084,8 @@ def _is_in_time_range(self, ann: Annotation, range_s: Union[int, float], range_e ann_s, ann_e = self.get_start(ann), self.get_end(ann) return (ann_s < range_s < ann_e) or (ann_s < range_e < ann_e) or (ann_s > range_s and ann_e < range_e) - def get_annotations_between_time(self, start: Union[int, float], end: Union[int, float], time_unit: str = "ms", - at_types: list[Union[ThingTypesBase, str]] = []) -> Iterator[Annotation]: + def get_annotations_between_time(self, start: int | float, end: int | float, time_unit: str = "ms", + at_types: list[ThingTypesBase | str] = []) -> Iterator[Annotation]: """ Finds annotations that are anchored between the given time points. @@ -1105,8 +1107,8 @@ def get_annotations_between_time(self, start: Union[int, float], end: Union[int, for view in self.get_all_views_contain(AnnotationTypes.TimeFrame) + self.get_all_views_contain(AnnotationTypes.TimePoint): time_unit_in_view = view.metadata.contains.get(AnnotationTypes.TimeFrame)["timeUnit"] - start_time = convert(start, time_unit, time_unit_in_view, 1) - end_time = convert(end, time_unit, time_unit_in_view, 1) + start_time = cast(int | float, convert(start, time_unit, time_unit_in_view, 1)) + end_time = cast(int | float, convert(end, time_unit, time_unit_in_view, 1)) for ann in view.get_annotations(): if ann.at_type in (AnnotationTypes.TimeFrame, AnnotationTypes.TimePoint) and self._is_in_time_range(ann, start_time, end_time): time_anchors_in_range.append(ann) @@ -1118,7 +1120,7 @@ def get_annotations_between_time(self, start: Union[int, float], end: Union[int, if not uniq_types or aligned.at_type in uniq_types: yield aligned - def _get_linear_anchor_point(self, ann: Annotation, targets_sorted=False, start: bool = True) -> Union[int, float]: + def _get_linear_anchor_point(self, ann: Annotation, targets_sorted=False, start: bool = True) -> int | float: # TODO (krim @ 2/5/24): Update the return type once timeunits are unified to `ms` as integers (https://github.com/clamsproject/mmif/issues/192) """ Retrieves the anchor point of the annotation. Currently, this method only supports linear anchors, @@ -1141,28 +1143,28 @@ def _get_linear_anchor_point(self, ann: Annotation, targets_sorted=False, start: point = math.inf if start else -1 comp = min if start else max for target_id in ann.get_property('targets'): - point = comp(point, self._get_linear_anchor_point(self.__getitem__(target_id), start=start)) + point = comp(point, self._get_linear_anchor_point(cast(Annotation, self.__getitem__(target_id)), start=start)) return point target_id = ann.get_property('targets')[0 if start else -1] - return self._get_linear_anchor_point(self.__getitem__(target_id), start=start) + return self._get_linear_anchor_point(cast(Annotation, self.__getitem__(target_id)), start=start) elif (start and 'start' in props) or (not start and 'end' in props): return ann.get_property('start' if start else 'end') else: raise ValueError(f"{ann.id} ({ann.at_type}) does not have a valid anchor point. Is it a valid 'Region' type?") - def get_start(self, annotation: Annotation) -> Union[int, float]: + def get_start(self, annotation: Annotation) -> int | float: """ An alias to `get_anchor_point` method with `start=True`. """ return self._get_linear_anchor_point(annotation, start=True) - def get_end(self, annotation: Annotation) -> Union[int, float]: + def get_end(self, annotation: Annotation) -> int | float: """ An alias to `get_anchor_point` method with `start=False`. """ return self._get_linear_anchor_point(annotation, start=False) - def __getitem__(self, item: str) -> Union[Document, View, Annotation, MmifMetadata]: + def __getitem__(self, item: str) -> Document | View | Annotation | MmifMetadata: """ High-level string-based access to MMIF objects by their IDs. @@ -1230,7 +1232,7 @@ def __getitem__(self, item: str) -> Union[Document, View, Annotation, MmifMetada else: # search for document first, then views # raise KeyError if nothing is found - ret = self.documents._items.get(item) + ret: Document | View | None = self.documents._items.get(item) if ret is None: ret = self.views._items.get(item) if ret is None: diff --git a/mmif/serialize/model.py b/mmif/serialize/model.py index fed3ed00..33ed3790 100644 --- a/mmif/serialize/model.py +++ b/mmif/serialize/model.py @@ -24,11 +24,11 @@ import json import warnings from collections.abc import Generator, Iterator -from typing import Any, ClassVar, Generic, Optional, TypeVar, Union +from typing import Any, ClassVar, Generic, TypeAlias, TypeVar, Union T = TypeVar('T') S = TypeVar('S') -PRMTV_TYPES: type = Union[str, int, float, bool, None] # noqa: UP007 +PRMTV_TYPES: TypeAlias = Union[str, int, float, bool, None] # noqa: UP007 __all__ = [ 'PRMTV_TYPES', @@ -104,12 +104,13 @@ class MmifObject(object): '_exclude_from_diff', '_contextual_attributes' } - _unnamed_attributes: Optional[dict] + _unnamed_attributes: dict | None + _required_attributes: list[str] _exclude_from_diff: set[str] _contextual_attributes: set[str] _attribute_classes: dict[str, type] = {} # Mapping: str -> Type - def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, mmif_obj: bytes | str | dict | None = None, *_) -> None: if isinstance(mmif_obj, bytes): mmif_obj = mmif_obj.decode('utf8') if not hasattr(self, '_required_attributes'): @@ -159,7 +160,7 @@ def set_additional_property(self, key: str, value: Any) -> None: stacklevel=2) self[attr] = value return - self._unnamed_attributes[key] = value # pytype: disable=unsupported-operands + self._unnamed_attributes[key] = value def _named_attributes(self) -> Generator[str, None, None]: """ @@ -183,7 +184,7 @@ def serialize(self, pretty: bool = False, include_context: bool = True) -> str: """ return json.dumps(self._serialize(include_context=include_context), indent=2 if pretty else None, cls=MmifObjectEncoder) - def _serialize(self, alt_container: Optional[dict] = None, include_context: bool = True) -> dict: + def _serialize(self, alt_container: dict | None = None, include_context: bool = True) -> dict: """ Maps a MMIF object to a plain python dict object, rewriting internal keys that start with '_' to @@ -198,8 +199,9 @@ def _serialize(self, alt_container: Optional[dict] = None, include_context: bool """ container = alt_container if alt_container is not None else self._unnamed_attributes serializing_obj = {} - try: - for k, v in container.items(): # pytype: disable=attribute-error + # `container` is None when additional properties are disallowed + if container is not None: + for k, v in container.items(): if v is None: continue k = str(k) @@ -212,9 +214,6 @@ def _serialize(self, alt_container: Optional[dict] = None, include_context: bool serializing_obj[k] = v._serialize(include_context=include_context) else: serializing_obj[k] = v - except AttributeError as e: - # means _unnamed_attributes is None, so nothing unnamed would be serialized - pass for k, v in self.__dict__.items(): if k in self.reserved_names: continue @@ -245,7 +244,7 @@ def is_empty(obj) -> bool: return False @staticmethod - def _load_json(json_obj: Union[dict, str]) -> dict: + def _load_json(json_obj: dict | str) -> dict: """ Maps JSON-format MMIF strings and dicts into Python dicts with identifier-compliant keys. To do this, it replaces "@" @@ -280,7 +279,7 @@ def deep_from_atsign(d: dict) -> dict: else: raise TypeError(f"tried to load MMIF JSON in a format other than str or dict: {type(json_obj)}") - def deserialize(self, mmif_json: Union[str, dict]) -> None: + def deserialize(self, mmif_json: str | dict) -> None: """ Takes a JSON-formatted string or a simple `dict` that's json-loaded from such a string as an input and populates object's fields with the values @@ -351,7 +350,7 @@ def __setitem__(self, key, value) -> None: else: self.set_additional_property(key, value) - def __contains__(self, key: str) -> bool: + def __contains__(self, key: object) -> bool: try: self.__getitem__(key) return True @@ -450,15 +449,15 @@ class DataList(MmifObject, Generic[T]): :param Union[str, list] mmif_obj: the data that the list contains """ - def __init__(self, mmif_obj: Optional[Union[bytes, str, list]] = None, *_): + def __init__(self, mmif_obj: bytes | str | list | None = None, *_): self.reserved_names.add('_items') self._items: dict[str, T] = dict() self.disallow_additional_properties() if mmif_obj is None: mmif_obj = [] - super().__init__(mmif_obj) + super().__init__(mmif_obj) # type: ignore[arg-type] - def _serialize(self, *args, **kwargs) -> list: # pytype: disable=signature-mismatch + def _serialize(self, *args, **kwargs) -> list: # type: ignore[override] """ Internal serialization method. Returns a list. @@ -466,22 +465,22 @@ def _serialize(self, *args, **kwargs) -> list: # pytype: disable=signature-mism """ return list(super()._serialize(self._items, **kwargs).values()) - def deserialize(self, mmif_json: Union[str, list]) -> None: # pytype: disable=signature-mismatch + def deserialize(self, mmif_json: str | list) -> None: # type: ignore[override] """ Passes the input data into the internal deserializer. """ - super().deserialize(mmif_json) + super().deserialize(mmif_json) # type: ignore[arg-type] @staticmethod - def _load_json(json_list: Union[list, str]) -> list: + def _load_json(json_list: list | str) -> list: # type: ignore[override] if type(json_list) is str: json_list = json.loads(json_list) return [MmifObject._load_json(obj) for obj in json_list] - def _deserialize(self, input_list: list) -> None: + def _deserialize(self, input_list: list) -> None: # type: ignore[override] raise NotImplementedError() - def get(self, key: str, default=None) -> Optional[T]: + def get(self, key: str, default=None) -> T | None: """ .. deprecated:: 1.1.3 Will be removed in 2.0.0. Use container-level access or positional @@ -547,7 +546,7 @@ def _append_with_key(self, key: str, value: T, overwrite=False) -> None: def append(self, value, overwrite): raise NotImplementedError() - def __getitem__(self, key: Union[int, slice]) -> Union[T, list[T]]: + def __getitem__(self, key: int | slice) -> T | list[T]: """ List-style positional access using integers or slices. @@ -611,7 +610,7 @@ def empty(self): class DataDict(MmifObject, Generic[T, S]): - def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_): + def __init__(self, mmif_obj: bytes | str | dict | None = None, *_): self.reserved_names.add('_items') self._items: dict[T, S] = dict() self.disallow_additional_properties() @@ -622,7 +621,7 @@ def __init__(self, mmif_obj: Optional[Union[bytes, str, dict]] = None, *_): def _serialize(self, *args, **kwargs) -> dict: return super()._serialize(self._items, **kwargs) - def get(self, key: T, default=None) -> Optional[S]: + def get(self, key: T, default=None) -> S | None: """ Dictionary-style safe access with optional default value. diff --git a/mmif/serialize/producer.py b/mmif/serialize/producer.py index f414aae1..ac8e7a64 100644 --- a/mmif/serialize/producer.py +++ b/mmif/serialize/producer.py @@ -7,7 +7,7 @@ import warnings from collections import defaultdict from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Optional, Union, cast +from typing import TYPE_CHECKING, Any, cast from mmif.serialize.view import View from mmif.vocabulary import AnnotationTypes @@ -54,7 +54,7 @@ def __init__(self, mmif: 'Mmif', signature: View) -> None: self.signature = signature self._warnings: list[Warning] = [] self._view_metadata: dict[str, Any] = {} - self._run_id: Optional[datetime] = None + self._run_id: datetime | None = None self._existing_view_ids = {view.id for view in mmif.views} self._entry_pending = {doc.id: dict(doc._props_pending.items()) for doc in self._documents()} @@ -63,12 +63,11 @@ def __init__(self, mmif: 'Mmif', signature: View) -> None: def __enter__(self) -> 'ProducerContext': return self - def __exit__(self, exc_type, exc_value, exc_tb) -> bool: + def __exit__(self, exc_type, exc_value, exc_tb) -> None: self.mmif._producer = None - return False def add_warnings(self, - *issued: Union[Warning, warnings.WarningMessage]) -> None: + *issued: Warning | warnings.WarningMessage) -> None: """ Records warnings of this run, written to one signed warnings view. @@ -163,7 +162,7 @@ def _plan_capital_annotations( :return: the plan, in the form that :meth:`mmif.serialize.mmif.Mmif._write_capital_annotations` takes """ - recorded = defaultdict(dict) + recorded: defaultdict[str, dict] = defaultdict(dict) for view in self.mmif.views: for ann in view.get_annotations(AnnotationTypes.Annotation): recorded[ann.get('document')].update( diff --git a/mmif/serialize/view.py b/mmif/serialize/view.py index 3494aa75..30d776b8 100644 --- a/mmif/serialize/view.py +++ b/mmif/serialize/view.py @@ -15,7 +15,7 @@ import warnings from datetime import datetime from collections.abc import Generator -from typing import Optional, Union, cast +from typing import cast from mmif import AnnotationTypes, ClamsTypesBase, DocumentTypes, ThingTypesBase from mmif.serialize.annotation import Annotation, Document, canonicalize_prop_name @@ -63,9 +63,9 @@ class View(MmifObject): some_anns = view.annotations[1:5] # Slice of annotations """ - def __init__(self, view_obj: Optional[Union[bytes, str, dict]] = None, parent_mmif=None, *_) -> None: + def __init__(self, view_obj: bytes | str | dict | None = None, parent_mmif=None, *_) -> None: # used to autogenerate annotation ids - self._id_counts = {} + self._id_counts: dict[str, int] = {} # used to access the parent MMIF object self._parent_mmif = parent_mmif self.reserved_names.update(("_parent_mmif", "_id_counts")) @@ -146,7 +146,7 @@ def _fix_old_short_ids(self): propv[i] = f"{self.id}{self.id_delimiter}{item}" self.annotations.append(annotation) - def new_contain(self, at_type: Union[str, ThingTypesBase], **contains_metadata) -> 'Contain': + def new_contain(self, at_type: str | ThingTypesBase, **contains_metadata) -> 'Contain': """ Adds a new element to the ``contains`` metadata. For a type that is already there, the metadata is merged into its element. @@ -175,7 +175,7 @@ def _set_ann_id(self, annotation: Annotation, identifier): self._id_counts[prefix] = new_num annotation.id = self.id + self.id_delimiter + new_id - def new_annotation(self, at_type: Union[str, ThingTypesBase], aid: Optional[str] = None, + def new_annotation(self, at_type: str | ThingTypesBase, aid: str | None = None, overwrite=False, **properties) -> 'Annotation': """ Generates a new :class:`mmif.serialize.annotation.Annotation` @@ -228,7 +228,7 @@ def add_annotation(self, annotation: 'Annotation', overwrite=False) -> 'Annotati self._parent_mmif._cache_alignment(annotation) return annotation - def new_textdocument(self, text: str, lang: str = "en", did: Optional[str] = None, + def new_textdocument(self, text: str, lang: str = "en", did: str | None = None, overwrite=False, **properties) -> 'Document': """ Generates a new :class:`mmif.serialize.annotation.Document` @@ -272,7 +272,7 @@ def add_document(self, document: Document, overwrite=False) -> Annotation: """ return self.add_annotation(document, overwrite) - def get_annotations(self, at_type: Optional[Union[str, ThingTypesBase]] = None, + def get_annotations(self, at_type: str | ThingTypesBase | None = None, **properties) -> Generator[Annotation, None, None]: """ Look for certain annotations in this view, specified by parameters @@ -398,7 +398,7 @@ def set_error(self, err_message: str, err_trace: str) -> None: self.metadata.set_error(err_message, err_trace) self.annotations.empty() - def get_error(self) -> Optional[str]: + def get_error(self) -> str | None: """ Get the "text" representation of the error occurred during processing. Text representation is supposed to be human-readable. @@ -431,15 +431,15 @@ class ViewMetadata(MmifObject): :param viewmetadata_obj: the JSON data that defines the metadata """ - def __init__(self, viewmetadata_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, viewmetadata_obj: bytes | str | dict | None = None, *_) -> None: self.document: str = '' - self.timestamp: Optional[datetime] = None + self.timestamp: datetime | None = None self.app: str = '' self.appTags: list[str] = [] self.contains: ContainsDict = ContainsDict() self.parameters: dict[str, str] = {} - self.appConfiguration: dict[str, Union[PRMTV_TYPES, list[PRMTV_TYPES]]] = {} - self.error: Union[dict, ErrorDict] = {} + self.appConfiguration: dict[str, PRMTV_TYPES | list[PRMTV_TYPES]] = {} + self.error: dict | ErrorDict = {} self.warnings: list[str] = [] self._required_attributes = ["app"] self._contextual_attributes = {"timestamp"} @@ -479,7 +479,7 @@ def get_error_as_text(self) -> str: else: raise KeyError(f"No error found") - def new_contain(self, at_type: Union[str, ThingTypesBase], **contains_metadata) -> 'Contain': + def new_contain(self, at_type: str | ThingTypesBase, **contains_metadata) -> 'Contain': """ Adds a new element to the ``contains`` dictionary. For a type that is already there, the metadata is merged into its element. @@ -500,16 +500,16 @@ def new_contain(self, at_type: Union[str, ThingTypesBase], **contains_metadata) existing[key] = value return existing - def add_contain(self, contain: 'Contain', at_type: Union[str, ThingTypesBase]) -> None: + def add_contain(self, contain: 'Contain', at_type: str | ThingTypesBase) -> None: self.contains[at_type] = contain - def add_app_configuration(self, config_key: str, config_value: Union[PRMTV_TYPES, list[PRMTV_TYPES]]) -> None: + def add_app_configuration(self, config_key: str, config_value: PRMTV_TYPES | list[PRMTV_TYPES]) -> None: """ Add a configuration key-value pair to the app_configuration dictionary. """ self.appConfiguration[config_key] = config_value - def get_app_configuration(self, config_key: str) -> Union[PRMTV_TYPES, list[PRMTV_TYPES]]: + def get_app_configuration(self, config_key: str) -> PRMTV_TYPES | list[PRMTV_TYPES]: """ Get a configuration value from the app_configuration dictionary. """ @@ -558,7 +558,7 @@ class ErrorDict(MmifObject): """ Error object that stores information about error occurred during processing. """ - def __init__(self, error_obj: Optional[Union[bytes, str, dict]] = None, *_) -> None: + def __init__(self, error_obj: bytes | str | dict | None = None, *_) -> None: self.message: str = '' self.stackTrace: str = '' self._contextual_attributes = {"stackTrace"} @@ -584,14 +584,14 @@ def _for_type(cls, at_type: ThingTypesBase, contains_metadata: dict) -> 'Contain return cls({canonicalize_prop_name(at_type, k): v for k, v in contains_metadata.items()}) -class AnnotationsList(DataList[Union[Annotation, Document]]): +class AnnotationsList(DataList[Annotation | Document]): """ AnnotationsList object that implements :class:`mmif.serialize.model.DataList` for :class:`mmif.serialize.annotation.Annotation`. """ - _items: dict[str, Union[Annotation, Document]] + _items: dict[str, Annotation | Document] - def _deserialize(self, input_list: list) -> None: # pytype: disable=signature-mismatch + def _deserialize(self, input_list: list) -> None: # type: ignore[override] """ Extends base ``_deserialize`` method to initialize ``items`` as a dict from annotation IDs to :class:`mmif.serialize.annotation.Annotation` objects. @@ -603,7 +603,7 @@ def _deserialize(self, input_list: list) -> None: # pytype: disable=signature-m if ClamsTypesBase.attype_iri_isdocument(item['_type']) else Annotation(item) for item in input_list} - def append(self, value: Union[Annotation, Document], overwrite=False) -> None: + def append(self, value: Annotation | Document, overwrite=False) -> None: """ Appends an annotation to the list. @@ -630,18 +630,18 @@ def _deserialize(self, input_dict: dict) -> None: key = ThingTypesBase.from_str(key) self._items[key] = Contain._for_type(key, value) - def update(self, other: 'Union[dict, ContainsDict]', overwrite=False): + def update(self, other: 'dict | ContainsDict', overwrite=False): for k, v in other.items(): if isinstance(k, str): k = ThingTypesBase.from_str(k) self._append_with_key(k, v, overwrite=overwrite) - def get(self, key: Union[str, ThingTypesBase], default=None): + def get(self, key: str | ThingTypesBase, default=None): if isinstance(key, str): key = ThingTypesBase.from_str(key) return self._items.get(key, default) - def __contains__(self, item: Union[str, ThingTypesBase]): + def __contains__(self, item: str | ThingTypesBase): if isinstance(item, str): # in general, when querying with a string, do not use fuzzy equality if 'vocab.lappsgrid.org' in item and item.split('/')[-1] in ThingTypesBase._old_lapps_type_shortnames: diff --git a/mmif/utils/cli/__init__.py b/mmif/utils/cli/__init__.py index 24983fa4..124c2591 100644 --- a/mmif/utils/cli/__init__.py +++ b/mmif/utils/cli/__init__.py @@ -8,7 +8,7 @@ import sys import types from collections.abc import Iterator -from typing import Optional, TextIO, Union, cast, get_args, get_origin +from typing import TextIO, Union, cast, get_args, get_origin from pydantic import BaseModel @@ -18,10 +18,10 @@ @contextlib.contextmanager def open_cli_io_arg( - path_or_dash: Optional[str], + path_or_dash: str | None, mode: str = "r", - encoding: Optional[str] = None, - errors: Optional[str] = None, + encoding: str | None = None, + errors: str | None = None, default_stdin: bool = False, ) -> Iterator[TextIO]: """ @@ -77,7 +77,7 @@ def open_cli_io_arg( should_use_stdio = path_or_dash == "-" or (path_or_dash is None and default_stdin) - file_handle: Optional[TextIO] = None + file_handle: TextIO | None = None should_close = False try: diff --git a/mmif/utils/cli/describe.py b/mmif/utils/cli/describe.py index fbd56618..ba6f5c15 100644 --- a/mmif/utils/cli/describe.py +++ b/mmif/utils/cli/describe.py @@ -3,7 +3,7 @@ import sys import textwrap from pathlib import Path -from typing import Union, cast +from typing import cast from mmif.utils.cli import open_cli_io_arg, generate_model_summary @@ -17,7 +17,7 @@ ) -def get_pipeline_specs(mmif_file: Union[str, Path]): +def get_pipeline_specs(mmif_file: str | Path): """ .. deprecated:: 1.2.1 Will be removed in 2.0.0. Use :func:`mmif.utils.workflow_helper.describe_single_mmif` instead. @@ -29,7 +29,7 @@ def get_pipeline_specs(mmif_file: Union[str, Path]): return describe_single_mmif(mmif_file) -def generate_pipeline_identifier(mmif_file: Union[str, Path]) -> str: +def generate_pipeline_identifier(mmif_file: str | Path) -> str: """ .. deprecated:: 1.2.1 Will be removed in 2.0.0. Use :func:`mmif.utils.workflow_helper.generate_workflow_identifier` instead. @@ -108,6 +108,7 @@ def main(args): """ if hasattr(args, 'help_schema') and args.help_schema is not None: schema_name = args.help_schema[0] + model_cls: type[SingleMmifDesc] | type[CollectionMmifDesc] if schema_name == 'mmif-file': model_cls = SingleMmifDesc elif schema_name == 'mmif-dir': diff --git a/mmif/utils/cli/source.py b/mmif/utils/cli/source.py index 1dc1e30b..b6542329 100644 --- a/mmif/utils/cli/source.py +++ b/mmif/utils/cli/source.py @@ -5,7 +5,7 @@ import sys import textwrap from collections.abc import Generator, Iterable -from typing import Optional, Union +from typing import cast from urllib.parse import urlparse from mmif import Mmif, Document, DocumentTypes, __specver__ @@ -14,10 +14,10 @@ __all__ = ['WorkflowSource'] -DOC_JSON = Union[str, dict] -DOC = Union[DOC_JSON, Document] -METADATA_JSON = Union[str, dict] -METADATA = Union[METADATA_JSON, MmifMetadata] +DOC_JSON = str | dict +DOC = DOC_JSON | Document +METADATA_JSON = str | dict +METADATA = METADATA_JSON | MmifMetadata class WorkflowSource: @@ -40,13 +40,15 @@ class WorkflowSource: def __init__( self, - common_documents_json: Optional[list[DOC_JSON]] = None, - common_metadata_json: Optional[METADATA_JSON] = None + common_documents_json: list[DOC_JSON] | None = None, + common_metadata_json: METADATA_JSON | None = None ) -> None: if common_documents_json is None: common_documents_json = [] if common_metadata_json is None: common_metadata_json = dict() + if isinstance(common_metadata_json, str): + common_metadata_json = json.loads(common_metadata_json) self.mmif_start: dict = {"documents": [json.loads(document) if isinstance(document, str) else document @@ -54,11 +56,11 @@ def __init__( "views": [], "metadata": { "mmif": f"http://mmif.clams.ai/{__specver__}", - **common_metadata_json + **cast(dict, common_metadata_json) }} self.prime() - def add_document(self, document: Union[str, dict, Document]) -> None: + def add_document(self, document: str | dict | Document) -> None: """ Adds a document to the working source MMIF. @@ -104,8 +106,8 @@ def produce(self) -> Mmif: def __call__( self, - documents: Optional[list[DOC]] = None, - metadata: Optional[METADATA] = None + documents: list[DOC] | None = None, + metadata: METADATA | None = None ) -> Mmif: """ Callable API that produces a new MMIF object from @@ -127,20 +129,20 @@ def __call__( if isinstance(documents, str): documents = json.loads(documents) if isinstance(metadata, MmifMetadata): - metadata = metadata.serialize() # pytype: disable=attribute-error # bug in pytype? (https://github.com/google/pytype/issues/533) + metadata = metadata.serialize() if isinstance(metadata, str): metadata = json.loads(metadata) for document in documents: self.add_document(document) - for key, value in metadata.items(): + for key, value in cast(dict, metadata).items(): self.change_metadata(key, value) return self.produce() def from_data( self, doc_lists: Iterable[list[DOC]], - metadata_objs: Optional[Iterable[Optional[METADATA]]] = None + metadata_objs: Iterable[METADATA | None] | None = None ) -> Generator[Mmif, None, None]: """ Provided with an iterable of document lists and an diff --git a/mmif/utils/sequence_helper.py b/mmif/utils/sequence_helper.py index 717f2f04..7dd1e485 100644 --- a/mmif/utils/sequence_helper.py +++ b/mmif/utils/sequence_helper.py @@ -18,11 +18,15 @@ import itertools import warnings from collections.abc import Callable, Iterable +from typing import TYPE_CHECKING import mmif from mmif import Annotation from mmif.serialize.model import PRMTV_TYPES +if TYPE_CHECKING: + import numpy + for seq_dep in ['numpy']: try: importlib.__import__(seq_dep) @@ -265,7 +269,7 @@ def build_label_remapper(src_labels: list[str], dst_labels: dict[str, PRMTV_TYPE def build_score_lists(classifications: list[dict], label_remapper: dict, score_remap_op: Callable[..., float] = max) \ - -> tuple[dict[str, int], "numpy.ndarray"]: # pytype: disable=name-error + -> tuple[dict[str, int], "numpy.ndarray"]: """ Build lists of scores indexed by the label names. @@ -278,7 +282,7 @@ def build_score_lists(classifications: list[dict], label_remapper: dict, 2. 2-d numpy array of scores, of which rows are indexed by label map dict (first return value) """ import numpy - scores = {lbl: [] for lbl in label_remapper.values()} + scores: dict[str, list] = {lbl: [] for lbl in label_remapper.values()} for c_idx, classification in enumerate(classifications): for src_label, src_score in classification.items(): dst_label = label_remapper[src_label] diff --git a/mmif/utils/summarizer/graph.py b/mmif/utils/summarizer/graph.py index b5ea40a2..d75ba7ff 100644 --- a/mmif/utils/summarizer/graph.py +++ b/mmif/utils/summarizer/graph.py @@ -4,7 +4,6 @@ from pathlib import Path import argparse -from typing import Any from mmif import Mmif from mmif.utils.summarizer import config @@ -31,13 +30,11 @@ class Graph(object): """ - def __init__(self, mmif: Any): - # TODO: the type hint should really be "MMif | str", but pytype did not - # like that. - self.mmif = mmif if type(mmif) is Mmif else Mmif(mmif) - self.documents = [] - self.nodes = {} - self.alignments = [] + def __init__(self, mmif: Mmif | str): + self.mmif = mmif if isinstance(mmif, Mmif) else Mmif(mmif) + self.documents: list = [] + self.nodes: dict = {} + self.alignments: list = [] self._init_nodes() self._init_edges() # Third pass to add links between text elements, in particular from @@ -115,7 +112,7 @@ def statistics(self) -> defaultdict: """ Collect counts for node types in each view. """ - stats = defaultdict(int) + stats: defaultdict[str, int] = defaultdict(int) for node in self.nodes.values(): stats[f'{str(node.view_id):4} {node.at_type.shortname}'] += 1 return stats @@ -191,7 +188,7 @@ class TokenIndex(object): # targets to tokens. def __init__(self, tokens): - self.tokens = {} + self.tokens: dict = {} self.token_count = len(tokens) for t in tokens: tup = ((t.properties['start'], t.properties['end']), t) diff --git a/mmif/utils/summarizer/nodes.py b/mmif/utils/summarizer/nodes.py index ac7e004d..e5a3142b 100644 --- a/mmif/utils/summarizer/nodes.py +++ b/mmif/utils/summarizer/nodes.py @@ -343,7 +343,7 @@ def anchor(self) -> dict: is always a time frame or a bounding box. """ # TODO: deal with the case where the primary document is not a video - self.paths = self.paths_to_docs() + self.paths = self.paths_to_docs() # type: ignore[attr-defined] bbtf = self.find_boundingbox_or_timeframe() # for path in paths: # print('... [') @@ -370,7 +370,7 @@ def anchor2(self): # where nodes have paths to anchors # TODO: deal with the case where the primary document is not a video if self._anchor is None: - self._paths = self.paths_to_docs() + self._paths = self.paths_to_docs() # type: ignore[attr-defined] bbtf = self.find_boundingbox_or_timeframe() # for path in self._paths: # print('... [') diff --git a/mmif/utils/summarizer/summary.py b/mmif/utils/summarizer/summary.py index 0db14e66..3d608ff8 100644 --- a/mmif/utils/summarizer/summary.py +++ b/mmif/utils/summarizer/summary.py @@ -13,6 +13,7 @@ import os import pathlib from collections import defaultdict +from typing import Any from mmif.serialize import Mmif from mmif.utils.summarizer import config @@ -109,7 +110,7 @@ def print_warnings(self): def pp(self): self.documents.pp() self.views.pp() - self.transcript.pp() + self.transcript.pp() # type: ignore[attr-defined] self.timeframes.pp() self.entities.pp() print() @@ -201,7 +202,7 @@ def __len__(self): #@staticmethod def get_view_summary(self, view): - annotation_types = defaultdict(int) + annotation_types: defaultdict[str, int] = defaultdict(int) for annotation in view.annotations: annotation_types[annotation.at_type.shortname] += 1 basic_info = { @@ -349,6 +350,9 @@ def __getitem__(self, i): def __len__(self): return len(self.nodes) + def __iter__(self): + return iter(self.nodes) + def add(self, node): self.nodes.append(node) @@ -445,7 +449,8 @@ def _collect_durations(self): def _collect_other_morsels(self): # First we want everything grouped by app and label timeframes = self.timeframes.data - grouped_timeframes = defaultdict(lambda: defaultdict(list)) + grouped_timeframes: defaultdict[str, defaultdict[str, list]] = defaultdict( + lambda: defaultdict(list)) for app in timeframes: for tf in timeframes[app]: label = tf.get('label') @@ -475,7 +480,7 @@ class Entities(Nodes): def __init__(self, summary): super().__init__(summary) self.nodes_idx = {} - self.bins = None + self.bins: Any = None for ent in self.graph.get_nodes(config.NAMED_ENTITY): self.add(ent) self._create_node_index() diff --git a/mmif/utils/summarizer/utils.py b/mmif/utils/summarizer/utils.py index 897a3830..c62115ed 100644 --- a/mmif/utils/summarizer/utils.py +++ b/mmif/utils/summarizer/utils.py @@ -101,7 +101,7 @@ class AnnotationsIndex: def __init__(self, view): self.view = view - self.idx = {} + self.idx: dict = {} self.tokens = [] for annotation in view.annotations: shortname = annotation.at_type.shortname diff --git a/mmif/utils/timeunit_helper.py b/mmif/utils/timeunit_helper.py index c516b7c6..7f75be37 100644 --- a/mmif/utils/timeunit_helper.py +++ b/mmif/utils/timeunit_helper.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Union +from typing import cast UNIT_NORMALIZATION = { @@ -28,7 +28,7 @@ def _isoformat_to_millisecond(isoformat: str) -> int: return int(1000 * (t.hour * 3600 + t.minute * 60 + t.second + t.microsecond / 1000000)) -def _millisecond_to_isoformat(millisecond: int) -> str: +def _millisecond_to_isoformat(millisecond: int | float) -> str: t = datetime.utcfromtimestamp(millisecond / 1000) return t.strftime('%H:%M:%S.%f')[:-3] @@ -38,7 +38,7 @@ def _second_to_isoformat(second: float) -> str: return t.strftime('%H:%M:%S.%f')[:-3] # python strftime will return "microsecond" with 6 digits -def convert(t: Union[int, float, str], in_unit: str, out_unit: str, fps: float) -> Union[int, float, str]: +def convert(t: int | float | str, in_unit: str, out_unit: str, fps: float) -> int | float | str: """ Converts time from one unit to another. Works with ``frames``, ``seconds``, ``milliseconds``. @@ -62,6 +62,7 @@ def convert(t: Union[int, float, str], in_unit: str, out_unit: str, fps: float) in_unit = 'millisecond' else: raise ValueError(f"Invalid time format: ISO format string expected, but got {t} of type {type(t)}") + t = cast(int | float, t) # s>s, ms>ms, f>f if in_unit == out_unit: return t @@ -70,11 +71,8 @@ def convert(t: Union[int, float, str], in_unit: str, out_unit: str, fps: float) if in_unit == 'millisecond': return round(t / 1000 * fps) # s>f - elif 'second' == in_unit: - return round(t * fps) - # i>f else: - return round(_isoformat_to_millisecond(t) / 1000 * fps) + return round(t * fps) # s>(ms or i) elif in_unit == 'second': return round(t * 1000) if out_unit == 'millisecond' else _second_to_isoformat(t) diff --git a/mmif/utils/video_document_helper.py b/mmif/utils/video_document_helper.py index 7f5be83c..9e03fdc8 100644 --- a/mmif/utils/video_document_helper.py +++ b/mmif/utils/video_document_helper.py @@ -5,7 +5,8 @@ import sys import warnings from collections.abc import Iterable, Iterator -from typing import Optional, Union +from collections.abc import Sequence +from typing import cast from contextlib import contextmanager from enum import Enum from io import StringIO @@ -161,7 +162,7 @@ def open_container(video_document: Document): if video_document is None or video_document.at_type != DocumentTypes.VideoDocument: raise ValueError(f'The document does not exist.') - path = video_document.location_path(nonexist_ok=False) + path = cast(str, video_document.location_path(nonexist_ok=False)) container = None with _ffmpeg_op('open', video_document, path): try: @@ -341,7 +342,7 @@ 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() + path = cast(str, video_document.location_path()) result_map = {} decode_errors: list[str] = [] try: @@ -405,7 +406,8 @@ def _emit(frame, t_ms): # left out of `result_map` to become a `None` entry below. while cur_ms is not None: if (prev_frame is not None - and cur_pts - prev_pts <= _tolerance_ticks(prev_frame)): + and cast(int, cur_pts) - cast(int, prev_pts) + <= _tolerance_ticks(prev_frame)): _emit(prev_frame, cur_ms) cur_ms, cur_pts = next(targets, (None, None)) finally: @@ -433,7 +435,8 @@ def _tp_ids_to_timepoints_ms(mmif: Mmif, tp_ids: list[str]) -> list[int]: # falls back to `frame / fps` ms math that ignores the container's PTS # start offset. Fully resolving this requires retiring timeUnit='frame' # (tracked in clams-vocabulary#15). - return [int(round(convert_timepoint(mmif, mmif[tp_id], 'ms'))) + return [int(round(cast(float, convert_timepoint( + mmif, cast(Annotation, mmif[tp_id]), 'ms')))) for tp_id in tp_ids] @@ -451,12 +454,12 @@ def _timeframe_to_timepoint_range_ms( :rtype: tuple """ start, end = convert_timeframe(mmif, time_frame, 'ms') - return int(round(start)), int(round(end)) + return int(round(cast(float, start))), int(round(cast(float, end))) def _sample_all_timepoint_pairs_ms( mmif: Mmif, time_frame: Annotation -) -> list[tuple[int, Optional[str]]]: +) -> list[tuple[int, str | None]]: """ Samples all (timepoint_ms, source_id) pairs from a TimeFrame. Uses all ``targets`` if present (source is the TP annotation id), otherwise @@ -475,7 +478,7 @@ def _sample_all_timepoint_pairs_ms( ms_list = _tp_ids_to_timepoints_ms(mmif, target_ids) return list(zip(ms_list, target_ids)) start_ms, end_ms = _timeframe_to_timepoint_range_ms(mmif, time_frame) - video_document = mmif[time_frame.get_property('document')] + video_document = cast(Document, mmif[time_frame.get_property('document')]) fps = get_framerate(video_document) step_ms = 1000.0 / fps return [(t, None) for t in sample_timepoints(start_ms, end_ms, step_ms)] @@ -505,7 +508,7 @@ def _sample_representatives_timepoint_pairs_ms( def _sample_single_timepoint_pair_ms( mmif: Mmif, time_frame: Annotation -) -> list[tuple[int, Optional[str]]]: +) -> list[tuple[int, str | None]]: """ Samples a single (timepoint_ms, source_id) pair from a TimeFrame. Uses the middle representative if ``representatives`` is present @@ -583,7 +586,7 @@ def extract_images_by_count_with_sources( selected_target_ids = [targets[i] for i in indices] timepoints_ms = _tp_ids_to_timepoints_ms(mmif, selected_target_ids) - video_doc = mmif[annotation.get_property('document')] + video_doc = cast(Document, mmif[annotation.get_property('document')]) images = extract_images_from_timepoints( video_doc, timepoints_ms, as_PIL=as_PIL) return images, selected_target_ids @@ -628,9 +631,9 @@ def extract_images_by_count( def extract_images_by_mode_with_sources( mmif: Mmif, time_frame: Annotation, - mode: Optional[SamplingMode] = None, + mode: SamplingMode | None = None, as_PIL: bool = False, -) -> tuple[list, list[Union[str, int]]]: +) -> tuple[list, list[str | int]]: """ Extracts images from a TimeFrame using a :py:class:`SamplingMode`, alongside the per-image source: a TP annotation id (``str``) when the @@ -656,6 +659,7 @@ def extract_images_by_mode_with_sources( """ if mode is None: mode = _sampling_mode.get() + pairs: Sequence[tuple[int, str | None]] if mode == SamplingMode.ALL: pairs = _sample_all_timepoint_pairs_ms(mmif, time_frame) elif mode == SamplingMode.REPRESENTATIVES: @@ -665,10 +669,10 @@ def extract_images_by_mode_with_sources( if not pairs: return [], [] timepoints_ms = [ms for ms, _ in pairs] - sources: list[Union[str, int]] = [ + sources: list[str | int] = [ tp_id if tp_id is not None else ms for ms, tp_id in pairs ] - video_doc = mmif[time_frame.get_property('document')] + video_doc = cast(Document, mmif[time_frame.get_property('document')]) images = extract_images_from_timepoints( video_doc, timepoints_ms, as_PIL=as_PIL) return images, sources @@ -677,7 +681,7 @@ def extract_images_by_mode_with_sources( def _to_timepoints( output_view: View, video_document: Document, - sources: list[Union[str, int]], + sources: list[str | int], ) -> list[str]: """ Turns per-image sources, as returned by @@ -713,7 +717,7 @@ def iter_timeframe_images( output_view: View, include_labels: Iterable[str] = (), exclude_labels: Iterable[str] = (), - mode: Optional[SamplingMode] = None, + mode: SamplingMode | None = None, as_PIL: bool = False, ) -> Iterator[tuple[Annotation, list, list[str]]]: """ @@ -757,7 +761,7 @@ def iter_timeframe_images( if image is not None] if not readable: continue - video_document = mmif[time_frame.get_property('document')] + video_document = cast(Document, mmif[time_frame.get_property('document')]) timepoint_ids = _to_timepoints( output_view, video_document, [source for _, source in readable]) yield time_frame, [image for image, _ in readable], timepoint_ids @@ -766,7 +770,7 @@ def iter_timeframe_images( def extract_images_by_mode( mmif: Mmif, time_frame: Annotation, - mode: Optional[SamplingMode] = None, + mode: SamplingMode | None = None, as_PIL: bool = False, ) -> list: """ @@ -796,7 +800,7 @@ def extract_images_by_mode( def sample_timepoints( start_ms: int, end_ms: int, - step_ms: Union[int, float], + step_ms: int | float, ) -> list[int]: """ Samples timepoints (in ms) from a half-open time interval @@ -825,7 +829,7 @@ def sample_timepoints( return timepoints -def convert_timepoint(mmif: Mmif, timepoint: Annotation, out_unit: str) -> Union[int, float, str]: +def convert_timepoint(mmif: Mmif, timepoint: Annotation, out_unit: str) -> int | float | str: """ Converts a time point included in an annotation to a different time unit. The input annotation must have ``timePoint`` property. @@ -836,11 +840,11 @@ def convert_timepoint(mmif: Mmif, timepoint: Annotation, out_unit: str) -> Union :return: frame number (integer) or second/millisecond (float) of input timepoint """ in_unit = timepoint.get_property('timeUnit') - vd = mmif[timepoint.get_property('document')] + vd = cast(Document, mmif[timepoint.get_property('document')]) return convert(timepoint.get_property('timePoint'), in_unit, out_unit, get_framerate(vd)) -def convert_timeframe(mmif: Mmif, time_frame: Annotation, out_unit: str) -> tuple[Union[int, float, str], Union[int, float, str]]: +def convert_timeframe(mmif: Mmif, time_frame: Annotation, out_unit: str) -> tuple[int | float | str, int | float | str]: """ Converts start and end points in a ``TimeFrame`` annotation a different time unit. @@ -850,7 +854,7 @@ def convert_timeframe(mmif: Mmif, time_frame: Annotation, out_unit: str) -> tupl :return: tuple of frame numbers, seconds/milliseconds, or ISO notation of TimeFrame's start and end """ in_unit = time_frame.get_property('timeUnit') - vd = mmif[time_frame.get_property('document')] + vd = cast(Document, mmif[time_frame.get_property('document')]) fps = get_framerate(vd) return convert(time_frame.get_property('start'), in_unit, out_unit, fps), convert(time_frame.get_property('end'), in_unit, out_unit, fps) @@ -966,9 +970,9 @@ def _get_mid_framenum(mmif: Mmif, time_frame: Annotation) -> int: :return: middle frame number as an integer """ timeunit = time_frame.get_property('timeUnit') - video_document = mmif[time_frame.get_property('document')] + video_document = cast(Document, mmif[time_frame.get_property('document')]) fps = get_framerate(video_document) - return int(convert(time_frame.get_property('start') + time_frame.get_property('end'), timeunit, 'frame', fps) // 2) + return int(cast(int, convert(time_frame.get_property('start') + time_frame.get_property('end'), timeunit, 'frame', fps)) // 2) def get_mid_framenum(mmif: Mmif, time_frame: Annotation) -> int: @@ -997,7 +1001,7 @@ def extract_mid_frame(mmif: Mmif, time_frame: Annotation, as_PIL: bool = False): warnings.warn('extract_mid_frame() is deprecated and will be removed in ' '2.0.0. Use extract_images_by_mode() instead.', DeprecationWarning, stacklevel=2) - vd = mmif[time_frame.get_property('document')] + vd = cast(Document, mmif[time_frame.get_property('document')]) fn = get_mid_framenum(mmif, time_frame) return extract_frames_as_images(vd, [fn], as_PIL=as_PIL)[0] @@ -1021,13 +1025,13 @@ def get_representative_framenums(mmif: Mmif, time_frame: Annotation) -> list[int if 'representatives' not in time_frame: return [_get_mid_framenum(mmif, time_frame)] timeunit = time_frame.get_property('timeUnit') - video_document = mmif[time_frame.get_property('document')] + video_document = cast(Document, mmif[time_frame.get_property('document')]) fps = get_framerate(video_document) representatives = time_frame.get_property('representatives') ref_frams = [] for rep_id in representatives: try: - rep_anno = mmif[rep_id] + rep_anno = cast(Annotation, mmif[rep_id]) except KeyError as ke: raise ValueError(f'Representative timepoint {rep_id} not found in any view. ({ke})') ref_frams.append(int(convert(rep_anno.get_property('timePoint'), timeunit, 'frame', fps))) @@ -1067,7 +1071,7 @@ def extract_representative_frame(mmif: Mmif, time_frame: Annotation, as_PIL: boo warnings.warn('extract_representative_frame() is deprecated and will be ' 'removed in 2.0.0. Use extract_images_by_mode() instead.', DeprecationWarning, stacklevel=2) - video_document = mmif[time_frame.get_property('document')] + video_document = cast(Document, mmif[time_frame.get_property('document')]) rep_frame_num = [get_representative_framenum(mmif, time_frame)] if first_only else get_representative_framenums(mmif, time_frame) return extract_frames_as_images(video_document, rep_frame_num, as_PIL=as_PIL)[0] @@ -1093,7 +1097,7 @@ def sample_frames(start_frame: int, end_frame: int, sample_rate: float = 1) -> l if sample_rate < 1: raise ValueError(f"Sample rate must be greater than 1, but got {sample_rate}") frame_nums: list[int] = [] - cur_f = start_frame + cur_f: float = start_frame while cur_f < end_frame: ceiling = math.ceil(cur_f) if ceiling < end_frame: diff --git a/mmif/utils/workflow_helper.py b/mmif/utils/workflow_helper.py index 824cc844..b6dee070 100644 --- a/mmif/utils/workflow_helper.py +++ b/mmif/utils/workflow_helper.py @@ -3,7 +3,7 @@ import itertools from collections import Counter from pathlib import Path -from typing import Any, Literal, Optional, Union, overload +from typing import Any, Literal, overload from pydantic import BaseModel, ConfigDict, Field @@ -50,7 +50,7 @@ def group_views_by_app(views: ViewsList) -> list[list[Any]]: def _split_appname_appversion( long_app_id: str -) -> tuple[Optional[str], Optional[str]]: +) -> tuple[str | None, str | None]: """ Split app name and version from a long app identifier. @@ -90,7 +90,7 @@ def generate_param_hash(params: dict) -> str: return hashlib.md5(param_string.encode('utf-8')).hexdigest() -def _read_mmif_from_path(mmif_input: Union[str, Path, Mmif]) -> Mmif: +def _read_mmif_from_path(mmif_input: str | Path | Mmif) -> Mmif: """ Helper function to get a Mmif object from various input types. @@ -111,20 +111,20 @@ def _read_mmif_from_path(mmif_input: Union[str, Path, Mmif]) -> Mmif: @overload -def generate_workflow_identifier(mmif_input: Union[str, Path, Mmif], +def generate_workflow_identifier(mmif_input: str | Path | Mmif, return_param_dicts: Literal[True] ) -> tuple[str, list[dict]]: ... @overload -def generate_workflow_identifier(mmif_input: Union[str, Path, Mmif], +def generate_workflow_identifier(mmif_input: str | Path | Mmif, return_param_dicts: Literal[False] = False ) -> str: ... -def generate_workflow_identifier(mmif_input: Union[str, Path, Mmif], +def generate_workflow_identifier(mmif_input: str | Path | Mmif, return_param_dicts: bool = False - ) -> Union[str, tuple[str, list[dict]]]: + ) -> str | tuple[str, list[dict]]: """ Generate a workflow identifier string from a MMIF file or object. @@ -208,7 +208,7 @@ class AppProfiling(BaseModel): """ model_config = ConfigDict(populate_by_name=True) - running_time_ms: Optional[int] = Field(default=None, alias="runningTimeMS", description="Execution time in milliseconds.") + running_time_ms: int | None = Field(default=None, alias="runningTimeMS", description="Execution time in milliseconds.") class AppExecution(BaseModel): """ @@ -262,7 +262,7 @@ def _get_profile_data(view) -> AppProfiling: return AppProfiling(runningTimeMS=milliseconds) -def describe_single_mmif(mmif_input: Union[str, Path, Mmif]) -> dict: +def describe_single_mmif(mmif_input: str | Path | Mmif) -> dict: """ Reads a MMIF file or object and extracts the workflow specification from it. @@ -319,7 +319,7 @@ def describe_single_mmif(mmif_input: Union[str, Path, Mmif]) -> dict: for group in view_groups: first_view = group[0] - execution_ann_counter = Counter() + execution_ann_counter: Counter[str] = Counter() for view in group: execution_ann_counter.update(Counter(str(ann.at_type) for ann in view.annotations)) @@ -358,7 +358,7 @@ def describe_single_mmif(mmif_input: Union[str, Path, Mmif]) -> dict: )) # aggregate total annotation counts - total_annotations_by_type = Counter() + total_annotations_by_type: Counter[str] = Counter() for execution in grouped_apps: # Only aggregate from actual apps, not the special unassigned entry if execution.app != "http://apps.clams.ai/non-existing-app/v1": @@ -394,10 +394,10 @@ class AppProfilingStats(BaseModel): """ model_config = ConfigDict(populate_by_name=True) - avg_running_time_ms: Optional[float] = Field(default=None, alias="avgRunningTimeMS", description="Average execution time in milliseconds.") - min_running_time_ms: Optional[float] = Field(default=None, alias="minRunningTimeMS", description="Minimum execution time in milliseconds.") - max_running_time_ms: Optional[float] = Field(default=None, alias="maxRunningTimeMS", description="Maximum execution time in milliseconds.") - stdev_running_time_ms: Optional[float] = Field(default=None, alias="stdevRunningTimeMS", description="Standard deviation of execution time.") + avg_running_time_ms: float | None = Field(default=None, alias="avgRunningTimeMS", description="Average execution time in milliseconds.") + min_running_time_ms: float | None = Field(default=None, alias="minRunningTimeMS", description="Minimum execution time in milliseconds.") + max_running_time_ms: float | None = Field(default=None, alias="maxRunningTimeMS", description="Maximum execution time in milliseconds.") + stdev_running_time_ms: float | None = Field(default=None, alias="stdevRunningTimeMS", description="Standard deviation of execution time.") @@ -448,7 +448,7 @@ class CollectionMmifDesc(BaseModel): annotation_count_by_type: dict[str, int] = Field(default_factory=dict, alias="annotationCountByType", description="Total annotation counts across the entire collection.") -def describe_mmif_collection(mmif_dir: Union[str, Path]) -> dict: +def describe_mmif_collection(mmif_dir: str | Path) -> dict: """ Reads all MMIF files in a directory and extracts a summarized workflow specification. @@ -476,7 +476,7 @@ def describe_mmif_collection(mmif_dir: Union[str, Path]) -> dict: invalid=0 ) - aggregated_counts = Counter() + aggregated_counts: Counter[str] = Counter() # Structure: {workflow_id: {'mmifs': [...], 'apps': {app_uri: {'appConfiguration': ..., 'execution_times': [...]}}}} workflows_data: dict[str, dict] = {} diff --git a/mmif/ver/__init__.py b/mmif/ver/__init__.py index bddd3a5c..c4ab731e 100644 --- a/mmif/ver/__init__.py +++ b/mmif/ver/__init__.py @@ -9,7 +9,7 @@ _urls = { k: v for k, v in ( u.split(", ", 1) - for u in metadata("mmif-python").get_all("Project-URL") + for u in metadata("mmif-python").get_all("Project-URL", []) ) } __specver__ = _urls["mmif-spec"].rstrip("/").rsplit("/", 1)[-1] diff --git a/mmif_docloc_http/__init__.py b/mmif_docloc_http/__init__.py index 1d954474..7697c594 100644 --- a/mmif_docloc_http/__init__.py +++ b/mmif_docloc_http/__init__.py @@ -9,7 +9,7 @@ import urllib.request import urllib.error -_cache = {} +_cache: dict = {} def resolve(docloc): diff --git a/pyproject.toml b/pyproject.toml index e23fb029..5691624b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,7 +45,7 @@ docs = ["sphinx>=7.0,<8.0", "furo", "m2r2", "autodoc-pydantic"] test = [ "mmif-python[cv]", "mmif-python[seq]", - "pytype", "pytest", "pytest-cov", + "mypy", "pytest", "pytest-cov", "hypothesis", "hypothesis-jsonschema", "pyyaml", "bs4", "lxml", ] @@ -58,6 +58,15 @@ include = ["mmif*"] [tool.setuptools.package-data] mmif = ["res/*"] +[tool.mypy] +files = ["mmif"] +check_untyped_defs = true + +[[tool.mypy.overrides]] +# dependencies that ship neither inline types nor a py.typed marker +module = ["clams_vocabulary.*", "jsonschema.*"] +ignore_missing_imports = true + [tool.pytest.ini_options] testpaths = ["mmif", "tests"] addopts = "--cov=mmif --cov-report=xml" diff --git a/tests/test_serialize.py b/tests/test_serialize.py index 4fc78d74..77bb1bf0 100644 --- a/tests/test_serialize.py +++ b/tests/test_serialize.py @@ -26,6 +26,16 @@ tester_appname = 'http://not.existing/app' +class TestPackageExports(unittest.TestCase): + + def test_literal_all_matches_submodules(self): + import mmif.serialize as serialize + expected = {name for module in (serialize.annotation, serialize.mmif, + serialize.producer, serialize.view) + for name in module.__all__} + self.assertEqual(set(serialize.__all__), expected) + + class TestMmif(unittest.TestCase): def setUp(self) -> None: