diff --git a/src/ria_toolkit_oss/annotations/annotation_transforms.py b/src/ria_toolkit_oss/annotations/annotation_transforms.py index 822a75a..bb165b6 100644 --- a/src/ria_toolkit_oss/annotations/annotation_transforms.py +++ b/src/ria_toolkit_oss/annotations/annotation_transforms.py @@ -45,7 +45,14 @@ def is_annotation_contained(inner: Annotation, outer: Annotation) -> bool: outer_sample_stop = outer.sample_start + outer.sample_count if inner.sample_start > outer.sample_start and inner_sample_stop < outer_sample_stop: - if inner.freq_lower_edge > outer.freq_lower_edge and inner.freq_upper_edge < outer.freq_upper_edge: + if ( + inner.freq_lower_edge is not None + and inner.freq_upper_edge is not None + and outer.freq_lower_edge is not None + and outer.freq_upper_edge is not None + and inner.freq_lower_edge > outer.freq_lower_edge + and inner.freq_upper_edge < outer.freq_upper_edge + ): return True return False diff --git a/src/ria_toolkit_oss/data/annotation.py b/src/ria_toolkit_oss/data/annotation.py index d565e17..fe55f97 100644 --- a/src/ria_toolkit_oss/data/annotation.py +++ b/src/ria_toolkit_oss/data/annotation.py @@ -17,10 +17,10 @@ class Annotation: :type sample_start: int :param sample_count: The index of the ending sample of the annotation, inclusive. :type sample_count: int - :param freq_lower_edge: The lower frequency of the annotation. - :type freq_lower_edge: float - :param freq_upper_edge: The upper frequency of the annotation. - :type freq_upper_edge: float + :param freq_lower_edge: The lower frequency of the annotation. Optional; None if not specified in source. + :type freq_lower_edge: float, optional + :param freq_upper_edge: The upper frequency of the annotation. Optional; None if not specified in source. + :type freq_upper_edge: float, optional :param label: The label that will be displayed with the bounding box in compatible viewers including IQEngine. Defaults to an emtpy string. :type label: str, optional @@ -34,8 +34,8 @@ class Annotation: self, sample_start: int, sample_count: int, - freq_lower_edge: float, - freq_upper_edge: float, + freq_lower_edge: Optional[float] = None, + freq_upper_edge: Optional[float] = None, label: Optional[str] = "", comment: Optional[str] = "", detail: Optional[dict] = None, @@ -43,8 +43,8 @@ class Annotation: """Initialize a new Annotation instance.""" self.sample_start = int(sample_start) self.sample_count = int(sample_count) - self.freq_lower_edge = float(freq_lower_edge) - self.freq_upper_edge = float(freq_upper_edge) + self.freq_lower_edge = float(freq_lower_edge) if freq_lower_edge is not None else None + self.freq_upper_edge = float(freq_upper_edge) if freq_upper_edge is not None else None self.label = str(label) self.comment = str(comment) @@ -62,6 +62,8 @@ class Annotation: :returns: True if valid, False if not. """ + if self.freq_lower_edge is None or self.freq_upper_edge is None: + return self.sample_count > 0 return self.sample_count > 0 and self.freq_lower_edge < self.freq_upper_edge def overlap(self, other): @@ -73,6 +75,14 @@ class Annotation: :returns: The area of the overlap in samples*frequency, or 0 if they do not overlap.""" + if ( + self.freq_lower_edge is None + or self.freq_upper_edge is None + or other.freq_lower_edge is None + or other.freq_upper_edge is None + ): + return 0 + sample_overlap_start = max(self.sample_start, other.sample_start) sample_overlap_end = min(self.sample_start + self.sample_count, other.sample_start + other.sample_count) @@ -91,6 +101,8 @@ class Annotation: :returns: sample length multiplied by bandwidth.""" + if self.freq_lower_edge is None or self.freq_upper_edge is None: + return 0 return self.sample_count * (self.freq_upper_edge - self.freq_lower_edge) def __eq__(self, other: Annotation) -> bool: @@ -103,13 +115,16 @@ class Annotation: annotation_dict = {SigMFFile.START_INDEX_KEY: self.sample_start, SigMFFile.LENGTH_INDEX_KEY: self.sample_count} - annotation_dict["metadata"] = { + metadata = { SigMFFile.LABEL_KEY: self.label, SigMFFile.COMMENT_KEY: self.comment, - SigMFFile.FHI_KEY: self.freq_upper_edge, - SigMFFile.FLO_KEY: self.freq_lower_edge, "ria:detail": self.detail, } + if self.freq_upper_edge is not None: + metadata[SigMFFile.FHI_KEY] = self.freq_upper_edge + if self.freq_lower_edge is not None: + metadata[SigMFFile.FLO_KEY] = self.freq_lower_edge + annotation_dict["metadata"] = metadata if _is_jsonable(annotation_dict): return annotation_dict diff --git a/src/ria_toolkit_oss/view/view_signal.py b/src/ria_toolkit_oss/view/view_signal.py index 628bb0e..875df85 100644 --- a/src/ria_toolkit_oss/view/view_signal.py +++ b/src/ria_toolkit_oss/view/view_signal.py @@ -81,6 +81,8 @@ def view_annotations( return 0 for annotation in sorted(annotations, key=_threshold_sort_key, reverse=True): + if annotation.freq_lower_edge is None or annotation.freq_upper_edge is None: + continue t_start = annotation.sample_start / sample_rate t_width = annotation.sample_count / sample_rate f_start = annotation.freq_lower_edge diff --git a/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py b/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py index 6477d40..b4c8a95 100644 --- a/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py +++ b/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py @@ -86,9 +86,7 @@ def save_recording_auto(recording, output_path, input_path, quiet=False, overwri input_path = Path(input_path) fmt = detect_input_format(input_path) - output_path = determine_output_path( - input_path=input_path, output_path=output_path, fmt=fmt, overwrite=overwrite - ) + output_path = determine_output_path(input_path=input_path, output_path=output_path, fmt=fmt, overwrite=overwrite) if not quiet: if fmt == "sigmf": @@ -258,7 +256,11 @@ def list(input, verbose): user_comment = ann.comment or "" # Basic info - freq_range = f"{format_frequency(ann.freq_lower_edge)} - {format_frequency(ann.freq_upper_edge)}" + freq_range = ( + f"{format_frequency(ann.freq_lower_edge)} - {format_frequency(ann.freq_upper_edge)}" + if ann.freq_lower_edge is not None and ann.freq_upper_edge is not None + else "N/A" + ) click.echo( f" [{i}] Samples {format_sample_count(ann.sample_start)}-" f"{format_sample_count(ann.sample_start + ann.sample_count)}: {ann.label}" @@ -502,8 +504,7 @@ def clear(input, output, overwrite, force, quiet): help="Annotation type", ) @click.option( - "--sample-rate", type=float, default=None, - help="Sample rate in Hz (overrides metadata; required if not in file)" + "--sample-rate", type=float, default=None, help="Sample rate in Hz (overrides metadata; required if not in file)" ) @click.option("--output", "-o", type=click.Path(), help="Output file path") @click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") @@ -617,8 +618,7 @@ def energy( help="Annotation type", ) @click.option( - "--sample-rate", type=float, default=None, - help="Sample rate in Hz (overrides metadata; required if not in file)" + "--sample-rate", type=float, default=None, help="Sample rate in Hz (overrides metadata; required if not in file)" ) @click.option("--output", "-o", type=click.Path(), help="Output file path") @click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") @@ -707,8 +707,7 @@ def cusum(input, label, min_duration, window_size, tolerance, annotation_type, s ) @click.option("--channel", type=int, default=0, help="Channel index to annotate (default: 0)") @click.option( - "--sample-rate", type=float, default=None, - help="Sample rate in Hz (overrides metadata; required if not in file)" + "--sample-rate", type=float, default=None, help="Sample rate in Hz (overrides metadata; required if not in file)" ) @click.option("--output", "-o", type=click.Path(), help="Output file path") @click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") @@ -787,8 +786,7 @@ def threshold(input, threshold, label, window_size, annotation_type, channel, sa @click.option("--noise-threshold-db", type=float, help="Noise floor threshold in dB (auto-estimated if not specified)") @click.option("--min-component-bw", type=float, default=50e3, help="Min component bandwidth in Hz") @click.option( - "--sample-rate", type=float, default=None, - help="Sample rate in Hz (overrides metadata; required if not in file)" + "--sample-rate", type=float, default=None, help="Sample rate in Hz (overrides metadata; required if not in file)" ) @click.option("--output", "-o", type=click.Path(), help="Output file path") @click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") @@ -809,7 +807,8 @@ def _log_separate_start(quiet, recording, indices_list, nfft, noise_threshold_db def separate( - input, indices, nfft, noise_threshold_db, min_component_bw, sample_rate, output, overwrite, quiet, verbose): + input, indices, nfft, noise_threshold_db, min_component_bw, sample_rate, output, overwrite, quiet, verbose +): """ Auto-detect parallel frequency-offset signals and split into sub-bands. @@ -883,7 +882,11 @@ def separate( click.echo("\n Details:") for i in range(initial_count, final_count): ann = recording.annotations[i] - freq_range = f"{format_frequency(ann.freq_lower_edge)} - {format_frequency(ann.freq_upper_edge)}" + freq_range = ( + f"{format_frequency(ann.freq_lower_edge)} - {format_frequency(ann.freq_upper_edge)}" + if ann.freq_lower_edge is not None and ann.freq_upper_edge is not None + else "N/A" + ) click.echo( f" [{i}] samples {format_sample_count(ann.sample_start)}-" f"{format_sample_count(ann.sample_start + ann.sample_count)}: {freq_range}" diff --git a/tests/datatypes/test_annotation.py b/tests/datatypes/test_annotation.py index 5dfdeb5..033ef6a 100644 --- a/tests/datatypes/test_annotation.py +++ b/tests/datatypes/test_annotation.py @@ -199,3 +199,44 @@ def test_annotation_to_sigmf_format_values(): values = list(result.values()) assert 50 in values or ann.sample_start in values assert 100 in values or ann.sample_count in values + + +# --------------------------------------------------------------------------- +# None freq-edge regression tests (SigMF optional fields) +# --------------------------------------------------------------------------- + + +def test_annotation_no_freq_edges(): + ann = Annotation(sample_start=0, sample_count=10, label="burst") + assert ann.freq_lower_edge is None + assert ann.freq_upper_edge is None + + +def test_annotation_is_valid_no_freq_edges(): + ann = Annotation(sample_start=0, sample_count=10, label="burst") + assert ann.is_valid() is True + + ann_zero = Annotation(sample_start=0, sample_count=0, label="burst") + assert ann_zero.is_valid() is False + + +def test_annotation_overlap_none_edges_returns_zero(): + ann1 = Annotation(sample_start=0, sample_count=10) + ann2 = Annotation(sample_start=0, sample_count=10, freq_lower_edge=0, freq_upper_edge=100) + assert ann1.overlap(ann2) == 0 + assert ann2.overlap(ann1) == 0 + + +def test_annotation_area_none_edges_returns_zero(): + ann = Annotation(sample_start=0, sample_count=10, label="burst") + assert ann.area() == 0 + + +def test_annotation_to_sigmf_omits_freq_keys_when_none(): + from sigmf import SigMFFile + + ann = Annotation(sample_start=0, sample_count=10, label="burst") + result = ann.to_sigmf_format() + metadata = result["metadata"] + assert SigMFFile.FLO_KEY not in metadata + assert SigMFFile.FHI_KEY not in metadata diff --git a/tests/io/test_recording_io.py b/tests/io/test_recording_io.py index 16383c2..81ff1cc 100644 --- a/tests/io/test_recording_io.py +++ b/tests/io/test_recording_io.py @@ -189,3 +189,21 @@ def test_sigmf_3(tmp_path): to_sigmf(recording=recording1, path=tmp_path, filename=filename.name) except IOError as e: assert str(e) == "File already exists" + + +def test_sigmf_annotation_without_freq_edges(tmp_path): + # Regression: annotations that omit the optional SigMF freq edge fields must + # load without error; edges should be None and the annotation still valid. + ann = Annotation(sample_start=0, sample_count=5, label="burst") + recording1 = Recording(data=complex_data_1, metadata=sample_metadata, annotations=[ann]) + + filename = tmp_path / "test" + to_sigmf(recording=recording1, path=tmp_path, filename=filename.name, overwrite=True) + recording2 = from_sigmf(filename) + + assert len(recording2.annotations) == 1 + loaded = recording2.annotations[0] + assert loaded.freq_lower_edge is None + assert loaded.freq_upper_edge is None + assert loaded.is_valid() is True + assert loaded.label == "burst"