Improve test suite for conditions

This commit is contained in:
mbsantiago 2026-04-03 17:07:26 +01:00
parent e04d86808d
commit 00961132a9
3 changed files with 757 additions and 537 deletions

View File

@ -1,21 +1,38 @@
import json import json
import textwrap
import uuid
from pathlib import Path from pathlib import Path
from pydantic import TypeAdapter
from soundevent import data from soundevent import data
from batdetect2.core import load_config
from batdetect2.data.conditions import ( from batdetect2.data.conditions import (
ClipAllOfConfig, ClipAnnotationConditionConfig,
ClipAnyOfConfig,
ClipNotConfig,
HasAllTagsConfig,
HasAnyTagConfig,
HasTagConfig,
IdInListConfig,
RecordingSatisfiesConfig,
build_clip_annotation_condition, 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( def test_recording_satisfies_condition(
tmp_path: Path, tmp_path: Path,
create_recording, create_recording,
@ -31,10 +48,14 @@ def test_recording_satisfies_condition(
ids_path = tmp_path / "recording_ids.json" ids_path = tmp_path / "recording_ids.json"
ids_path.write_text(json.dumps([str(recording_a.uuid)])) ids_path.write_text(json.dumps([str(recording_a.uuid)]))
condition = build_clip_annotation_condition( condition = build_clip_condition_from_yaml(
RecordingSatisfiesConfig( tmp_path,
condition=IdInListConfig(path=ids_path), f"""
) name: recording_satisfies
condition:
name: id_in_list
path: {ids_path}
""",
) )
assert condition(clip_annotation_a) 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 = tmp_path / "clip_annotation_ids.json"
ids_path.write_text(json.dumps([str(clip_annotation_a.uuid)])) 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 condition(clip_annotation_a)
assert not condition(clip_annotation_b) assert not condition(clip_annotation_b)
@ -68,7 +95,6 @@ def test_clip_has_tag_conditions(
) -> None: ) -> None:
reviewed = data.Tag(key="status", value="reviewed") reviewed = data.Tag(key="status", value="reviewed")
train = data.Tag(key="split", value="train") train = data.Tag(key="split", value="train")
val = data.Tag(key="split", value="val")
recording = create_recording(path=tmp_path / "rec.wav") recording = create_recording(path=tmp_path / "rec.wav")
clip = create_clip(recording) clip = create_clip(recording)
@ -76,21 +102,96 @@ def test_clip_has_tag_conditions(
clip, clip,
clip_tags=[reviewed, train], clip_tags=[reviewed, train],
) )
clip_annotation_missing = create_clip_annotation(
has_tag = build_clip_annotation_condition(HasTagConfig(tag=reviewed)) create_clip(recording),
has_all = build_clip_annotation_condition( clip_tags=[train],
HasAllTagsConfig(tags=[reviewed, train])
)
has_any = build_clip_annotation_condition(
HasAnyTagConfig(tags=[val, train])
) )
assert has_tag(clip_annotation) condition = build_clip_condition_from_yaml(
assert has_all(clip_annotation) tmp_path,
assert has_any(clip_annotation) """
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, tmp_path: Path,
create_recording, create_recording,
create_clip, create_clip,
@ -98,7 +199,6 @@ def test_clip_logical_conditions(
) -> None: ) -> None:
reviewed = data.Tag(key="status", value="reviewed") reviewed = data.Tag(key="status", value="reviewed")
train = data.Tag(key="split", value="train") train = data.Tag(key="split", value="train")
val = data.Tag(key="split", value="val")
recording = create_recording(path=tmp_path / "rec.wav") recording = create_recording(path=tmp_path / "rec.wav")
clip = create_clip(recording) clip = create_clip(recording)
@ -106,27 +206,98 @@ def test_clip_logical_conditions(
clip, clip,
clip_tags=[reviewed, train], clip_tags=[reviewed, train],
) )
clip_annotation_missing = create_clip_annotation(
all_condition = build_clip_annotation_condition( create_clip(recording),
ClipAllOfConfig( clip_tags=[reviewed],
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))
) )
assert all_condition(clip_annotation) condition = build_clip_condition_from_yaml(
assert any_condition(clip_annotation) tmp_path,
assert not_condition(clip_annotation) """
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)

View File

@ -1,28 +1,53 @@
import json import json
import textwrap
import uuid
from pathlib import Path from pathlib import Path
import pytest import pytest
from pydantic import TypeAdapter
from soundevent import data from soundevent import data
from batdetect2.core import load_config
from batdetect2.data.conditions import ( from batdetect2.data.conditions import (
HasAllTagsConfig, RecordingConditionConfig,
HasAnyTagConfig,
HasTagConfig,
IdInListConfig,
RecordingAllOfConfig,
RecordingAnyOfConfig,
RecordingNotConfig,
build_recording_condition, 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: def test_id_in_list_condition(tmp_path: Path, create_recording) -> None:
recording_a = 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") recording_b = create_recording(path=tmp_path / "b.wav")
ids_path = tmp_path / "recording_ids.json" ids_path = tmp_path / "recording_ids.json"
ids_path.write_text(json.dumps([str(recording_a.uuid)])) 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 condition(recording_a)
assert not condition(recording_b) assert not condition(recording_b)
@ -32,18 +57,24 @@ def test_id_in_list_condition_uses_base_dir(
tmp_path: Path, tmp_path: Path,
create_recording, create_recording,
) -> None: ) -> 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 = tmp_path / "splits"
split_dir.mkdir() split_dir.mkdir()
ids_path = split_dir / "train_ids.json" 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( condition = build_recording_condition_from_yaml(
IdInListConfig(path=Path("splits/train_ids.json")), tmp_path,
"""
name: id_in_list
path: splits/train_ids.json
""",
base_dir=tmp_path, 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( 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"})) ids_path.write_text(json.dumps({"id": "foo"}))
with pytest.raises(TypeError, match="Expected JSON list"): 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: 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"])) ids_path.write_text(json.dumps(["not-a-uuid"]))
with pytest.raises(ValueError, match="Invalid ID"): 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( 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 = tmp_path / "recording_ids.txt"
ids_path.write_text(f"{recording_a.uuid}\n") ids_path.write_text(f"{recording_a.uuid}\n")
condition = build_recording_condition( condition = build_recording_condition_from_yaml(
IdInListConfig(path=ids_path, list_format="txt") tmp_path,
f"""
name: id_in_list
path: {ids_path}
list_format: txt
""",
) )
assert condition(recording_a) assert condition(recording_a)
assert not condition(recording_b) assert not condition(recording_b)
def test_recording_has_tag_conditions( def test_has_tag_condition(tmp_path: Path, create_recording) -> None:
tmp_path: Path, create_recording train = data.Tag(key="split", value="train")
) -> None: 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") train = data.Tag(key="split", value="train")
uk = data.Tag(key="region", value="uk") uk = data.Tag(key="region", value="uk")
eu = data.Tag(key="region", value="eu")
recording = create_recording( recording_a = create_recording(
path=tmp_path / "rec.wav", path=tmp_path / "a.wav",
tags=[train, uk], tags=[train, uk],
) )
recording_b = create_recording(
path=tmp_path / "b.wav",
tags=[train],
)
has_train = build_recording_condition(HasTagConfig(tag=train)) condition = build_recording_condition_from_yaml(
has_all = build_recording_condition(HasAllTagsConfig(tags=[train, uk])) tmp_path,
has_any = build_recording_condition(HasAnyTagConfig(tags=[eu, uk])) """
name: has_all_tags
tags:
- key: split
value: train
- key: region
value: uk
""",
)
assert has_train(recording) assert condition(recording_a)
assert has_all(recording) assert not condition(recording_b)
assert has_any(recording)
def test_recording_logical_conditions( def test_has_any_tag_condition(tmp_path: Path, create_recording) -> None:
tmp_path: Path, create_recording uk = data.Tag(key="region", value="uk")
) -> None: 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") train = data.Tag(key="split", value="train")
uk = data.Tag(key="region", value="uk") uk = data.Tag(key="region", value="uk")
eu = data.Tag(key="region", value="eu") us = data.Tag(key="region", value="us")
recording = create_recording( recording_a = create_recording(
path=tmp_path / "rec.wav", path=tmp_path / "a.wav",
tags=[train, uk], tags=[train, uk],
) )
recording_b = create_recording(
all_condition = build_recording_condition( path=tmp_path / "b.wav",
RecordingAllOfConfig( tags=[train, us],
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))
) )
assert all_condition(recording) condition = build_recording_condition_from_yaml(
assert any_condition(recording) tmp_path,
assert not_condition(recording) """
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)

View File

@ -1,442 +1,347 @@
import json import json
import textwrap import textwrap
import uuid
from pathlib import Path from pathlib import Path
import pytest import pytest
import yaml
from pydantic import TypeAdapter from pydantic import TypeAdapter
from soundevent import data from soundevent import data
from batdetect2.core import load_config
from batdetect2.data.conditions import ( from batdetect2.data.conditions import (
IdInListConfig,
SoundEventConditionConfig, SoundEventConditionConfig,
build_sound_event_condition, build_sound_event_condition,
) )
def build_condition_from_str(content, base_dir: Path | None = None): def load_sound_event_condition_config(
content = textwrap.dedent(content) tmp_path: Path,
content = yaml.safe_load(content) yaml_string: str,
config = TypeAdapter(SoundEventConditionConfig).validate_python(content) ) -> 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) return build_sound_event_condition(config, base_dir=base_dir)
def test_has_tag(sound_event: data.SoundEvent): def create_sound_event_annotation(
condition = build_condition_from_str(""" 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 [],
)
def test_has_tag_condition(
sound_event: data.SoundEvent, tmp_path: Path
) -> None:
condition = build_condition_from_str(
tmp_path,
"""
name: has_tag name: has_tag
tag: tag:
key: species key: species
value: Myotis myotis value: Myotis myotis
""") """,
)
sound_event_annotation = data.SoundEventAnnotation( passing = data.SoundEventAnnotation(
sound_event=sound_event, sound_event=sound_event,
tags=[data.Tag(key="species", value="Myotis myotis")], tags=[data.Tag(key="species", value="Myotis myotis")],
) )
assert condition(sound_event_annotation) failing = data.SoundEventAnnotation(
sound_event_annotation = data.SoundEventAnnotation(
sound_event=sound_event, sound_event=sound_event,
tags=[data.Tag(key="species", value="Eptesicus fuscus")], 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): def test_has_all_tags_condition(
condition = build_condition_from_str(""" sound_event: data.SoundEvent,
tmp_path: Path,
) -> None:
condition = build_condition_from_str(
tmp_path,
"""
name: has_all_tags name: has_all_tags
tags: tags:
- key: species - key: species
value: Myotis myotis value: Myotis myotis
- key: event - key: event
value: Echolocation 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, sound_event=sound_event,
tags=[data.Tag(key="species", value="Myotis myotis")], tags=[data.Tag(key="species", value="Myotis myotis")],
) )
assert not condition(sound_event_annotation)
sound_event_annotation = data.SoundEventAnnotation( assert condition(passing)
sound_event=sound_event, assert not condition(failing)
tags=[
data.Tag(key="species", value="Eptesicus fuscus"),
data.Tag(key="event", value="Echolocation"),
],
)
assert not 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(
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): def test_has_any_tag_condition(
condition = build_condition_from_str(""" sound_event: data.SoundEvent,
tmp_path: Path,
) -> None:
condition = build_condition_from_str(
tmp_path,
"""
name: has_any_tag name: has_any_tag
tags: tags:
- key: species - key: species
value: Myotis myotis value: Myotis myotis
- key: event - key: event
value: Echolocation 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( passing = data.SoundEventAnnotation(
sound_event=sound_event, sound_event=sound_event,
tags=[ tags=[data.Tag(key="event", value="Echolocation")],
data.Tag(key="species", value="Eptesicus fuscus"),
data.Tag(key="event", value="Echolocation"),
],
) )
assert condition(sound_event_annotation) failing = data.SoundEventAnnotation(
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(
sound_event=sound_event, sound_event=sound_event,
tags=[ tags=[
data.Tag(key="species", value="Eptesicus fuscus"), data.Tag(key="species", value="Eptesicus fuscus"),
data.Tag(key="event", value="Social"), 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): def test_not_condition(sound_event: data.SoundEvent, tmp_path: Path) -> None:
condition = build_condition_from_str(""" condition = build_condition_from_str(
tmp_path,
"""
name: not name: not
condition: condition:
name: has_tag name: has_tag
tag: tag:
key: species key: species
value: Myotis myotis value: Myotis myotis
""") """,
sound_event_annotation = 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( passing = data.SoundEventAnnotation(
sound_event=sound_event, sound_event=sound_event,
tags=[data.Tag(key="species", value="Eptesicus fuscus")], tags=[data.Tag(key="species", value="Eptesicus fuscus")],
) )
assert condition(sound_event_annotation) failing = data.SoundEventAnnotation(
sound_event_annotation = data.SoundEventAnnotation(
sound_event=sound_event, sound_event=sound_event,
tags=[ tags=[data.Tag(key="species", value="Myotis myotis")],
data.Tag(key="species", value="Myotis myotis"),
data.Tag(key="event", value="Echolocation"),
],
) )
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): def test_id_in_list_condition(
se1 = data.SoundEventAnnotation(sound_event=sound_event) sound_event: data.SoundEvent, tmp_path: Path
se2 = data.SoundEventAnnotation(sound_event=sound_event) ) -> 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 = 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 condition(passing)
assert not condition(se2) 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, sound_event: data.SoundEvent,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> 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 = tmp_path / "splits"
split_dir.mkdir() split_dir.mkdir()
ids_path = split_dir / "sound_event_ids.json" 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( condition = build_condition_from_str(
IdInListConfig(path=Path("splits/sound_event_ids.json")), tmp_path,
"""
name: id_in_list
path: splits/sound_event_ids.json
""",
base_dir=tmp_path, base_dir=tmp_path,
) )
assert condition(se) assert condition(passing)
assert not condition(failing)
def test_duration(recording: data.Recording): @pytest.mark.parametrize(
se1 = data.SoundEventAnnotation( "operator,seconds,passing_duration,failing_duration",
sound_event=data.SoundEvent( [
recording=recording, geometry=data.TimeInterval(coordinates=[0, 1]) ("lt", 2, 1, 2),
), ("lte", 2, 2, 3),
) ("gt", 2, 3, 2),
se2 = data.SoundEventAnnotation( ("gte", 2, 2, 1),
sound_event=data.SoundEvent( ("eq", 2, 2, 3),
recording=recording, geometry=data.TimeInterval(coordinates=[0, 2]) ],
), )
) def test_duration_condition(
se3 = data.SoundEventAnnotation( tmp_path: Path,
sound_event=data.SoundEvent( recording: data.Recording,
recording=recording, geometry=data.TimeInterval(coordinates=[0, 3]) operator: str,
), seconds: int,
) passing_duration: int,
failing_duration: int,
condition = build_condition_from_str(""" ) -> None:
condition = build_condition_from_str(
tmp_path,
f"""
name: duration name: duration
operator: lt operator: {operator}
seconds: 2 seconds: {seconds}
""") """,
assert condition(se1) )
assert not condition(se2)
assert not condition(se3)
condition = build_condition_from_str(""" passing = create_sound_event_annotation(
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)
def test_frequency(recording: data.Recording):
se12 = data.SoundEventAnnotation(
sound_event=data.SoundEvent(
recording=recording, recording=recording,
geometry=data.BoundingBox(coordinates=[0, 100, 1, 200]), 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]
), ),
) )
se13 = data.SoundEventAnnotation( failing = create_sound_event_annotation(
sound_event=data.SoundEvent(
recording=recording, recording=recording,
geometry=data.BoundingBox(coordinates=[0, 100, 2, 300]), geometry=data.BoundingBox(
), coordinates=[float(value) for value in failing_bbox]
)
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]),
), ),
) )
condition = build_condition_from_str(""" assert condition(passing)
name: frequency assert not condition(failing)
boundary: high
operator: lt
hertz: 300
""")
assert condition(se12)
assert not condition(se13)
assert not condition(se14)
condition = build_condition_from_str("""
name: frequency
boundary: high
operator: lte
hertz: 300
""")
assert condition(se12) def test_frequency_condition_is_false_for_temporal_geometries(
assert condition(se13) tmp_path: Path,
assert not condition(se14) recording: data.Recording,
) -> None:
condition = build_condition_from_str(""" condition = build_condition_from_str(
name: frequency tmp_path,
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 name: frequency
boundary: low boundary: low
operator: eq operator: eq
hertz: 200 hertz: 200
""") """,
)
assert not condition(se14) passing = create_sound_event_annotation(
assert condition(se24) recording=recording,
assert not condition(se34) geometry=data.BoundingBox(coordinates=[0, 200, 1, 400]),
)
failing = create_sound_event_annotation(
def test_frequency_is_false_for_temporal_geometries(recording: data.Recording): recording=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]), geometry=data.TimeInterval(coordinates=[0, 3]),
recording=recording,
) )
)
assert not condition(se)
se = data.SoundEventAnnotation( assert condition(passing)
sound_event=data.SoundEvent( assert not condition(failing)
geometry=data.TimeStamp(coordinates=3),
recording=recording,
)
)
assert not condition(se)
def test_has_tags_fails_if_empty(): def test_has_all_tags_fails_if_empty(tmp_path: Path) -> None:
with pytest.raises(ValueError): with pytest.raises(ValueError, match="at least one tag"):
build_condition_from_str(""" build_condition_from_str(
name: has_tags tmp_path,
"""
name: has_all_tags
tags: [] tags: []
""") """,
)
def test_all_of(recording: data.Recording): def test_all_of_condition(tmp_path: Path, recording: data.Recording) -> None:
condition = build_condition_from_str(""" condition = build_condition_from_str(
tmp_path,
"""
name: all_of name: all_of
conditions: conditions:
- name: has_tag - name: has_tag
@ -446,37 +351,28 @@ def test_all_of(recording: data.Recording):
- name: duration - name: duration
operator: lt operator: lt
seconds: 1 seconds: 1
""") """,
se = data.SoundEventAnnotation( )
sound_event=data.SoundEvent(
geometry=data.TimeInterval(coordinates=[0, 0.5]), passing = create_sound_event_annotation(
recording=recording, recording=recording,
), geometry=data.TimeInterval(coordinates=[0, 0.5]),
tags=[data.Tag(key="species", value="Myotis myotis")], tags=[data.Tag(key="species", value="Myotis myotis")],
) )
assert condition(se) failing = create_sound_event_annotation(
recording=recording,
se = data.SoundEventAnnotation(
sound_event=data.SoundEvent(
geometry=data.TimeInterval(coordinates=[0, 2]), geometry=data.TimeInterval(coordinates=[0, 2]),
recording=recording,
),
tags=[data.Tag(key="species", value="Myotis myotis")], tags=[data.Tag(key="species", value="Myotis myotis")],
) )
assert not condition(se)
se = data.SoundEventAnnotation( assert condition(passing)
sound_event=data.SoundEvent( assert not condition(failing)
geometry=data.TimeInterval(coordinates=[0, 0.5]),
recording=recording,
),
tags=[data.Tag(key="species", value="Eptesicus fuscus")],
)
assert not condition(se)
def test_any_of(recording: data.Recording): def test_any_of_condition(tmp_path: Path, recording: data.Recording) -> None:
condition = build_condition_from_str(""" condition = build_condition_from_str(
tmp_path,
"""
name: any_of name: any_of
conditions: conditions:
- name: has_tag - name: has_tag
@ -486,39 +382,19 @@ def test_any_of(recording: data.Recording):
- name: duration - name: duration
operator: lt operator: lt
seconds: 1 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( passing = create_sound_event_annotation(
sound_event=data.SoundEvent(
geometry=data.TimeInterval(coordinates=[0, 0.5]),
recording=recording, recording=recording,
), geometry=data.TimeInterval(coordinates=[0, 2]),
tags=[data.Tag(key="species", value="Myotis myotis")], tags=[data.Tag(key="species", value="Myotis myotis")],
) )
assert condition(se) failing = create_sound_event_annotation(
recording=recording,
se = data.SoundEventAnnotation(
sound_event=data.SoundEvent(
geometry=data.TimeInterval(coordinates=[0, 2]), 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")], tags=[data.Tag(key="species", value="Eptesicus fuscus")],
) )
assert condition(se)
assert condition(passing)
assert not condition(failing)