Add normalization for missing strings in DICOM tags snapshot and enhance tests

Author: dmandache

src/imperandi/ingest/parse.py

@@ -550,6 +550,21 @@ def read_dicom_header_with_force(fp, force):
     return read_dicom_header(fp, force=force)
 
 
+def _normalize_snapshot_missing_strings(value):
+    if isinstance(value, str):
+        return None if value.strip() == "" else value
+    if isinstance(value, list):
+        return [_normalize_snapshot_missing_strings(v) for v in value]
+    if isinstance(value, tuple):
+        return [_normalize_snapshot_missing_strings(v) for v in value]
+    if isinstance(value, dict):
+        return {
+            key: _normalize_snapshot_missing_strings(val)
+            for key, val in value.items()
+        }
+    return value
+
+
 def build_global_readers(
     *,
     initial_archive_mode: bool,
@@ -653,7 +668,7 @@ def write_dicom_tags_snapshot(
                 "_relative_path": row.get("_relative_path"),
                 "snapshot_seed": seed,
                 "snapshot_index": int(idx),
-                "tags": tags_series.to_dict(),
+                "tags": _normalize_snapshot_missing_strings(tags_series.to_dict()),
             }
             handle.write(json.dumps(record, ensure_ascii=True) + "\n")
             written += 1

tests/test_clean.py

@@ -10,6 +10,16 @@
 
 from imperandi.ingest import clean
 
+ACQUISITION_TEMP_COLS = {
+    "_acq_timestamp",
+    "_series_number_sort",
+    "_acquisition_number_sort",
+}
+
+
+def _assert_no_acquisition_temp_cols(df: pd.DataFrame) -> None:
+    assert ACQUISITION_TEMP_COLS.isdisjoint(df.columns)
+
 
 def test_normalize_clean_args_prefers_optional_csv_path(tmp_path):
     csv_pos = tmp_path / "pos.csv"
@@ -344,6 +354,7 @@ def test_compute_visit_and_acquisition_order():
     assert set(out2["acquisition_order"].dropna()) == {0, 1, 2}
     assert (out2["delay_since_prev_acq_sec"].dropna() >= 0).all()
     assert (out2["delay_since_first_acq_sec"].dropna() >= 0).all()
+    _assert_no_acquisition_temp_cols(out2)
 
     # Handles aggregated time values represented as datetime.time objects or repr strings
     df3 = pd.DataFrame(
@@ -367,6 +378,7 @@ def test_compute_visit_and_acquisition_order():
     assert out3.set_index("volume_id").loc["v3", "acquisition_order"] == 2
     assert (out3["delay_since_prev_acq_sec"].dropna() >= 0).all()
     assert (out3["delay_since_first_acq_sec"].dropna() >= 0).all()
+    _assert_no_acquisition_temp_cols(out3)
 
     # Ensure ordering uses acquisition timestamp, not lexical volume_id order.
     df4 = pd.DataFrame(
@@ -391,6 +403,7 @@ def test_compute_visit_and_acquisition_order():
     assert out4_by_volume.loc["v1", "acquisition_order"] == 2
     assert (out4["delay_since_prev_acq_sec"].dropna() >= 0).all()
     assert (out4["delay_since_first_acq_sec"].dropna() >= 0).all()
+    _assert_no_acquisition_temp_cols(out4)
 
 
 def test_compute_acquisition_order_without_time_uses_series_and_acquisition_number():
@@ -413,6 +426,7 @@ def test_compute_acquisition_order_without_time_uses_series_and_acquisition_numb
     assert out_by_volume.loc["v2", "acquisition_order"] == 0
     assert out_by_volume.loc["v3", "acquisition_order"] == 1
     assert out_by_volume.loc["v1", "acquisition_order"] == 2
+    _assert_no_acquisition_temp_cols(out)
 
 
 def test_compute_acquisition_order_without_date_and_time_falls_back_to_numbers():
@@ -433,6 +447,7 @@ def test_compute_acquisition_order_without_date_and_time_falls_back_to_numbers()
     assert out_by_volume.loc["v2", "acquisition_order"] == 0
     assert out_by_volume.loc["v3", "acquisition_order"] == 1
     assert out_by_volume.loc["v1", "acquisition_order"] == 2
+    _assert_no_acquisition_temp_cols(out)
 
 
 def test_compute_acquisition_order_tie_breaks_by_volume_id_when_no_sort_keys():
@@ -450,6 +465,28 @@ def test_compute_acquisition_order_tie_breaks_by_volume_id_when_no_sort_keys():
     assert out_by_volume.loc["v1", "acquisition_order"] == 0
     assert out_by_volume.loc["v10", "acquisition_order"] == 1
     assert out_by_volume.loc["v2", "acquisition_order"] == 2
+    _assert_no_acquisition_temp_cols(out)
+
+
+def test_compute_acquisition_order_drops_internal_sort_columns():
+    df = pd.DataFrame(
+        {
+            "patient_key": ["p", "p"],
+            "study_id": ["s", "s"],
+            "volume_id": ["v1", "v2"],
+            "date": [pd.Timestamp("2020-01-01"), pd.Timestamp("2020-01-01")],
+            "time": [dt_time(12, 0, 0), dt_time(12, 1, 0)],
+            "SeriesNumber": [1, 1],
+            "AcquisitionNumber": [1, 2],
+        }
+    )
+
+    out = clean.compute_acquisition_order(df.copy())
+
+    _assert_no_acquisition_temp_cols(out)
+    assert "acquisition_order" in out.columns
+    assert "delay_since_prev_acq_sec" in out.columns
+    assert "delay_since_first_acq_sec" in out.columns
 
 
 def test_group_volumes_sorts_acquisition_number_numerically():

tests/test_parse.py

@@ -686,3 +686,39 @@ def test_write_dicom_tags_snapshot_samples_unique_series_when_available(tmp_path
     assert n == 4
     assert len(lines) == 4
     assert len({item["tags"]["Source"] for item in lines}) == 4
+
+
+def test_write_dicom_tags_snapshot_normalizes_missing_empty_strings(tmp_path):
+    df = pd.DataFrame(
+        {
+            "dicom_path": ["p0.dcm"],
+            "_scan_root": ["root"],
+            "_relative_path": ["rel/0.dcm"],
+            "_read_path": ["read/0.dcm"],
+        }
+    )
+
+    out = tmp_path / "snap_empty.ndjson"
+    n = parse.write_dicom_tags_snapshot(
+        df=df,
+        output_path=out,
+        sample_size=1,
+        seed=42,
+        read_full_func=lambda _: pd.Series(
+            {
+                "EmptyTag": "",
+                "WhitespaceTag": "   ",
+                "Nested": ["", "ok"],
+                "PresentTag": "value",
+            }
+        ),
+    )
+
+    assert n == 1
+    line = out.read_text(encoding="utf-8").strip()
+    record = json.loads(line)
+    tags = record["tags"]
+    assert tags["EmptyTag"] is None
+    assert tags["WhitespaceTag"] is None
+    assert tags["Nested"] == [None, "ok"]
+    assert tags["PresentTag"] == "value"