diff --git a/src/labelformat/formats/__init__.py b/src/labelformat/formats/__init__.py index 844a330..0175a75 100644 --- a/src/labelformat/formats/__init__.py +++ b/src/labelformat/formats/__init__.py @@ -1,3 +1,4 @@ +from labelformat.formats.activitynet import ActivityNetTemporalClassificationInput from labelformat.formats.coco import ( COCOInstanceSegmentationInput, COCOInstanceSegmentationOutput, @@ -76,6 +77,7 @@ ) __all__ = [ + "ActivityNetTemporalClassificationInput", "COCOInstanceSegmentationInput", "COCOInstanceSegmentationOutput", "COCOObjectDetectionInput", diff --git a/src/labelformat/formats/activitynet.py b/src/labelformat/formats/activitynet.py new file mode 100644 index 0000000..e275e67 --- /dev/null +++ b/src/labelformat/formats/activitynet.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import json +from argparse import ArgumentParser +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Iterable + +from labelformat.model.category import Category +from labelformat.model.temporal_classification import ( + TemporalClassificationInput, + TemporalEvent, + VideoTemporalClassification, +) +from labelformat.types import JsonDict, ParseError + +_DURATION_OVERFLOW_TOLERANCE_S = 0.1 + + +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", + ) + 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, 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, split=input_split + ) + + 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.""" + + +def _parse_activitynet_data( + data: JsonDict, + split: str | None = None, +) -> tuple[list[VideoTemporalClassification], list[Category]]: + if "database" in data: + 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.") + + # 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(str(video_id), video_entry, is_database) + 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, + ) + ) + + return _filter_by_split(labels, split), list(categories.values()) + + +@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, +) -> 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 and + 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): + 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}'.") + meta = _VideoMetadata( + duration_s=float(_require_field(video_entry, "duration", video_id)), + subset=_require_field(video_entry, "subset", video_id), + resolution=video_entry.get("resolution"), + url=video_entry.get("url"), + ) + 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( + labels: list[VideoTemporalClassification], + split: str | None, +) -> list[VideoTemporalClassification]: + if split is None: + return labels + filtered = [label for label in labels if label.subset == split] + if not filtered: + available = sorted( + {label.subset for label in labels if label.subset is not None} + ) + raise ParseError( + f"Split '{split}' not found in ActivityNet data. " + f"Available subsets: {available}." + ) + return filtered + + +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: + 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." + ) + if duration_s is not None and end_time_s > 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( + category=category_for(label), + start_time_s=start_time_s, + end_time_s=end_time_s, + confidence=float(score) if score is not None else None, + ) diff --git a/src/labelformat/model/temporal_classification.py b/src/labelformat/model/temporal_classification.py new file mode 100644 index 0000000..4dcdcea --- /dev/null +++ b/src/labelformat/model/temporal_classification.py @@ -0,0 +1,52 @@ +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 + subset: str | None = None + resolution: str | None = None + url: str | 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() diff --git a/tests/unit/formats/test_activitynet.py b/tests/unit/formats/test_activitynet.py new file mode 100644 index 0000000..54b1d5c --- /dev/null +++ b/tests/unit/formats/test_activitynet.py @@ -0,0 +1,255 @@ +import json +from pathlib import Path + +import pytest + +from labelformat.formats.activitynet import ActivityNetTemporalClassificationInput +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, + subset="validation", + resolution="270x480", + url="https://www.youtube.com/watch?v=v_test_video", + 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, + ), + ], + ) + ] + + 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_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( + 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) + + 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: + 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", + 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) + + +def _write_activitynet_database_json(input_file: Path) -> Path: + data = { + "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", + "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