From 1debd29d379b943e793afa2dbec3b5ca5f48c440 Mon Sep 17 00:00:00 2001 From: Glen Beane <356266+gbeane@users.noreply.github.com> Date: Mon, 10 Aug 2026 05:14:56 +0000 Subject: [PATCH] build timeline annotations through the Annotation dataclass on load TimelineAnnotations.load() hand-built the same interval payload that add_annotation() builds, so the two could drift. It also rebound its own `data` parameter to that per-entry dict inside the loop that iterates `data`, which is harmless today only because the iterator is captured before the first rebind. Route load() through cls.Annotation/add_annotation instead, rename the loop variable to `entry`, and correct the docstring, which described `data` as a dict rather than a list of serialized annotations. Also replace the module's print() calls with the standard logging module per the project's logging conventions. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Ua3Fkt11kvBCZ6iuVbsbVc --- src/jabs/project/timeline_annotations.py | 73 ++++++++++++---------- tests/project/test_timeline_annotations.py | 64 +++++++++++++++++++ 2 files changed, 103 insertions(+), 34 deletions(-) diff --git a/src/jabs/project/timeline_annotations.py b/src/jabs/project/timeline_annotations.py index b8bb27fe..dfab0e11 100644 --- a/src/jabs/project/timeline_annotations.py +++ b/src/jabs/project/timeline_annotations.py @@ -1,10 +1,12 @@ -import sys +import logging from collections.abc import Callable from dataclasses import dataclass from typing import Any from intervaltree import Interval, IntervalTree +logger = logging.getLogger(__name__) + MAX_TAG_LEN = 32 @@ -40,7 +42,8 @@ def add_annotation(self, annotation: Annotation) -> None: """Add a new timeline annotation to the interval tree. Args: - annotation (Annotation): The annotation to add. + annotation (Annotation): The annotation to add. Its ``start`` and ``end`` + frames are inclusive, so the stored interval ends at ``end + 1``. """ self._tree[annotation.start : annotation.end + 1] = { "tag": annotation.tag, @@ -54,52 +57,53 @@ def add_annotation(self, annotation: Annotation) -> None: def load( cls, data: list[dict[str, Any]], id_index_to_display: Callable[[int], str] | None = None ) -> "TimelineAnnotations": - """Load a TimelineAnnotations instance from a JSON-compatible dictionary. + """Load a TimelineAnnotations instance from a list of serialized annotations. Args: - data (dict): The dictionary containing serialized timeline annotations. + data (list[dict[str, Any]]): Serialized timeline annotations, as produced by + :meth:`serialize`. The ``start`` and ``end`` frames are inclusive. id_index_to_display (Callable[[int], str] | None): Optional function to map identity index to a display string. Returns: TimelineAnnotations: An instance loaded with the provided data. - Note: loading currently skips invalid entries with a warning printed to stderr. Consider + Note: loading currently skips invalid entries with a logged warning. Consider raising an exception for stricter handling in the future. """ annotations = cls() - for annotation in data: + for entry in data: try: - start = annotation["start"] - end = annotation["end"] - tag = annotation["tag"] - color = annotation["color"] + start = entry["start"] + end = entry["end"] + tag = entry["tag"] + color = entry["color"] except KeyError: - print( - "Missing required annotation fields, loading skipped for annotation:", - annotation, - file=sys.stderr, + logger.warning( + "Missing required annotation fields, loading skipped for annotation: %s", entry ) continue # validate the tag format: if len(tag) < 1 or len(tag) > MAX_TAG_LEN: - print( - f"Annotation tag must be 1 to {MAX_TAG_LEN} characters in length, skipping annotation: \n\t{annotation}", - file=sys.stderr, + logger.warning( + "Annotation tag must be 1 to %d characters in length, " + "skipping annotation: \n\t%s", + MAX_TAG_LEN, + entry, ) continue # only allow alphanumeric characters, underscores, and hyphens if not all(c.isalnum() or c in "_-" for c in tag): - print( - f"Annotation tag can only contain alphanumeric characters, underscores, and hyphens. Skipping annotation: \n\t{annotation}", - file=sys.stderr, + logger.warning( + "Annotation tag can only contain alphanumeric characters, underscores, " + "and hyphens. Skipping annotation: \n\t%s", + entry, ) continue - # Create a data dict for the interval. # Note: description and identity are optional fields - identity_index = annotation.get("identity") + identity_index = entry.get("identity") if identity_index is not None: if id_index_to_display: display_identity = id_index_to_display(identity_index) @@ -107,17 +111,18 @@ def load( display_identity = str(identity_index) else: display_identity = None - data = { - "tag": tag, - "color": color, - "description": annotation.get("description"), - "identity": identity_index, - "display_identity": display_identity, - } - - # Add the annotation to the IntervalTree. - # The start and end contained in the JSON file are inclusive, so we add 1 to end. - annotations._tree[start : end + 1] = data + + annotations.add_annotation( + cls.Annotation( + start=start, + end=end, + tag=tag, + color=color, + description=entry.get("description"), + identity_index=identity_index, + display_identity=display_identity, + ) + ) return annotations def serialize(self) -> list[dict]: @@ -137,7 +142,7 @@ def serialize(self) -> list[dict]: "color": element.data["color"], } except KeyError as e: - print(f"Missing required annotation data: {e}") + logger.warning("Missing required annotation data: %s", e) continue # optional fields diff --git a/tests/project/test_timeline_annotations.py b/tests/project/test_timeline_annotations.py index 10844c5f..246f8996 100644 --- a/tests/project/test_timeline_annotations.py +++ b/tests/project/test_timeline_annotations.py @@ -137,6 +137,70 @@ def test_load_skips_invalid_entries(): assert rebuilt2.annotation_exists(start=0, end=1, tag="good_tag-1", identity_index=None) +def test_load_stores_same_payload_as_add_annotation(): + """load() and add_annotation() must produce identical interval payloads.""" + loaded = TimelineAnnotations.load( + [ + { + "start": 5, + "end": 9, + "tag": "tag1", + "color": "#abcdef", + "description": "desc", + "identity": 3, + } + ], + id_index_to_display=identity_index_to_display, + ) + added = TimelineAnnotations() + added.add_annotation( + TimelineAnnotations.Annotation( + start=5, + end=9, + tag="tag1", + color="#abcdef", + description="desc", + identity_index=3, + display_identity="ID-3", + ) + ) + + assert sorted(loaded._tree) == sorted(added._tree) + + +def test_load_continues_after_skipping_invalid_entry(): + """A skipped entry must not stop the remaining entries from loading.""" + data = [ + {"start": 0, "end": 1, "color": "#fff"}, # missing tag, skipped + {"start": 2, "end": 3, "tag": "ok_one", "color": "#fff"}, + {"start": 4, "end": 5, "tag": "bad tag!", "color": "#fff"}, # invalid tag, skipped + {"start": 6, "end": 7, "tag": "ok_two", "color": "#fff"}, + ] + rebuilt = TimelineAnnotations.load(data) + + assert len(rebuilt) == 2 + assert rebuilt.annotation_exists(start=2, end=3, tag="ok_one", identity_index=None) + assert rebuilt.annotation_exists(start=6, end=7, tag="ok_two", identity_index=None) + + +def test_load_without_display_mapping_stringifies_identity(): + """Without a mapping function the identity index is stored as a string.""" + rebuilt = TimelineAnnotations.load( + [{"start": 0, "end": 1, "tag": "tag1", "color": "#fff", "identity": 7}] + ) + (interval,) = list(rebuilt._tree) + assert interval.data["identity"] == 7 + assert interval.data["display_identity"] == "7" + + +def test_load_warns_on_skipped_entry(caplog): + """Skipped entries are reported through the module logger, not stdout/stderr.""" + with caplog.at_level("WARNING", logger="jabs.project.timeline_annotations"): + TimelineAnnotations.load([{"start": 0, "end": 1, "tag": "bad tag!", "color": "#fff"}]) + + assert any("Skipping annotation" in record.getMessage() for record in caplog.records) + + @pytest.mark.parametrize( "tag, ok", [