From adeaf09cd10ec1b6f105da6886c00370fd894103 Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Wed, 8 Jul 2026 10:16:58 +0300 Subject: [PATCH 1/9] Add ActivityNet format --- src/labelformat/formats/__init__.py | 8 + src/labelformat/formats/activitynet.py | 260 ++++++++++++++++++ .../model/temporal_classification.py | 60 ++++ tests/unit/formats/test_activitynet.py | 163 +++++++++++ 4 files changed, 491 insertions(+) create mode 100644 src/labelformat/formats/activitynet.py create mode 100644 src/labelformat/model/temporal_classification.py create mode 100644 tests/unit/formats/test_activitynet.py diff --git a/src/labelformat/formats/__init__.py b/src/labelformat/formats/__init__.py index 844a330..905b776 100644 --- a/src/labelformat/formats/__init__.py +++ b/src/labelformat/formats/__init__.py @@ -1,3 +1,8 @@ +from labelformat.formats.activitynet import ( + ActivityNetTemporalClassificationDatabaseOutput, + ActivityNetTemporalClassificationInput, + ActivityNetTemporalClassificationResultsOutput, +) from labelformat.formats.coco import ( COCOInstanceSegmentationInput, COCOInstanceSegmentationOutput, @@ -76,6 +81,9 @@ ) __all__ = [ + "ActivityNetTemporalClassificationDatabaseOutput", + "ActivityNetTemporalClassificationInput", + "ActivityNetTemporalClassificationResultsOutput", "COCOInstanceSegmentationInput", "COCOInstanceSegmentationOutput", "COCOObjectDetectionInput", diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py new file mode 100644 index 0000000..c4bbaa9 --- /dev/null +++ b/src/labelformat/formats/activitynet.py @@ -0,0 +1,260 @@ +from __future__ import annotations + +import json +from argparse import ArgumentParser +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +from labelformat.model.category import Category +from labelformat.model.temporal_classification import ( + TemporalClassificationInput, + TemporalClassificationOutput, + TemporalEvent, + VideoTemporalClassification, +) +from labelformat.types import JsonDict, ParseError + + +class _ActivityNetBaseInput: + @staticmethod + def add_cli_arguments(parser: ArgumentParser) -> None: + parser.add_argument( + "--input-file", + type=Path, + required=True, + help="Path to input ActivityNet JSON file", + ) + + def __init__(self, input_file: Path) -> None: + with input_file.open(encoding="utf-8") as file: + data = json.load(file) + self._labels, self._categories = _parse_activitynet_data(data=data) + + def get_categories(self) -> Iterable[Category]: + yield from self._categories + + def get_labels(self) -> Iterable[VideoTemporalClassification]: + yield from self._labels + + +class ActivityNetTemporalClassificationInput( + _ActivityNetBaseInput, TemporalClassificationInput +): + """Import ActivityNet-style temporal classification annotations.""" + + +class _ActivityNetBaseOutput: + @staticmethod + def add_cli_arguments(parser: ArgumentParser) -> None: + parser.add_argument( + "--output-file", + type=Path, + required=True, + help="Path to output ActivityNet JSON file", + ) + + def __init__(self, output_file: Path) -> None: + self.output_file = output_file + + +class ActivityNetTemporalClassificationResultsOutput( + _ActivityNetBaseOutput, TemporalClassificationOutput +): + """Export ActivityNet submission-style JSON with a top-level ``results`` key.""" + + def save(self, label_input: TemporalClassificationInput) -> None: + data: JsonDict = { + "results": _get_output_results_dict(label_input.get_labels()), + } + self.output_file.parent.mkdir(parents=True, exist_ok=True) + with self.output_file.open("w", encoding="utf-8") as file: + json.dump(data, file, indent=2) + + +class ActivityNetTemporalClassificationDatabaseOutput( + _ActivityNetBaseOutput, TemporalClassificationOutput +): + """Export ActivityNet ground-truth-style JSON with a top-level ``database`` key.""" + + def save(self, label_input: TemporalClassificationInput) -> None: + data: JsonDict = { + "database": _get_output_database_dict(label_input.get_labels()), + } + self.output_file.parent.mkdir(parents=True, exist_ok=True) + with self.output_file.open("w", encoding="utf-8") as file: + json.dump(data, file, indent=2) + + +def _parse_activitynet_data( + data: JsonDict, +) -> tuple[list[VideoTemporalClassification], list[Category]]: + if "database" in data: + return _parse_database(database=data["database"]) + if "results" in data: + return _parse_results(results=data["results"]) + + raise ParseError("ActivityNet JSON must contain a 'database' or 'results' key.") + + +def _parse_database( + database: JsonDict, +) -> tuple[list[VideoTemporalClassification], list[Category]]: + label_names: dict[str, None] = {} + parsed_by_video: list[tuple[str, list[_ParsedEvent], float | None]] = [] + + for video_id, video_entry in database.items(): + if not isinstance(video_entry, dict): + raise ParseError(f"Invalid database entry for video '{video_id}'.") + raw_annotations = video_entry.get("annotations", []) + if not isinstance(raw_annotations, list): + raise ParseError(f"Invalid annotations for video '{video_id}'.") + duration = video_entry.get("duration") + duration_s = float(duration) if duration is not None else None + events = [_parse_event(annotation=annotation) for annotation in raw_annotations] + label_names.update((event.label, None) for event in events) + parsed_by_video.append((str(video_id), events, duration_s)) + + categories = _categories_from_label_names(label_names=label_names) + category_name_to_id = {category.name: category.id for category in categories} + labels = [ + VideoTemporalClassification( + video_id=video_id, + events=_events_from_parsed( + events=events, category_name_to_id=category_name_to_id + ), + duration_s=duration_s, + ) + for video_id, events, duration_s in parsed_by_video + ] + return labels, categories + + +def _parse_results( + results: JsonDict, +) -> tuple[list[VideoTemporalClassification], list[Category]]: + label_names: dict[str, None] = {} + parsed_by_video: list[tuple[str, list[_ParsedEvent]]] = [] + + for video_id, raw_annotations in results.items(): + if not isinstance(raw_annotations, list): + raise ParseError(f"Invalid results entry for video '{video_id}'.") + events = [_parse_event(annotation=annotation) for annotation in raw_annotations] + label_names.update((event.label, None) for event in events) + parsed_by_video.append((str(video_id), events)) + + categories = _categories_from_label_names(label_names=label_names) + category_name_to_id = {category.name: category.id for category in categories} + labels = [ + VideoTemporalClassification( + video_id=video_id, + events=_events_from_parsed( + events=events, category_name_to_id=category_name_to_id + ), + duration_s=None, + ) + for video_id, events in parsed_by_video + ] + return labels, categories + + +@dataclass(frozen=True) +class _ParsedEvent: + label: str + start_time_s: float + end_time_s: float + confidence: float | None + + +def _parse_event(annotation: JsonDict) -> _ParsedEvent: + label = annotation.get("label") + segment = annotation.get("segment") + if not isinstance(label, str) or not label: + raise ParseError("ActivityNet event must contain a non-empty 'label'.") + if not isinstance(segment, list) or len(segment) != 2: + raise ParseError( + "ActivityNet event must contain 'segment' as [start_s, end_s]." + ) + + start_time_s = float(segment[0]) + end_time_s = float(segment[1]) + if start_time_s < 0 or start_time_s >= end_time_s: + raise ParseError( + f"Invalid segment [{start_time_s}, {end_time_s}] for label '{label}': " + "start must be non-negative and less than end." + ) + + confidence = annotation.get("score") + if confidence is not None: + confidence = float(confidence) + + return _ParsedEvent( + label=label, + start_time_s=start_time_s, + end_time_s=end_time_s, + confidence=confidence, + ) + + +def _categories_from_label_names(label_names: Iterable[str]) -> list[Category]: + return [ + Category(id=index, name=label_name) + for index, label_name in enumerate(label_names, start=1) + ] + + +def _events_from_parsed( + events: list[_ParsedEvent], + category_name_to_id: dict[str, int], +) -> list[TemporalEvent]: + return [ + TemporalEvent( + category=Category( + id=category_name_to_id[event.label], + name=event.label, + ), + start_time_s=event.start_time_s, + end_time_s=event.end_time_s, + confidence=event.confidence, + ) + for event in events + ] + + +def _get_output_results_dict( + labels: Iterable[VideoTemporalClassification], +) -> JsonDict: + results: JsonDict = {} + for label in labels: + annotations: list[JsonDict] = [] + for event in label.events: + annotation: JsonDict = { + "label": event.category.name, + "segment": [event.start_time_s, event.end_time_s], + } + if event.confidence is not None: + annotation["score"] = event.confidence + annotations.append(annotation) + results[label.video_id] = annotations + return results + + +def _get_output_database_dict( + labels: Iterable[VideoTemporalClassification], +) -> JsonDict: + database: JsonDict = {} + for label in labels: + annotations: list[JsonDict] = [] + for event in label.events: + annotation: JsonDict = { + "label": event.category.name, + "segment": [event.start_time_s, event.end_time_s], + } + if event.confidence is not None: + annotation["score"] = event.confidence + annotations.append(annotation) + video_entry: JsonDict = {"annotations": annotations} + if label.duration_s is not None: + video_entry["duration"] = label.duration_s + database[label.video_id] = video_entry + return database diff --git a/src/labelformat/model/temporal_classification.py b/src/labelformat/model/temporal_classification.py new file mode 100644 index 0000000..2868a45 --- /dev/null +++ b/src/labelformat/model/temporal_classification.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from argparse import ArgumentParser +from dataclasses import dataclass +from typing import Iterable + +from labelformat.model.category import Category + + +@dataclass(frozen=True) +class TemporalEvent: + """A single temporal classification event on a video.""" + + category: Category + start_time_s: float + end_time_s: float + confidence: float | None = None + + def __post_init__(self) -> None: + if self.start_time_s < 0 or self.start_time_s >= self.end_time_s: + raise ValueError( + f"Invalid segment [{self.start_time_s}, {self.end_time_s}] " + f"for label '{self.category.name}': start must be non-negative and less than end." + ) + + +@dataclass(frozen=True) +class VideoTemporalClassification: + """All temporal classification events for one video.""" + + video_id: str + events: list[TemporalEvent] + duration_s: float | None = None + + +class TemporalClassificationInput(ABC): + @staticmethod + @abstractmethod + def add_cli_arguments(parser: ArgumentParser) -> None: + raise NotImplementedError() + + @abstractmethod + def get_categories(self) -> Iterable[Category]: + raise NotImplementedError() + + @abstractmethod + def get_labels(self) -> Iterable[VideoTemporalClassification]: + raise NotImplementedError() + + +class TemporalClassificationOutput(ABC): + @staticmethod + @abstractmethod + def add_cli_arguments(parser: ArgumentParser) -> None: + raise NotImplementedError() + + @abstractmethod + def save(self, label_input: TemporalClassificationInput) -> None: + raise NotImplementedError() diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py new file mode 100644 index 0000000..950c0d3 --- /dev/null +++ b/tests/unit/formats/test_activitynet.py @@ -0,0 +1,163 @@ +import json +from pathlib import Path + +import pytest + +from labelformat.formats.activitynet import ( + ActivityNetTemporalClassificationDatabaseOutput, + ActivityNetTemporalClassificationInput, + ActivityNetTemporalClassificationResultsOutput, +) +from labelformat.model.category import Category +from labelformat.model.temporal_classification import ( + TemporalEvent, + VideoTemporalClassification, +) +from labelformat.types import ParseError + + +class TestActivityNetTemporalClassificationDatabaseInput: + def test_get_categories(self, tmp_path: Path) -> None: + input_file = _write_activitynet_database_json(tmp_path / "activity_net.json") + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + + assert list(label_input.get_categories()) == [ + Category(id=1, name="Person walking"), + Category(id=2, name="Rock climbing"), + ] + + def test_get_labels(self, tmp_path: Path) -> None: + input_file = _write_activitynet_database_json(tmp_path / "activity_net.json") + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + + assert list(label_input.get_labels()) == [ + VideoTemporalClassification( + video_id="v_test_video", + duration_s=82.75, + events=[ + TemporalEvent( + category=Category(id=1, name="Person walking"), + start_time_s=0.58, + end_time_s=6.16, + ), + TemporalEvent( + category=Category(id=2, name="Rock climbing"), + start_time_s=10.0, + end_time_s=20.0, + confidence=0.9, + ), + ], + ) + ] + + +class TestActivityNetTemporalClassificationResultsInput: + def test_get_labels(self, tmp_path: Path) -> None: + input_file = _write_activitynet_results_json(tmp_path / "results.json") + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + + assert list(label_input.get_labels()) == [ + VideoTemporalClassification( + video_id="v_test_video", + duration_s=None, + events=[ + TemporalEvent( + category=Category(id=1, name="Person walking"), + start_time_s=0.58, + end_time_s=6.16, + confidence=0.75, + ) + ], + ) + ] + + def test_rejects_invalid_top_level_key(self, tmp_path: Path) -> None: + input_file = tmp_path / "invalid.json" + input_file.write_text(json.dumps({"videos": {}})) + + with pytest.raises(ParseError, match="database' or 'results'"): + ActivityNetTemporalClassificationInput(input_file=input_file) + + def test_rejects_invalid_segment(self, tmp_path: Path) -> None: + input_file = tmp_path / "invalid.json" + input_file.write_text( + json.dumps( + { + "results": { + "v_test_video": [ + {"label": "Person walking", "segment": [1.0]}, + ] + } + } + ) + ) + + with pytest.raises(ParseError, match="segment"): + ActivityNetTemporalClassificationInput(input_file=input_file) + + +class TestActivityNetTemporalClassificationExportImport: + def test_database_import_export(self, tmp_path: Path) -> None: + input_file = _write_activitynet_database_json(tmp_path / "activity_net.json") + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + + output_path = tmp_path / "activity_net_out.json" + ActivityNetTemporalClassificationDatabaseOutput(output_file=output_path).save( + label_input=label_input + ) + + output_json = json.loads(output_path.read_text()) + expected_json = json.loads(input_file.read_text()) + assert output_json == expected_json + + def test_results_import_export(self, tmp_path: Path) -> None: + input_file = _write_activitynet_results_json(tmp_path / "results.json") + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + + output_path = tmp_path / "results_out.json" + ActivityNetTemporalClassificationResultsOutput(output_file=output_path).save( + label_input=label_input + ) + + output_json = json.loads(output_path.read_text()) + expected_json = json.loads(input_file.read_text()) + assert output_json == expected_json + + +def _write_activitynet_database_json(input_file: Path) -> Path: + data = { + "database": { + "v_test_video": { + "duration": 82.75, + "annotations": [ + { + "label": "Person walking", + "segment": [0.58, 6.16], + }, + { + "label": "Rock climbing", + "segment": [10.0, 20.0], + "score": 0.9, + }, + ], + } + } + } + input_file.write_text(json.dumps(data)) + return input_file + + +def _write_activitynet_results_json(input_file: Path) -> Path: + data = { + "results": { + "v_test_video": [ + { + "label": "Person walking", + "segment": [0.58, 6.16], + "score": 0.75, + } + ] + } + } + input_file.write_text(json.dumps(data)) + return input_file From ed48cc899a73b076f4472b737f0a73f770e65ab6 Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Wed, 8 Jul 2026 11:54:01 +0300 Subject: [PATCH 2/9] remove output --- src/labelformat/formats/__init__.py | 4 - src/labelformat/formats/activitynet.py | 161 ++++-------------- .../model/temporal_classification.py | 12 -- tests/unit/formats/test_activitynet.py | 32 ---- 4 files changed, 33 insertions(+), 176 deletions(-) diff --git a/src/labelformat/formats/__init__.py b/src/labelformat/formats/__init__.py index 905b776..25313c8 100644 --- a/src/labelformat/formats/__init__.py +++ b/src/labelformat/formats/__init__.py @@ -1,7 +1,5 @@ from labelformat.formats.activitynet import ( - ActivityNetTemporalClassificationDatabaseOutput, ActivityNetTemporalClassificationInput, - ActivityNetTemporalClassificationResultsOutput, ) from labelformat.formats.coco import ( COCOInstanceSegmentationInput, @@ -81,9 +79,7 @@ ) __all__ = [ - "ActivityNetTemporalClassificationDatabaseOutput", "ActivityNetTemporalClassificationInput", - "ActivityNetTemporalClassificationResultsOutput", "COCOInstanceSegmentationInput", "COCOInstanceSegmentationOutput", "COCOObjectDetectionInput", diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index c4bbaa9..e8fead9 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -9,7 +9,6 @@ from labelformat.model.category import Category from labelformat.model.temporal_classification import ( TemporalClassificationInput, - TemporalClassificationOutput, TemporalEvent, VideoTemporalClassification, ) @@ -44,101 +43,26 @@ class ActivityNetTemporalClassificationInput( """Import ActivityNet-style temporal classification annotations.""" -class _ActivityNetBaseOutput: - @staticmethod - def add_cli_arguments(parser: ArgumentParser) -> None: - parser.add_argument( - "--output-file", - type=Path, - required=True, - help="Path to output ActivityNet JSON file", - ) - - def __init__(self, output_file: Path) -> None: - self.output_file = output_file - - -class ActivityNetTemporalClassificationResultsOutput( - _ActivityNetBaseOutput, TemporalClassificationOutput -): - """Export ActivityNet submission-style JSON with a top-level ``results`` key.""" - - def save(self, label_input: TemporalClassificationInput) -> None: - data: JsonDict = { - "results": _get_output_results_dict(label_input.get_labels()), - } - self.output_file.parent.mkdir(parents=True, exist_ok=True) - with self.output_file.open("w", encoding="utf-8") as file: - json.dump(data, file, indent=2) - - -class ActivityNetTemporalClassificationDatabaseOutput( - _ActivityNetBaseOutput, TemporalClassificationOutput -): - """Export ActivityNet ground-truth-style JSON with a top-level ``database`` key.""" - - def save(self, label_input: TemporalClassificationInput) -> None: - data: JsonDict = { - "database": _get_output_database_dict(label_input.get_labels()), - } - self.output_file.parent.mkdir(parents=True, exist_ok=True) - with self.output_file.open("w", encoding="utf-8") as file: - json.dump(data, file, indent=2) - - def _parse_activitynet_data( data: JsonDict, ) -> tuple[list[VideoTemporalClassification], list[Category]]: if "database" in data: - return _parse_database(database=data["database"]) - if "results" in data: - return _parse_results(results=data["results"]) - - raise ParseError("ActivityNet JSON must contain a 'database' or 'results' key.") - - -def _parse_database( - database: JsonDict, -) -> tuple[list[VideoTemporalClassification], list[Category]]: - label_names: dict[str, None] = {} - parsed_by_video: list[tuple[str, list[_ParsedEvent], float | None]] = [] - - for video_id, video_entry in database.items(): - if not isinstance(video_entry, dict): - raise ParseError(f"Invalid database entry for video '{video_id}'.") - raw_annotations = video_entry.get("annotations", []) - if not isinstance(raw_annotations, list): - raise ParseError(f"Invalid annotations for video '{video_id}'.") - duration = video_entry.get("duration") - duration_s = float(duration) if duration is not None else None - events = [_parse_event(annotation=annotation) for annotation in raw_annotations] - label_names.update((event.label, None) for event in events) - parsed_by_video.append((str(video_id), events, duration_s)) - - categories = _categories_from_label_names(label_names=label_names) - category_name_to_id = {category.name: category.id for category in categories} - labels = [ - VideoTemporalClassification( - video_id=video_id, - events=_events_from_parsed( - events=events, category_name_to_id=category_name_to_id - ), - duration_s=duration_s, + entries = data["database"] + is_database = True + elif "results" in data: + entries = data["results"] + is_database = False + else: + raise ParseError( + "ActivityNet JSON must contain a 'database' or 'results' key." ) - for video_id, events, duration_s in parsed_by_video - ] - return labels, categories - -def _parse_results( - results: JsonDict, -) -> tuple[list[VideoTemporalClassification], list[Category]]: label_names: dict[str, None] = {} parsed_by_video: list[tuple[str, list[_ParsedEvent]]] = [] - - for video_id, raw_annotations in results.items(): - if not isinstance(raw_annotations, list): - raise ParseError(f"Invalid results entry for video '{video_id}'.") + for video_id, video_entry in entries.items(): + raw_annotations = _extract_annotations( + video_id=video_id, video_entry=video_entry, is_database=is_database + ) events = [_parse_event(annotation=annotation) for annotation in raw_annotations] label_names.update((event.label, None) for event in events) parsed_by_video.append((str(video_id), events)) @@ -151,13 +75,33 @@ def _parse_results( events=_events_from_parsed( events=events, category_name_to_id=category_name_to_id ), - duration_s=None, ) for video_id, events in parsed_by_video ] return labels, categories +def _extract_annotations( + video_id: str, + video_entry: object, + is_database: bool, +) -> list[JsonDict]: + """Extract the raw annotation list for one video. + + In the ``database`` format each entry is a dict with an ``annotations`` list; + in the ``results`` format the entry is the list itself. + """ + if is_database: + if not isinstance(video_entry, dict): + raise ParseError(f"Invalid database entry for video '{video_id}'.") + raw_annotations = video_entry.get("annotations", []) + else: + raw_annotations = video_entry + if not isinstance(raw_annotations, list): + raise ParseError(f"Invalid annotations for video '{video_id}'.") + return raw_annotations + + @dataclass(frozen=True) class _ParsedEvent: label: str @@ -219,42 +163,3 @@ def _events_from_parsed( ) for event in events ] - - -def _get_output_results_dict( - labels: Iterable[VideoTemporalClassification], -) -> JsonDict: - results: JsonDict = {} - for label in labels: - annotations: list[JsonDict] = [] - for event in label.events: - annotation: JsonDict = { - "label": event.category.name, - "segment": [event.start_time_s, event.end_time_s], - } - if event.confidence is not None: - annotation["score"] = event.confidence - annotations.append(annotation) - results[label.video_id] = annotations - return results - - -def _get_output_database_dict( - labels: Iterable[VideoTemporalClassification], -) -> JsonDict: - database: JsonDict = {} - for label in labels: - annotations: list[JsonDict] = [] - for event in label.events: - annotation: JsonDict = { - "label": event.category.name, - "segment": [event.start_time_s, event.end_time_s], - } - if event.confidence is not None: - annotation["score"] = event.confidence - annotations.append(annotation) - video_entry: JsonDict = {"annotations": annotations} - if label.duration_s is not None: - video_entry["duration"] = label.duration_s - database[label.video_id] = video_entry - return database diff --git a/src/labelformat/model/temporal_classification.py b/src/labelformat/model/temporal_classification.py index 2868a45..975b7ce 100644 --- a/src/labelformat/model/temporal_classification.py +++ b/src/labelformat/model/temporal_classification.py @@ -31,7 +31,6 @@ class VideoTemporalClassification: video_id: str events: list[TemporalEvent] - duration_s: float | None = None class TemporalClassificationInput(ABC): @@ -47,14 +46,3 @@ def get_categories(self) -> Iterable[Category]: @abstractmethod def get_labels(self) -> Iterable[VideoTemporalClassification]: raise NotImplementedError() - - -class TemporalClassificationOutput(ABC): - @staticmethod - @abstractmethod - def add_cli_arguments(parser: ArgumentParser) -> None: - raise NotImplementedError() - - @abstractmethod - def save(self, label_input: TemporalClassificationInput) -> None: - raise NotImplementedError() diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py index 950c0d3..7943616 100644 --- a/tests/unit/formats/test_activitynet.py +++ b/tests/unit/formats/test_activitynet.py @@ -4,9 +4,7 @@ import pytest from labelformat.formats.activitynet import ( - ActivityNetTemporalClassificationDatabaseOutput, ActivityNetTemporalClassificationInput, - ActivityNetTemporalClassificationResultsOutput, ) from labelformat.model.category import Category from labelformat.model.temporal_classification import ( @@ -33,7 +31,6 @@ def test_get_labels(self, tmp_path: Path) -> None: assert list(label_input.get_labels()) == [ VideoTemporalClassification( video_id="v_test_video", - duration_s=82.75, events=[ TemporalEvent( category=Category(id=1, name="Person walking"), @@ -59,7 +56,6 @@ def test_get_labels(self, tmp_path: Path) -> None: assert list(label_input.get_labels()) == [ VideoTemporalClassification( video_id="v_test_video", - duration_s=None, events=[ TemporalEvent( category=Category(id=1, name="Person walking"), @@ -96,34 +92,6 @@ def test_rejects_invalid_segment(self, tmp_path: Path) -> None: ActivityNetTemporalClassificationInput(input_file=input_file) -class TestActivityNetTemporalClassificationExportImport: - def test_database_import_export(self, tmp_path: Path) -> None: - input_file = _write_activitynet_database_json(tmp_path / "activity_net.json") - label_input = ActivityNetTemporalClassificationInput(input_file=input_file) - - output_path = tmp_path / "activity_net_out.json" - ActivityNetTemporalClassificationDatabaseOutput(output_file=output_path).save( - label_input=label_input - ) - - output_json = json.loads(output_path.read_text()) - expected_json = json.loads(input_file.read_text()) - assert output_json == expected_json - - def test_results_import_export(self, tmp_path: Path) -> None: - input_file = _write_activitynet_results_json(tmp_path / "results.json") - label_input = ActivityNetTemporalClassificationInput(input_file=input_file) - - output_path = tmp_path / "results_out.json" - ActivityNetTemporalClassificationResultsOutput(output_file=output_path).save( - label_input=label_input - ) - - output_json = json.loads(output_path.read_text()) - expected_json = json.loads(input_file.read_text()) - assert output_json == expected_json - - def _write_activitynet_database_json(input_file: Path) -> Path: data = { "database": { From 4efec1559fdf938f6d57d9fb6252425b1fee7d2e Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Wed, 8 Jul 2026 11:55:35 +0300 Subject: [PATCH 3/9] format --- src/labelformat/formats/__init__.py | 4 +--- src/labelformat/formats/activitynet.py | 4 +--- tests/unit/formats/test_activitynet.py | 4 +--- 3 files changed, 3 insertions(+), 9 deletions(-) diff --git a/src/labelformat/formats/__init__.py b/src/labelformat/formats/__init__.py index 25313c8..0175a75 100644 --- a/src/labelformat/formats/__init__.py +++ b/src/labelformat/formats/__init__.py @@ -1,6 +1,4 @@ -from labelformat.formats.activitynet import ( - ActivityNetTemporalClassificationInput, -) +from labelformat.formats.activitynet import ActivityNetTemporalClassificationInput from labelformat.formats.coco import ( COCOInstanceSegmentationInput, COCOInstanceSegmentationOutput, diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index e8fead9..d756893 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -53,9 +53,7 @@ def _parse_activitynet_data( entries = data["results"] is_database = False else: - raise ParseError( - "ActivityNet JSON must contain a 'database' or 'results' key." - ) + raise ParseError("ActivityNet JSON must contain a 'database' or 'results' key.") label_names: dict[str, None] = {} parsed_by_video: list[tuple[str, list[_ParsedEvent]]] = [] diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py index 7943616..e4cb7c4 100644 --- a/tests/unit/formats/test_activitynet.py +++ b/tests/unit/formats/test_activitynet.py @@ -3,9 +3,7 @@ import pytest -from labelformat.formats.activitynet import ( - ActivityNetTemporalClassificationInput, -) +from labelformat.formats.activitynet import ActivityNetTemporalClassificationInput from labelformat.model.category import Category from labelformat.model.temporal_classification import ( TemporalEvent, From 72506634cbcb15e9fe3abb91031c0d1ae2037106 Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Mon, 13 Jul 2026 10:41:21 +0300 Subject: [PATCH 4/9] read video metadata --- src/labelformat/formats/activitynet.py | 117 ++++++++++++++---- .../model/temporal_classification.py | 4 + tests/unit/formats/test_activitynet.py | 68 ++++++++++ 3 files changed, 167 insertions(+), 22 deletions(-) diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index d756893..7bf8931 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -4,7 +4,7 @@ from argparse import ArgumentParser from dataclasses import dataclass from pathlib import Path -from typing import Iterable +from typing import Any, Iterable from labelformat.model.category import Category from labelformat.model.temporal_classification import ( @@ -24,11 +24,22 @@ def add_cli_arguments(parser: ArgumentParser) -> None: required=True, help="Path to input ActivityNet JSON file", ) + parser.add_argument( + "--input-split", + type=str, + default=None, + help=( + "Only import videos whose 'subset' matches this split " + "(e.g. 'training', 'validation'). Imports all videos if not set." + ), + ) - def __init__(self, input_file: Path) -> None: + def __init__(self, input_file: Path, input_split: str | None = None) -> None: with input_file.open(encoding="utf-8") as file: data = json.load(file) - self._labels, self._categories = _parse_activitynet_data(data=data) + self._labels, self._categories = _parse_activitynet_data( + data=data, split=input_split + ) def get_categories(self) -> Iterable[Category]: yield from self._categories @@ -45,6 +56,7 @@ class ActivityNetTemporalClassificationInput( def _parse_activitynet_data( data: JsonDict, + split: str | None = None, ) -> tuple[list[VideoTemporalClassification], list[Category]]: if "database" in data: entries = data["database"] @@ -56,48 +68,104 @@ def _parse_activitynet_data( raise ParseError("ActivityNet JSON must contain a 'database' or 'results' key.") label_names: dict[str, None] = {} - parsed_by_video: list[tuple[str, list[_ParsedEvent]]] = [] + parsed_by_video: list[tuple[str, list[_ParsedEvent], _VideoMetadata]] = [] for video_id, video_entry in entries.items(): - raw_annotations = _extract_annotations( - video_id=video_id, video_entry=video_entry, is_database=is_database + raw_annotations, meta = _extract_video( + video_id=str(video_id), video_entry=video_entry, is_database=is_database ) - events = [_parse_event(annotation=annotation) for annotation in raw_annotations] + events = [ + _parse_event(annotation=annotation, duration_s=meta.duration_s) + for annotation in raw_annotations + ] label_names.update((event.label, None) for event in events) - parsed_by_video.append((str(video_id), events)) + parsed_by_video.append((str(video_id), events, meta)) categories = _categories_from_label_names(label_names=label_names) category_name_to_id = {category.name: category.id for category in categories} + + parsed_by_video = _filter_by_split(parsed_by_video=parsed_by_video, split=split) labels = [ VideoTemporalClassification( video_id=video_id, events=_events_from_parsed( events=events, category_name_to_id=category_name_to_id ), + duration_s=meta.duration_s, + subset=meta.subset, + resolution=meta.resolution, + url=meta.url, ) - for video_id, events in parsed_by_video + for video_id, events, meta in parsed_by_video ] return labels, categories -def _extract_annotations( +@dataclass(frozen=True) +class _VideoMetadata: + duration_s: float | None + subset: str | None + resolution: str | None + url: str | None + + +def _extract_video( video_id: str, video_entry: object, is_database: bool, -) -> list[JsonDict]: - """Extract the raw annotation list for one video. +) -> tuple[list[JsonDict], _VideoMetadata]: + """Extract the raw annotation list and video metadata for one video. - In the ``database`` format each entry is a dict with an ``annotations`` list; - in the ``results`` format the entry is the list itself. + In the ``database`` format each entry is a dict with an ``annotations`` list and + video metadata (``duration``, ``subset``, ``resolution``, ``url``); in the + ``results`` format the entry is the annotation list itself, without metadata. """ - if is_database: - if not isinstance(video_entry, dict): - raise ParseError(f"Invalid database entry for video '{video_id}'.") - raw_annotations = video_entry.get("annotations", []) - else: - raw_annotations = video_entry + if not is_database: + if not isinstance(video_entry, list): + raise ParseError(f"Invalid annotations for video '{video_id}'.") + return video_entry, _VideoMetadata( + duration_s=None, subset=None, resolution=None, url=None + ) + + if not isinstance(video_entry, dict): + raise ParseError(f"Invalid database entry for video '{video_id}'.") + raw_annotations = video_entry.get("annotations", []) if not isinstance(raw_annotations, list): raise ParseError(f"Invalid annotations for video '{video_id}'.") - return raw_annotations + meta = _VideoMetadata( + duration_s=float(_require_field(video_entry, "duration", video_id)), + subset=_require_field(video_entry, "subset", video_id), + resolution=_require_field(video_entry, "resolution", video_id), + url=_require_field(video_entry, "url", video_id), + ) + return raw_annotations, meta + + +def _require_field(video_entry: JsonDict, key: str, video_id: str) -> Any: + if key not in video_entry: + raise ParseError( + f"Database entry for video '{video_id}' is missing required field '{key}'." + ) + return video_entry[key] + + +def _filter_by_split( + parsed_by_video: list[tuple[str, list[_ParsedEvent], _VideoMetadata]], + split: str | None, +) -> list[tuple[str, list[_ParsedEvent], _VideoMetadata]]: + if split is None: + return parsed_by_video + filtered = [ + video for video in parsed_by_video if video[2].subset == split + ] + if not filtered: + available = sorted( + {meta.subset for _, _, meta in parsed_by_video if meta.subset is not None} + ) + raise ParseError( + f"Split '{split}' not found in ActivityNet data. " + f"Available subsets: {available}." + ) + return filtered @dataclass(frozen=True) @@ -108,7 +176,7 @@ class _ParsedEvent: confidence: float | None -def _parse_event(annotation: JsonDict) -> _ParsedEvent: +def _parse_event(annotation: JsonDict, duration_s: float | None = None) -> _ParsedEvent: label = annotation.get("label") segment = annotation.get("segment") if not isinstance(label, str) or not label: @@ -125,6 +193,11 @@ def _parse_event(annotation: JsonDict) -> _ParsedEvent: f"Invalid segment [{start_time_s}, {end_time_s}] for label '{label}': " "start must be non-negative and less than end." ) + if duration_s is not None and end_time_s > duration_s: + raise ParseError( + f"Invalid segment [{start_time_s}, {end_time_s}] for label '{label}': " + f"end must not exceed the video duration ({duration_s})." + ) confidence = annotation.get("score") if confidence is not None: diff --git a/src/labelformat/model/temporal_classification.py b/src/labelformat/model/temporal_classification.py index 975b7ce..4dcdcea 100644 --- a/src/labelformat/model/temporal_classification.py +++ b/src/labelformat/model/temporal_classification.py @@ -31,6 +31,10 @@ class VideoTemporalClassification: video_id: str events: list[TemporalEvent] + duration_s: float | None = None + subset: str | None = None + resolution: str | None = None + url: str | None = None class TemporalClassificationInput(ABC): diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py index e4cb7c4..b9708c0 100644 --- a/tests/unit/formats/test_activitynet.py +++ b/tests/unit/formats/test_activitynet.py @@ -29,6 +29,10 @@ def test_get_labels(self, tmp_path: Path) -> None: assert list(label_input.get_labels()) == [ VideoTemporalClassification( video_id="v_test_video", + duration_s=82.75, + subset="validation", + resolution="270x480", + url="https://www.youtube.com/watch?v=v_test_video", events=[ TemporalEvent( category=Category(id=1, name="Person walking"), @@ -45,6 +49,67 @@ def test_get_labels(self, tmp_path: Path) -> None: ) ] + def test_filters_by_split(self, tmp_path: Path) -> None: + input_file = _write_activitynet_database_json(tmp_path / "activity_net.json") + + validation = ActivityNetTemporalClassificationInput( + input_file=input_file, input_split="validation" + ) + assert [label.video_id for label in validation.get_labels()] == ["v_test_video"] + + def test_rejects_unknown_split(self, tmp_path: Path) -> None: + input_file = _write_activitynet_database_json(tmp_path / "activity_net.json") + + with pytest.raises(ParseError, match="Split 'training' not found"): + ActivityNetTemporalClassificationInput( + input_file=input_file, input_split="training" + ) + + def test_rejects_segment_exceeding_duration(self, tmp_path: Path) -> None: + input_file = tmp_path / "invalid.json" + input_file.write_text( + json.dumps( + { + "database": { + "v_test_video": { + "duration": 5.0, + "subset": "validation", + "resolution": "270x480", + "url": "https://www.youtube.com/watch?v=v_test_video", + "annotations": [ + {"label": "Person walking", "segment": [1.0, 6.0]}, + ], + } + } + } + ) + ) + + with pytest.raises(ParseError, match="exceed the video duration"): + ActivityNetTemporalClassificationInput(input_file=input_file) + + def test_rejects_missing_required_field(self, tmp_path: Path) -> None: + input_file = tmp_path / "invalid.json" + input_file.write_text( + json.dumps( + { + "database": { + "v_test_video": { + "duration": 82.75, + "resolution": "270x480", + "url": "https://www.youtube.com/watch?v=v_test_video", + "annotations": [ + {"label": "Person walking", "segment": [0.58, 6.16]}, + ], + } + } + } + ) + ) + + with pytest.raises(ParseError, match="missing required field 'subset'"): + ActivityNetTemporalClassificationInput(input_file=input_file) + class TestActivityNetTemporalClassificationResultsInput: def test_get_labels(self, tmp_path: Path) -> None: @@ -95,6 +160,9 @@ def _write_activitynet_database_json(input_file: Path) -> Path: "database": { "v_test_video": { "duration": 82.75, + "subset": "validation", + "resolution": "270x480", + "url": "https://www.youtube.com/watch?v=v_test_video", "annotations": [ { "label": "Person walking", From 219e0551295632c77447201518a4768c6cca6458 Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Mon, 13 Jul 2026 10:42:36 +0300 Subject: [PATCH 5/9] format --- src/labelformat/formats/activitynet.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index 7bf8931..d439f18 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -154,9 +154,7 @@ def _filter_by_split( ) -> list[tuple[str, list[_ParsedEvent], _VideoMetadata]]: if split is None: return parsed_by_video - filtered = [ - video for video in parsed_by_video if video[2].subset == split - ] + filtered = [video for video in parsed_by_video if video[2].subset == split] if not filtered: available = sorted( {meta.subset for _, _, meta in parsed_by_video if meta.subset is not None} From 859c6f532c0e5682922e312c866cf21cb0f3247f Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Mon, 13 Jul 2026 11:41:59 +0300 Subject: [PATCH 6/9] update --- src/labelformat/formats/activitynet.py | 10 ++++++---- tests/unit/formats/test_activitynet.py | 27 ++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index d439f18..afe7bc6 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -116,8 +116,10 @@ def _extract_video( """Extract the raw annotation list and video metadata for one video. In the ``database`` format each entry is a dict with an ``annotations`` list and - video metadata (``duration``, ``subset``, ``resolution``, ``url``); in the - ``results`` format the entry is the annotation list itself, without metadata. + video metadata. ``duration`` and ``subset`` are required (``subset`` is needed + for split filtering); ``resolution`` and ``url`` are optional and default to + ``None``. In the ``results`` format the entry is the annotation list itself, + without metadata. """ if not is_database: if not isinstance(video_entry, list): @@ -134,8 +136,8 @@ def _extract_video( meta = _VideoMetadata( duration_s=float(_require_field(video_entry, "duration", video_id)), subset=_require_field(video_entry, "subset", video_id), - resolution=_require_field(video_entry, "resolution", video_id), - url=_require_field(video_entry, "url", video_id), + resolution=video_entry.get("resolution"), + url=video_entry.get("url"), ) return raw_annotations, meta diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py index b9708c0..1b6f89f 100644 --- a/tests/unit/formats/test_activitynet.py +++ b/tests/unit/formats/test_activitynet.py @@ -110,6 +110,33 @@ def test_rejects_missing_required_field(self, tmp_path: Path) -> None: with pytest.raises(ParseError, match="missing required field 'subset'"): ActivityNetTemporalClassificationInput(input_file=input_file) + def test_optional_metadata_defaults_to_none(self, tmp_path: Path) -> None: + input_file = tmp_path / "no_optional.json" + input_file.write_text( + json.dumps( + { + "database": { + "v_test_video": { + "duration": 82.75, + "subset": "validation", + "annotations": [ + {"label": "Person walking", "segment": [0.58, 6.16]}, + ], + } + } + } + ) + ) + + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + labels = list(label_input.get_labels()) + + assert len(labels) == 1 + assert labels[0].duration_s == 82.75 + assert labels[0].subset == "validation" + assert labels[0].resolution is None + assert labels[0].url is None + class TestActivityNetTemporalClassificationResultsInput: def test_get_labels(self, tmp_path: Path) -> None: From acfa6d6909c9a1678d70ee6bb265ef507843c7e2 Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Mon, 13 Jul 2026 12:13:10 +0300 Subject: [PATCH 7/9] simplify --- src/labelformat/formats/activitynet.py | 114 +++++++++---------------- 1 file changed, 40 insertions(+), 74 deletions(-) diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index afe7bc6..e5416c0 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -4,7 +4,7 @@ from argparse import ArgumentParser from dataclasses import dataclass from pathlib import Path -from typing import Any, Iterable +from typing import Any, Callable, Iterable from labelformat.model.category import Category from labelformat.model.temporal_classification import ( @@ -67,37 +67,35 @@ def _parse_activitynet_data( else: raise ParseError("ActivityNet JSON must contain a 'database' or 'results' key.") - label_names: dict[str, None] = {} - parsed_by_video: list[tuple[str, list[_ParsedEvent], _VideoMetadata]] = [] + # Assign category ids by first appearance across all videos, so that ids stay + # stable regardless of any split filter applied afterwards. + categories: dict[str, Category] = {} + + def category_for(label: str) -> Category: + if label not in categories: + categories[label] = Category(id=len(categories) + 1, name=label) + return categories[label] + + labels = [] for video_id, video_entry in entries.items(): raw_annotations, meta = _extract_video( - video_id=str(video_id), video_entry=video_entry, is_database=is_database + str(video_id), video_entry, is_database ) - events = [ - _parse_event(annotation=annotation, duration_s=meta.duration_s) - for annotation in raw_annotations - ] - label_names.update((event.label, None) for event in events) - parsed_by_video.append((str(video_id), events, meta)) - - categories = _categories_from_label_names(label_names=label_names) - category_name_to_id = {category.name: category.id for category in categories} - - parsed_by_video = _filter_by_split(parsed_by_video=parsed_by_video, split=split) - labels = [ - VideoTemporalClassification( - video_id=video_id, - events=_events_from_parsed( - events=events, category_name_to_id=category_name_to_id - ), - duration_s=meta.duration_s, - subset=meta.subset, - resolution=meta.resolution, - url=meta.url, + labels.append( + VideoTemporalClassification( + video_id=str(video_id), + events=[ + _parse_event(annotation, category_for, meta.duration_s) + for annotation in raw_annotations + ], + duration_s=meta.duration_s, + subset=meta.subset, + resolution=meta.resolution, + url=meta.url, + ) ) - for video_id, events, meta in parsed_by_video - ] - return labels, categories + + return _filter_by_split(labels, split), list(categories.values()) @dataclass(frozen=True) @@ -151,15 +149,15 @@ def _require_field(video_entry: JsonDict, key: str, video_id: str) -> Any: def _filter_by_split( - parsed_by_video: list[tuple[str, list[_ParsedEvent], _VideoMetadata]], + labels: list[VideoTemporalClassification], split: str | None, -) -> list[tuple[str, list[_ParsedEvent], _VideoMetadata]]: +) -> list[VideoTemporalClassification]: if split is None: - return parsed_by_video - filtered = [video for video in parsed_by_video if video[2].subset == split] + return labels + filtered = [label for label in labels if label.subset == split] if not filtered: available = sorted( - {meta.subset for _, _, meta in parsed_by_video if meta.subset is not None} + {label.subset for label in labels if label.subset is not None} ) raise ParseError( f"Split '{split}' not found in ActivityNet data. " @@ -168,15 +166,11 @@ def _filter_by_split( return filtered -@dataclass(frozen=True) -class _ParsedEvent: - label: str - start_time_s: float - end_time_s: float - confidence: float | None - - -def _parse_event(annotation: JsonDict, duration_s: float | None = None) -> _ParsedEvent: +def _parse_event( + annotation: JsonDict, + category_for: Callable[[str], Category], + duration_s: float | None, +) -> TemporalEvent: label = annotation.get("label") segment = annotation.get("segment") if not isinstance(label, str) or not label: @@ -199,38 +193,10 @@ def _parse_event(annotation: JsonDict, duration_s: float | None = None) -> _Pars f"end must not exceed the video duration ({duration_s})." ) - confidence = annotation.get("score") - if confidence is not None: - confidence = float(confidence) - - return _ParsedEvent( - label=label, + score = annotation.get("score") + return TemporalEvent( + category=category_for(label), start_time_s=start_time_s, end_time_s=end_time_s, - confidence=confidence, + confidence=float(score) if score is not None else None, ) - - -def _categories_from_label_names(label_names: Iterable[str]) -> list[Category]: - return [ - Category(id=index, name=label_name) - for index, label_name in enumerate(label_names, start=1) - ] - - -def _events_from_parsed( - events: list[_ParsedEvent], - category_name_to_id: dict[str, int], -) -> list[TemporalEvent]: - return [ - TemporalEvent( - category=Category( - id=category_name_to_id[event.label], - name=event.label, - ), - start_time_s=event.start_time_s, - end_time_s=event.end_time_s, - confidence=event.confidence, - ) - for event in events - ] From 48fa95d40d5e0af286d6a20c9c20cfdffc872ccc Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Mon, 13 Jul 2026 12:20:06 +0300 Subject: [PATCH 8/9] format --- src/labelformat/formats/activitynet.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index e5416c0..478935f 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -78,9 +78,7 @@ def category_for(label: str) -> Category: labels = [] for video_id, video_entry in entries.items(): - raw_annotations, meta = _extract_video( - str(video_id), video_entry, is_database - ) + raw_annotations, meta = _extract_video(str(video_id), video_entry, is_database) labels.append( VideoTemporalClassification( video_id=str(video_id), From 74634c3fa3330106498dc34caa7d6c65e93a6f36 Mon Sep 17 00:00:00 2001 From: Horatiu Almasan Date: Mon, 13 Jul 2026 16:20:25 +0300 Subject: [PATCH 9/9] fix segment rejection due to float comparrison --- src/labelformat/formats/activitynet.py | 13 +++++++---- tests/unit/formats/test_activitynet.py | 31 ++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py index 478935f..e275e67 100644 --- a/src/labelformat/formats/activitynet.py +++ b/src/labelformat/formats/activitynet.py @@ -14,6 +14,8 @@ ) from labelformat.types import JsonDict, ParseError +_DURATION_OVERFLOW_TOLERANCE_S = 0.1 + class _ActivityNetBaseInput: @staticmethod @@ -186,10 +188,13 @@ def _parse_event( "start must be non-negative and less than end." ) if duration_s is not None and end_time_s > duration_s: - raise ParseError( - f"Invalid segment [{start_time_s}, {end_time_s}] for label '{label}': " - f"end must not exceed the video duration ({duration_s})." - ) + if end_time_s - duration_s > _DURATION_OVERFLOW_TOLERANCE_S: + raise ParseError( + f"Invalid segment [{start_time_s}, {end_time_s}] for label " + f"'{label}': end must not exceed the video duration ({duration_s})." + ) + # Rounding overflow within tolerance: clip the end to the duration. + end_time_s = duration_s score = annotation.get("score") return TemporalEvent( diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py index 1b6f89f..54b1d5c 100644 --- a/tests/unit/formats/test_activitynet.py +++ b/tests/unit/formats/test_activitynet.py @@ -88,6 +88,37 @@ def test_rejects_segment_exceeding_duration(self, tmp_path: Path) -> None: with pytest.raises(ParseError, match="exceed the video duration"): ActivityNetTemporalClassificationInput(input_file=input_file) + def test_clips_rounding_overflow_from_official_ground_truth( + self, tmp_path: Path + ) -> None: + # The stored duration is rounded while the segment end is not. This must not be rejected. + input_file = tmp_path / "official.json" + input_file.write_text( + json.dumps( + { + "database": { + "amCD-2TIKw0": { + "duration": 124.18, + "subset": "validation", + "annotations": [ + { + "label": "Rock climbing", + "segment": [10.0, 124.18031746031745], + }, + ], + } + } + } + ) + ) + + label_input = ActivityNetTemporalClassificationInput(input_file=input_file) + events = list(label_input.get_labels())[0].events + + assert len(events) == 1 + # The tiny overflow is clipped back to the video duration. + assert events[0].end_time_s == 124.18 + def test_rejects_missing_required_field(self, tmp_path: Path) -> None: input_file = tmp_path / "invalid.json" input_file.write_text(