From 00961132a9330f8021ce65eaa376321df51451bc Mon Sep 17 00:00:00 2001 From: mbsantiago Date: Fri, 3 Apr 2026 17:07:26 +0100 Subject: [PATCH] Improve test suite for conditions --- tests/test_data/test_conditions/test_clip.py | 267 +++++-- .../test_conditions/test_recording.py | 287 +++++-- .../test_conditions/test_sound_events.py | 740 ++++++++---------- 3 files changed, 757 insertions(+), 537 deletions(-) diff --git a/tests/test_data/test_conditions/test_clip.py b/tests/test_data/test_conditions/test_clip.py index e8c32f3..dc9a642 100644 --- a/tests/test_data/test_conditions/test_clip.py +++ b/tests/test_data/test_conditions/test_clip.py @@ -1,21 +1,38 @@ import json +import textwrap +import uuid from pathlib import Path +from pydantic import TypeAdapter from soundevent import data +from batdetect2.core import load_config from batdetect2.data.conditions import ( - ClipAllOfConfig, - ClipAnyOfConfig, - ClipNotConfig, - HasAllTagsConfig, - HasAnyTagConfig, - HasTagConfig, - IdInListConfig, - RecordingSatisfiesConfig, + ClipAnnotationConditionConfig, build_clip_annotation_condition, ) +def load_clip_condition_config( + tmp_path: Path, + yaml_string: str, +) -> ClipAnnotationConditionConfig: + config_path = tmp_path / f"{uuid.uuid4().hex}.yaml" + config_path.write_text(textwrap.dedent(yaml_string).strip()) + return load_config( + config_path, schema=TypeAdapter(ClipAnnotationConditionConfig) + ) + + +def build_clip_condition_from_yaml( + tmp_path: Path, + yaml_string: str, + base_dir: Path | None = None, +): + config = load_clip_condition_config(tmp_path, yaml_string) + return build_clip_annotation_condition(config, base_dir=base_dir) + + def test_recording_satisfies_condition( tmp_path: Path, create_recording, @@ -31,10 +48,14 @@ def test_recording_satisfies_condition( ids_path = tmp_path / "recording_ids.json" ids_path.write_text(json.dumps([str(recording_a.uuid)])) - condition = build_clip_annotation_condition( - RecordingSatisfiesConfig( - condition=IdInListConfig(path=ids_path), - ) + condition = build_clip_condition_from_yaml( + tmp_path, + f""" + name: recording_satisfies + condition: + name: id_in_list + path: {ids_path} + """, ) assert condition(clip_annotation_a) @@ -54,7 +75,13 @@ def test_clip_id_in_list_condition( ids_path = tmp_path / "clip_annotation_ids.json" ids_path.write_text(json.dumps([str(clip_annotation_a.uuid)])) - condition = build_clip_annotation_condition(IdInListConfig(path=ids_path)) + condition = build_clip_condition_from_yaml( + tmp_path, + f""" + name: id_in_list + path: {ids_path} + """, + ) assert condition(clip_annotation_a) assert not condition(clip_annotation_b) @@ -68,7 +95,6 @@ def test_clip_has_tag_conditions( ) -> None: reviewed = data.Tag(key="status", value="reviewed") train = data.Tag(key="split", value="train") - val = data.Tag(key="split", value="val") recording = create_recording(path=tmp_path / "rec.wav") clip = create_clip(recording) @@ -76,21 +102,96 @@ def test_clip_has_tag_conditions( clip, clip_tags=[reviewed, train], ) - - has_tag = build_clip_annotation_condition(HasTagConfig(tag=reviewed)) - has_all = build_clip_annotation_condition( - HasAllTagsConfig(tags=[reviewed, train]) - ) - has_any = build_clip_annotation_condition( - HasAnyTagConfig(tags=[val, train]) + clip_annotation_missing = create_clip_annotation( + create_clip(recording), + clip_tags=[train], ) - assert has_tag(clip_annotation) - assert has_all(clip_annotation) - assert has_any(clip_annotation) + condition = build_clip_condition_from_yaml( + tmp_path, + """ + name: has_tag + tag: + key: status + value: reviewed + """, + ) + + assert condition(clip_annotation) + assert not condition(clip_annotation_missing) -def test_clip_logical_conditions( +def test_clip_has_all_tags_condition( + tmp_path: Path, + create_recording, + create_clip, + create_clip_annotation, +) -> None: + reviewed = data.Tag(key="status", value="reviewed") + train = data.Tag(key="split", value="train") + + recording = create_recording(path=tmp_path / "rec.wav") + clip_annotation = create_clip_annotation( + create_clip(recording), + clip_tags=[reviewed, train], + ) + clip_annotation_missing = create_clip_annotation( + create_clip(recording), + clip_tags=[reviewed], + ) + + condition = build_clip_condition_from_yaml( + tmp_path, + """ + name: has_all_tags + tags: + - key: status + value: reviewed + - key: split + value: train + """, + ) + + assert condition(clip_annotation) + assert not condition(clip_annotation_missing) + + +def test_clip_has_any_tag_condition( + tmp_path: Path, + create_recording, + create_clip, + create_clip_annotation, +) -> None: + reviewed = data.Tag(key="status", value="reviewed") + train = data.Tag(key="split", value="train") + + recording = create_recording(path=tmp_path / "rec.wav") + clip_annotation = create_clip_annotation( + create_clip(recording), + clip_tags=[reviewed, train], + ) + clip_annotation_missing = create_clip_annotation( + create_clip(recording), + clip_tags=[data.Tag(key="split", value="test")], + ) + + condition = build_clip_condition_from_yaml( + tmp_path, + """ + name: has_any_tag + tags: + - key: split + value: val + - key: split + value: train + """, + ) + + assert condition(clip_annotation) + assert not condition(clip_annotation_missing) + + +def test_clip_all_of_condition( tmp_path: Path, create_recording, create_clip, @@ -98,7 +199,6 @@ def test_clip_logical_conditions( ) -> None: reviewed = data.Tag(key="status", value="reviewed") train = data.Tag(key="split", value="train") - val = data.Tag(key="split", value="val") recording = create_recording(path=tmp_path / "rec.wav") clip = create_clip(recording) @@ -106,27 +206,98 @@ def test_clip_logical_conditions( clip, clip_tags=[reviewed, train], ) - - all_condition = build_clip_annotation_condition( - ClipAllOfConfig( - conditions=[ - HasTagConfig(tag=reviewed), - HasAnyTagConfig(tags=[train, val]), - ] - ) - ) - any_condition = build_clip_annotation_condition( - ClipAnyOfConfig( - conditions=[ - HasTagConfig(tag=val), - HasTagConfig(tag=reviewed), - ] - ) - ) - not_condition = build_clip_annotation_condition( - ClipNotConfig(condition=HasTagConfig(tag=val)) + clip_annotation_missing = create_clip_annotation( + create_clip(recording), + clip_tags=[reviewed], ) - assert all_condition(clip_annotation) - assert any_condition(clip_annotation) - assert not_condition(clip_annotation) + condition = build_clip_condition_from_yaml( + tmp_path, + """ + name: all_of + conditions: + - name: has_tag + tag: + key: status + value: reviewed + - name: has_any_tag + tags: + - key: split + value: train + - key: split + value: val + """, + ) + + assert condition(clip_annotation) + assert not condition(clip_annotation_missing) + + +def test_clip_any_of_condition( + tmp_path: Path, + create_recording, + create_clip, + create_clip_annotation, +) -> None: + reviewed = data.Tag(key="status", value="reviewed") + + recording = create_recording(path=tmp_path / "rec.wav") + clip_annotation = create_clip_annotation( + create_clip(recording), + clip_tags=[reviewed], + ) + clip_annotation_missing = create_clip_annotation( + create_clip(recording), + clip_tags=[data.Tag(key="status", value="unchecked")], + ) + + condition = build_clip_condition_from_yaml( + tmp_path, + """ + name: any_of + conditions: + - name: has_tag + tag: + key: split + value: val + - name: has_tag + tag: + key: status + value: reviewed + """, + ) + + assert condition(clip_annotation) + assert not condition(clip_annotation_missing) + + +def test_clip_not_condition( + tmp_path: Path, + create_recording, + create_clip, + create_clip_annotation, +) -> None: + recording = create_recording(path=tmp_path / "rec.wav") + clip_annotation = create_clip_annotation( + create_clip(recording), + clip_tags=[data.Tag(key="split", value="train")], + ) + clip_annotation_missing = create_clip_annotation( + create_clip(recording), + clip_tags=[data.Tag(key="split", value="val")], + ) + + condition = build_clip_condition_from_yaml( + tmp_path, + """ + name: not + condition: + name: has_tag + tag: + key: split + value: val + """, + ) + + assert condition(clip_annotation) + assert not condition(clip_annotation_missing) diff --git a/tests/test_data/test_conditions/test_recording.py b/tests/test_data/test_conditions/test_recording.py index 526106e..2290bdb 100644 --- a/tests/test_data/test_conditions/test_recording.py +++ b/tests/test_data/test_conditions/test_recording.py @@ -1,28 +1,53 @@ import json +import textwrap +import uuid from pathlib import Path import pytest +from pydantic import TypeAdapter from soundevent import data +from batdetect2.core import load_config from batdetect2.data.conditions import ( - HasAllTagsConfig, - HasAnyTagConfig, - HasTagConfig, - IdInListConfig, - RecordingAllOfConfig, - RecordingAnyOfConfig, - RecordingNotConfig, + RecordingConditionConfig, build_recording_condition, ) +def load_recording_condition_config( + tmp_path: Path, + yaml_string: str, +) -> RecordingConditionConfig: + config_path = tmp_path / f"{uuid.uuid4().hex}.yaml" + config_path.write_text(textwrap.dedent(yaml_string).strip()) + return load_config( + config_path, + schema=TypeAdapter(RecordingConditionConfig), + ) + + +def build_recording_condition_from_yaml( + tmp_path: Path, + yaml_string: str, + base_dir: Path | None = None, +): + config = load_recording_condition_config(tmp_path, yaml_string) + return build_recording_condition(config, base_dir=base_dir) + + def test_id_in_list_condition(tmp_path: Path, create_recording) -> None: recording_a = create_recording(path=tmp_path / "a.wav") recording_b = create_recording(path=tmp_path / "b.wav") ids_path = tmp_path / "recording_ids.json" ids_path.write_text(json.dumps([str(recording_a.uuid)])) - condition = build_recording_condition(IdInListConfig(path=ids_path)) + condition = build_recording_condition_from_yaml( + tmp_path, + f""" + name: id_in_list + path: {ids_path} + """, + ) assert condition(recording_a) assert not condition(recording_b) @@ -32,18 +57,24 @@ def test_id_in_list_condition_uses_base_dir( tmp_path: Path, create_recording, ) -> None: - recording = create_recording(path=tmp_path / "a.wav") + recording_a = create_recording(path=tmp_path / "a.wav") + recording_b = create_recording(path=tmp_path / "b.wav") split_dir = tmp_path / "splits" split_dir.mkdir() ids_path = split_dir / "train_ids.json" - ids_path.write_text(json.dumps([str(recording.uuid)])) + ids_path.write_text(json.dumps([str(recording_a.uuid)])) - condition = build_recording_condition( - IdInListConfig(path=Path("splits/train_ids.json")), + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: id_in_list + path: splits/train_ids.json + """, base_dir=tmp_path, ) - assert condition(recording) + assert condition(recording_a) + assert not condition(recording_b) def test_id_in_list_condition_raises_for_non_list_json( @@ -53,7 +84,13 @@ def test_id_in_list_condition_raises_for_non_list_json( ids_path.write_text(json.dumps({"id": "foo"})) with pytest.raises(TypeError, match="Expected JSON list"): - build_recording_condition(IdInListConfig(path=ids_path)) + build_recording_condition_from_yaml( + tmp_path, + f""" + name: id_in_list + path: {ids_path} + """, + ) def test_id_in_list_condition_raises_for_invalid_id(tmp_path: Path) -> None: @@ -61,7 +98,13 @@ def test_id_in_list_condition_raises_for_invalid_id(tmp_path: Path) -> None: ids_path.write_text(json.dumps(["not-a-uuid"])) with pytest.raises(ValueError, match="Invalid ID"): - build_recording_condition(IdInListConfig(path=ids_path)) + build_recording_condition_from_yaml( + tmp_path, + f""" + name: id_in_list + path: {ids_path} + """, + ) def test_id_in_list_condition_supports_txt_format( @@ -73,67 +116,197 @@ def test_id_in_list_condition_supports_txt_format( ids_path = tmp_path / "recording_ids.txt" ids_path.write_text(f"{recording_a.uuid}\n") - condition = build_recording_condition( - IdInListConfig(path=ids_path, list_format="txt") + condition = build_recording_condition_from_yaml( + tmp_path, + f""" + name: id_in_list + path: {ids_path} + list_format: txt + """, ) assert condition(recording_a) assert not condition(recording_b) -def test_recording_has_tag_conditions( - tmp_path: Path, create_recording -) -> None: +def test_has_tag_condition(tmp_path: Path, create_recording) -> None: + train = data.Tag(key="split", value="train") + val = data.Tag(key="split", value="val") + + recording_a = create_recording( + path=tmp_path / "a.wav", + tags=[train], + ) + recording_b = create_recording( + path=tmp_path / "b.wav", + tags=[val], + ) + + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: has_tag + tag: + key: split + value: train + """, + ) + + assert condition(recording_a) + assert not condition(recording_b) + + +def test_has_all_tags_condition(tmp_path: Path, create_recording) -> None: train = data.Tag(key="split", value="train") uk = data.Tag(key="region", value="uk") - eu = data.Tag(key="region", value="eu") - recording = create_recording( - path=tmp_path / "rec.wav", + recording_a = create_recording( + path=tmp_path / "a.wav", tags=[train, uk], ) + recording_b = create_recording( + path=tmp_path / "b.wav", + tags=[train], + ) - has_train = build_recording_condition(HasTagConfig(tag=train)) - has_all = build_recording_condition(HasAllTagsConfig(tags=[train, uk])) - has_any = build_recording_condition(HasAnyTagConfig(tags=[eu, uk])) + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: has_all_tags + tags: + - key: split + value: train + - key: region + value: uk + """, + ) - assert has_train(recording) - assert has_all(recording) - assert has_any(recording) + assert condition(recording_a) + assert not condition(recording_b) -def test_recording_logical_conditions( - tmp_path: Path, create_recording -) -> None: +def test_has_any_tag_condition(tmp_path: Path, create_recording) -> None: + uk = data.Tag(key="region", value="uk") + us = data.Tag(key="region", value="us") + + recording_a = create_recording( + path=tmp_path / "a.wav", + tags=[uk], + ) + recording_b = create_recording( + path=tmp_path / "b.wav", + tags=[us], + ) + + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: has_any_tag + tags: + - key: region + value: eu + - key: region + value: uk + """, + ) + + assert condition(recording_a) + assert not condition(recording_b) + + +def test_all_of_condition(tmp_path: Path, create_recording) -> None: train = data.Tag(key="split", value="train") uk = data.Tag(key="region", value="uk") - eu = data.Tag(key="region", value="eu") + us = data.Tag(key="region", value="us") - recording = create_recording( - path=tmp_path / "rec.wav", + recording_a = create_recording( + path=tmp_path / "a.wav", tags=[train, uk], ) - - all_condition = build_recording_condition( - RecordingAllOfConfig( - conditions=[ - HasTagConfig(tag=train), - HasAnyTagConfig(tags=[eu, uk]), - ] - ) - ) - any_condition = build_recording_condition( - RecordingAnyOfConfig( - conditions=[ - HasTagConfig(tag=eu), - HasTagConfig(tag=train), - ] - ) - ) - not_condition = build_recording_condition( - RecordingNotConfig(condition=HasTagConfig(tag=eu)) + recording_b = create_recording( + path=tmp_path / "b.wav", + tags=[train, us], ) - assert all_condition(recording) - assert any_condition(recording) - assert not_condition(recording) + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: all_of + conditions: + - name: has_tag + tag: + key: split + value: train + - name: has_any_tag + tags: + - key: region + value: eu + - key: region + value: uk + """, + ) + + assert condition(recording_a) + assert not condition(recording_b) + + +def test_any_of_condition(tmp_path: Path, create_recording) -> None: + train = data.Tag(key="split", value="train") + us = data.Tag(key="region", value="us") + + recording_a = create_recording( + path=tmp_path / "a.wav", + tags=[train], + ) + recording_b = create_recording( + path=tmp_path / "b.wav", + tags=[us], + ) + + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: any_of + conditions: + - name: has_tag + tag: + key: region + value: eu + - name: has_tag + tag: + key: split + value: train + """, + ) + + assert condition(recording_a) + assert not condition(recording_b) + + +def test_not_condition(tmp_path: Path, create_recording) -> None: + uk = data.Tag(key="region", value="uk") + us = data.Tag(key="region", value="us") + + recording_a = create_recording( + path=tmp_path / "a.wav", + tags=[uk], + ) + recording_b = create_recording( + path=tmp_path / "b.wav", + tags=[us], + ) + + condition = build_recording_condition_from_yaml( + tmp_path, + """ + name: not + condition: + name: has_tag + tag: + key: region + value: us + """, + ) + + assert condition(recording_a) + assert not condition(recording_b) diff --git a/tests/test_data/test_conditions/test_sound_events.py b/tests/test_data/test_conditions/test_sound_events.py index f7cf7f8..9e7893f 100644 --- a/tests/test_data/test_conditions/test_sound_events.py +++ b/tests/test_data/test_conditions/test_sound_events.py @@ -1,524 +1,400 @@ import json import textwrap +import uuid from pathlib import Path import pytest -import yaml from pydantic import TypeAdapter from soundevent import data +from batdetect2.core import load_config from batdetect2.data.conditions import ( - IdInListConfig, SoundEventConditionConfig, build_sound_event_condition, ) -def build_condition_from_str(content, base_dir: Path | None = None): - content = textwrap.dedent(content) - content = yaml.safe_load(content) - config = TypeAdapter(SoundEventConditionConfig).validate_python(content) +def load_sound_event_condition_config( + tmp_path: Path, + yaml_string: str, +) -> SoundEventConditionConfig: + config_path = tmp_path / f"{uuid.uuid4().hex}.yaml" + config_path.write_text(textwrap.dedent(yaml_string).strip()) + return load_config( + config_path, + schema=TypeAdapter(SoundEventConditionConfig), + ) + + +def build_condition_from_str( + tmp_path: Path, + yaml_string: str, + base_dir: Path | None = None, +): + config = load_sound_event_condition_config(tmp_path, yaml_string) return build_sound_event_condition(config, base_dir=base_dir) -def test_has_tag(sound_event: data.SoundEvent): - condition = build_condition_from_str(""" - name: has_tag - tag: - key: species - value: Myotis myotis - """) +def create_sound_event_annotation( + recording: data.Recording, + geometry: data.Geometry, + tags: list[data.Tag] | None = None, +) -> data.SoundEventAnnotation: + return data.SoundEventAnnotation( + sound_event=data.SoundEvent( + recording=recording, + geometry=geometry, + ), + tags=tags or [], + ) - sound_event_annotation = data.SoundEventAnnotation( + +def test_has_tag_condition( + sound_event: data.SoundEvent, tmp_path: Path +) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: has_tag + tag: + key: species + value: Myotis myotis + """, + ) + + passing = data.SoundEventAnnotation( sound_event=sound_event, tags=[data.Tag(key="species", value="Myotis myotis")], ) - assert condition(sound_event_annotation) - - sound_event_annotation = data.SoundEventAnnotation( + failing = data.SoundEventAnnotation( sound_event=sound_event, tags=[data.Tag(key="species", value="Eptesicus fuscus")], ) - assert not condition(sound_event_annotation) + + assert condition(passing) + assert not condition(failing) -def test_has_all_tags(sound_event: data.SoundEvent): - condition = build_condition_from_str(""" - name: has_all_tags - tags: - - key: species - value: Myotis myotis - - key: event - value: Echolocation - """) +def test_has_all_tags_condition( + sound_event: data.SoundEvent, + tmp_path: Path, +) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: has_all_tags + tags: + - key: species + value: Myotis myotis + - key: event + value: Echolocation + """, + ) - sound_event_annotation = data.SoundEventAnnotation( + passing = data.SoundEventAnnotation( + sound_event=sound_event, + tags=[ + data.Tag(key="species", value="Myotis myotis"), + data.Tag(key="event", value="Echolocation"), + ], + ) + failing = data.SoundEventAnnotation( sound_event=sound_event, tags=[data.Tag(key="species", value="Myotis myotis")], ) - assert not condition(sound_event_annotation) - sound_event_annotation = data.SoundEventAnnotation( - sound_event=sound_event, - tags=[ - data.Tag(key="species", value="Eptesicus fuscus"), - data.Tag(key="event", value="Echolocation"), - ], + assert condition(passing) + assert not condition(failing) + + +def test_has_any_tag_condition( + sound_event: data.SoundEvent, + tmp_path: Path, +) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: has_any_tag + tags: + - key: species + value: Myotis myotis + - key: event + value: Echolocation + """, ) - assert not condition(sound_event_annotation) - sound_event_annotation = data.SoundEventAnnotation( + passing = data.SoundEventAnnotation( sound_event=sound_event, - tags=[ - data.Tag(key="species", value="Myotis myotis"), - data.Tag(key="event", value="Echolocation"), - ], + tags=[data.Tag(key="event", value="Echolocation")], ) - assert condition(sound_event_annotation) - - sound_event_annotation = data.SoundEventAnnotation( - sound_event=sound_event, - tags=[ - data.Tag(key="species", value="Myotis myotis"), - data.Tag(key="event", value="Echolocation"), - data.Tag(key="sex", value="Female"), - ], - ) - assert condition(sound_event_annotation) - - -def test_has_any_tags(sound_event: data.SoundEvent): - condition = build_condition_from_str(""" - name: has_any_tag - tags: - - key: species - value: Myotis myotis - - key: event - value: Echolocation - """) - - sound_event_annotation = data.SoundEventAnnotation( - sound_event=sound_event, - tags=[data.Tag(key="species", value="Myotis myotis")], - ) - assert condition(sound_event_annotation) - - sound_event_annotation = data.SoundEventAnnotation( - sound_event=sound_event, - tags=[ - data.Tag(key="species", value="Eptesicus fuscus"), - data.Tag(key="event", value="Echolocation"), - ], - ) - assert condition(sound_event_annotation) - - sound_event_annotation = data.SoundEventAnnotation( - sound_event=sound_event, - tags=[ - data.Tag(key="species", value="Myotis myotis"), - data.Tag(key="event", value="Echolocation"), - ], - ) - assert condition(sound_event_annotation) - - sound_event_annotation = data.SoundEventAnnotation( + failing = data.SoundEventAnnotation( sound_event=sound_event, tags=[ data.Tag(key="species", value="Eptesicus fuscus"), data.Tag(key="event", value="Social"), ], ) - assert not condition(sound_event_annotation) + + assert condition(passing) + assert not condition(failing) -def test_not(sound_event: data.SoundEvent): - condition = build_condition_from_str(""" - name: not - condition: - name: has_tag - tag: - key: species - value: Myotis myotis - """) - - sound_event_annotation = data.SoundEventAnnotation( - sound_event=sound_event, - tags=[data.Tag(key="species", value="Myotis myotis")], +def test_not_condition(sound_event: data.SoundEvent, tmp_path: Path) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: not + condition: + name: has_tag + tag: + key: species + value: Myotis myotis + """, ) - assert not condition(sound_event_annotation) - sound_event_annotation = data.SoundEventAnnotation( + passing = data.SoundEventAnnotation( sound_event=sound_event, tags=[data.Tag(key="species", value="Eptesicus fuscus")], ) - assert condition(sound_event_annotation) - - sound_event_annotation = data.SoundEventAnnotation( + failing = data.SoundEventAnnotation( sound_event=sound_event, - tags=[ - data.Tag(key="species", value="Myotis myotis"), - data.Tag(key="event", value="Echolocation"), - ], + tags=[data.Tag(key="species", value="Myotis myotis")], ) - assert not condition(sound_event_annotation) + + assert condition(passing) + assert not condition(failing) -def test_id_in_list(sound_event: data.SoundEvent, tmp_path: Path): - se1 = data.SoundEventAnnotation(sound_event=sound_event) - se2 = data.SoundEventAnnotation(sound_event=sound_event) +def test_id_in_list_condition( + sound_event: data.SoundEvent, tmp_path: Path +) -> None: + passing = data.SoundEventAnnotation(sound_event=sound_event) + failing = data.SoundEventAnnotation(sound_event=sound_event) ids_path = tmp_path / "sound_event_ids.json" - ids_path.write_text(json.dumps([str(se1.uuid)])) + ids_path.write_text(json.dumps([str(passing.uuid)])) - condition = build_sound_event_condition(IdInListConfig(path=ids_path)) + condition = build_condition_from_str( + tmp_path, + f""" + name: id_in_list + path: {ids_path} + """, + ) - assert condition(se1) - assert not condition(se2) + assert condition(passing) + assert not condition(failing) -def test_id_in_list_uses_base_dir( +def test_id_in_list_condition_uses_base_dir( sound_event: data.SoundEvent, tmp_path: Path, ) -> None: - se = data.SoundEventAnnotation(sound_event=sound_event) + passing = data.SoundEventAnnotation(sound_event=sound_event) + failing = data.SoundEventAnnotation(sound_event=sound_event) split_dir = tmp_path / "splits" split_dir.mkdir() ids_path = split_dir / "sound_event_ids.json" - ids_path.write_text(json.dumps([str(se.uuid)])) + ids_path.write_text(json.dumps([str(passing.uuid)])) - condition = build_sound_event_condition( - IdInListConfig(path=Path("splits/sound_event_ids.json")), + condition = build_condition_from_str( + tmp_path, + """ + name: id_in_list + path: splits/sound_event_ids.json + """, base_dir=tmp_path, ) - assert condition(se) + assert condition(passing) + assert not condition(failing) -def test_duration(recording: data.Recording): - se1 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, geometry=data.TimeInterval(coordinates=[0, 1]) +@pytest.mark.parametrize( + "operator,seconds,passing_duration,failing_duration", + [ + ("lt", 2, 1, 2), + ("lte", 2, 2, 3), + ("gt", 2, 3, 2), + ("gte", 2, 2, 1), + ("eq", 2, 2, 3), + ], +) +def test_duration_condition( + tmp_path: Path, + recording: data.Recording, + operator: str, + seconds: int, + passing_duration: int, + failing_duration: int, +) -> None: + condition = build_condition_from_str( + tmp_path, + f""" + name: duration + operator: {operator} + seconds: {seconds} + """, + ) + + passing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, passing_duration]), + ) + failing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, failing_duration]), + ) + + assert condition(passing) + assert not condition(failing) + + +@pytest.mark.parametrize( + "boundary,operator,hertz,passing_bbox,failing_bbox", + [ + ("high", "lt", 300, [0, 100, 1, 200], [0, 100, 1, 300]), + ("high", "lte", 300, [0, 100, 1, 300], [0, 100, 1, 400]), + ("high", "gt", 300, [0, 100, 1, 400], [0, 100, 1, 300]), + ("high", "gte", 300, [0, 100, 1, 300], [0, 100, 1, 200]), + ("high", "eq", 300, [0, 100, 1, 300], [0, 100, 1, 400]), + ("low", "lt", 200, [0, 100, 1, 400], [0, 200, 1, 400]), + ("low", "lte", 200, [0, 200, 1, 400], [0, 300, 1, 400]), + ("low", "gt", 200, [0, 300, 1, 400], [0, 200, 1, 400]), + ("low", "gte", 200, [0, 200, 1, 400], [0, 100, 1, 400]), + ("low", "eq", 200, [0, 200, 1, 400], [0, 300, 1, 400]), + ], +) +def test_frequency_condition( + tmp_path: Path, + recording: data.Recording, + boundary: str, + operator: str, + hertz: int, + passing_bbox: list[int], + failing_bbox: list[int], +) -> None: + condition = build_condition_from_str( + tmp_path, + f""" + name: frequency + boundary: {boundary} + operator: {operator} + hertz: {hertz} + """, + ) + + passing = create_sound_event_annotation( + recording=recording, + geometry=data.BoundingBox( + coordinates=[float(value) for value in passing_bbox] ), ) - se2 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, geometry=data.TimeInterval(coordinates=[0, 2]) - ), - ) - se3 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, geometry=data.TimeInterval(coordinates=[0, 3]) + failing = create_sound_event_annotation( + recording=recording, + geometry=data.BoundingBox( + coordinates=[float(value) for value in failing_bbox] ), ) - condition = build_condition_from_str(""" - name: duration - operator: lt - seconds: 2 - """) - assert condition(se1) - assert not condition(se2) - assert not condition(se3) - - condition = build_condition_from_str(""" - name: duration - operator: lte - seconds: 2 - """) - - assert condition(se1) - assert condition(se2) - assert not condition(se3) - - condition = build_condition_from_str(""" - name: duration - operator: gt - seconds: 2 - """) - - assert not condition(se1) - assert not condition(se2) - assert condition(se3) - - condition = build_condition_from_str(""" - name: duration - operator: gte - seconds: 2 - """) - - assert not condition(se1) - assert condition(se2) - assert condition(se3) - - condition = build_condition_from_str(""" - name: duration - operator: eq - seconds: 2 - """) - - assert not condition(se1) - assert condition(se2) - assert not condition(se3) + assert condition(passing) + assert not condition(failing) -def test_frequency(recording: data.Recording): - se12 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, - geometry=data.BoundingBox(coordinates=[0, 100, 1, 200]), - ), - ) - se13 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, - geometry=data.BoundingBox(coordinates=[0, 100, 2, 300]), - ), - ) - se14 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, - geometry=data.BoundingBox(coordinates=[0, 100, 3, 400]), - ), - ) - se24 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, - geometry=data.BoundingBox(coordinates=[0, 200, 3, 400]), - ), - ) - se34 = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - recording=recording, - geometry=data.BoundingBox(coordinates=[0, 300, 3, 400]), - ), +def test_frequency_condition_is_false_for_temporal_geometries( + tmp_path: Path, + recording: data.Recording, +) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: frequency + boundary: low + operator: eq + hertz: 200 + """, ) - condition = build_condition_from_str(""" - name: frequency - boundary: high - operator: lt - hertz: 300 - """) - assert condition(se12) - assert not condition(se13) - assert not condition(se14) + passing = create_sound_event_annotation( + recording=recording, + geometry=data.BoundingBox(coordinates=[0, 200, 1, 400]), + ) + failing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, 3]), + ) - condition = build_condition_from_str(""" - name: frequency - boundary: high - operator: lte - hertz: 300 - """) - - assert condition(se12) - assert condition(se13) - assert not condition(se14) - - condition = build_condition_from_str(""" - name: frequency - boundary: high - operator: gt - hertz: 300 - """) - - assert not condition(se12) - assert not condition(se13) - assert condition(se14) - - condition = build_condition_from_str(""" - name: frequency - boundary: high - operator: gte - hertz: 300 - """) - - assert not condition(se12) - assert condition(se13) - assert condition(se14) - - condition = build_condition_from_str(""" - name: frequency - boundary: high - operator: eq - hertz: 300 - """) - - assert not condition(se12) - assert condition(se13) - assert not condition(se14) - - # LOW - - condition = build_condition_from_str(""" - name: frequency - boundary: low - operator: lt - hertz: 200 - """) - assert condition(se14) - assert not condition(se24) - assert not condition(se34) - - condition = build_condition_from_str(""" - name: frequency - boundary: low - operator: lte - hertz: 200 - """) - - assert condition(se14) - assert condition(se24) - assert not condition(se34) - - condition = build_condition_from_str(""" - name: frequency - boundary: low - operator: gt - hertz: 200 - """) - - assert not condition(se14) - assert not condition(se24) - assert condition(se34) - - condition = build_condition_from_str(""" - name: frequency - boundary: low - operator: gte - hertz: 200 - """) - - assert not condition(se14) - assert condition(se24) - assert condition(se34) - - condition = build_condition_from_str(""" - name: frequency - boundary: low - operator: eq - hertz: 200 - """) - - assert not condition(se14) - assert condition(se24) - assert not condition(se34) + assert condition(passing) + assert not condition(failing) -def test_frequency_is_false_for_temporal_geometries(recording: data.Recording): - condition = build_condition_from_str(""" - name: frequency - boundary: low - operator: eq - hertz: 200 - """) - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 3]), - recording=recording, +def test_has_all_tags_fails_if_empty(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="at least one tag"): + build_condition_from_str( + tmp_path, + """ + name: has_all_tags + tags: [] + """, ) + + +def test_all_of_condition(tmp_path: Path, recording: data.Recording) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: all_of + conditions: + - name: has_tag + tag: + key: species + value: Myotis myotis + - name: duration + operator: lt + seconds: 1 + """, ) - assert not condition(se) - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeStamp(coordinates=3), - recording=recording, - ) - ) - assert not condition(se) - - -def test_has_tags_fails_if_empty(): - with pytest.raises(ValueError): - build_condition_from_str(""" - name: has_tags - tags: [] - """) - - -def test_all_of(recording: data.Recording): - condition = build_condition_from_str(""" - name: all_of - conditions: - - name: has_tag - tag: - key: species - value: Myotis myotis - - name: duration - operator: lt - seconds: 1 - """) - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 0.5]), - recording=recording, - ), + passing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, 0.5]), tags=[data.Tag(key="species", value="Myotis myotis")], ) - assert condition(se) - - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 2]), - recording=recording, - ), + failing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, 2]), tags=[data.Tag(key="species", value="Myotis myotis")], ) - assert not condition(se) - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 0.5]), - recording=recording, - ), + assert condition(passing) + assert not condition(failing) + + +def test_any_of_condition(tmp_path: Path, recording: data.Recording) -> None: + condition = build_condition_from_str( + tmp_path, + """ + name: any_of + conditions: + - name: has_tag + tag: + key: species + value: Myotis myotis + - name: duration + operator: lt + seconds: 1 + """, + ) + + passing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, 2]), + tags=[data.Tag(key="species", value="Myotis myotis")], + ) + failing = create_sound_event_annotation( + recording=recording, + geometry=data.TimeInterval(coordinates=[0, 2]), tags=[data.Tag(key="species", value="Eptesicus fuscus")], ) - assert not condition(se) - -def test_any_of(recording: data.Recording): - condition = build_condition_from_str(""" - name: any_of - conditions: - - name: has_tag - tag: - key: species - value: Myotis myotis - - name: duration - operator: lt - seconds: 1 - """) - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 2]), - recording=recording, - ), - tags=[data.Tag(key="species", value="Eptesicus fuscus")], - ) - assert not condition(se) - - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 0.5]), - recording=recording, - ), - tags=[data.Tag(key="species", value="Myotis myotis")], - ) - assert condition(se) - - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 2]), - recording=recording, - ), - tags=[data.Tag(key="species", value="Myotis myotis")], - ) - assert condition(se) - - se = data.SoundEventAnnotation( - sound_event=data.SoundEvent( - geometry=data.TimeInterval(coordinates=[0, 0.5]), - recording=recording, - ), - tags=[data.Tag(key="species", value="Eptesicus fuscus")], - ) - assert condition(se) + assert condition(passing) + assert not condition(failing)