Report a two-lefts stereo collision instead of welding it silently (#11)

src/samplerdisc/extract.py

@@ -158,7 +158,11 @@ def extract_volume(
     """
     made = False
     originals_made = False
-    parsed: dict[str, _Pairable] = {}
+    #: Mono samples in the order they were walked, each with the filesystem name
+    #: the stereo heuristic pairs on. A list, not a name-keyed dict: two entries
+    #: can share a name, and a dict would drop one before the pairing ever saw
+    #: the collision -- the silent wrong join of issue #11.
+    mono: list[tuple[str, _Pairable]] = []
     #: sha256 of each WAV payload written, against the name it was written
     #: from and whether it carried a smpl chunk, so a duplicate can say which
     #: file already holds the audio and whether it holds the metadata too.
@@ -256,7 +260,7 @@ def extract_volume(
             # both -- 2 843 declare two channels, 14 are name-paired, and the
             # two sets do not intersect -- so this changes nothing today and
             # is here so it cannot start to (ADR-0026).
-            parsed[entry.name] = sample
+            mono.append((entry.name, sample))
         yield Extracted(
             volume=volume.name,
             name=entry.name,
@@ -275,20 +279,26 @@ def extract_volume(
         yield _convert_aiff(image, backend, origin, volume, entry, out_dir, written_audio)
 
     if join_stereo:
-        yield from _join_pairs(volume, parsed, out_dir)
+        yield from _join_pairs(volume, mono, out_dir)
 
 
 def _join_pairs(
-    volume: Volume, parsed: dict[str, _Pairable], out_dir: str
+    volume: Volume, mono: list[tuple[str, _Pairable]], out_dir: str
 ) -> Iterator[Skipped | Joined]:
-    pairs = find_pairs(list(parsed))
-    if not pairs:
+    result = find_pairs(mono)
+    for amb in result.ambiguous:
+        # Two lefts for one base, or three files: a pairing was possible and
+        # refused rather than guessed. The mono halves are already written, so
+        # nothing is lost by declining -- the report is what makes the choice
+        # visible instead of silent (ADR-0007, issue #11).
+        yield Skipped(volume.name, amb.base, amb.reason, volume.partition)
+    if not result.pairs:
         return
     stereo_dir = os.path.join(out_dir, "stereo")
     made = False
-    for pair in pairs:
-        left = parsed[pair.left]
-        right = parsed[pair.right]
+    for pair in result.pairs:
+        left = pair.left
+        right = pair.right
         if left.rate != right.rate:
             # Different rates means these are not two halves of one sound,
             # whatever the names say.

src/samplerdisc/stereo.py

@@ -23,6 +23,13 @@
 import sys
 from array import array
 from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+#: What each name carries alongside it -- the parsed sample, to ``extract`` --
+#: threaded through the heuristic untouched so a matched pair hands back the
+#: object itself rather than a name to look up again. Keeping it opaque is what
+#: lets this stay in brand-neutral core: the pairing knows names, nothing more.
+T = TypeVar("T")
 
 #: A trailing side letter behind a separator, optionally with padding around
 #: it. Both name forms are fixed width and space-padded, so the side marker is
@@ -36,10 +43,36 @@
 
 
 @dataclass(frozen=True)
-class Pair:
+class Pair(Generic[T]):
+    base: str
+    #: The payloads the two halves arrived carrying, not their names: a matched
+    #: pair hands back the samples themselves, so ``extract`` never looks one up
+    #: by a name that a second entry may share (issue #11).
+    left: T
+    right: T
+
+
+@dataclass(frozen=True)
+class Ambiguous:
+    """A base whose halves could not be paired without a guess.
+
+    Two files marked ``-L`` and one marked ``-R`` is not a pair. Before this was
+    reported it was silently dropped, and a duplicate name upstream could defeat
+    the drop and weld two unrelated sounds together (issue #11). Reported, it
+    becomes a ``Skipped`` -- the same way ``extract`` handles every other
+    ambiguity it will not resolve on its own.
+    """
+
     base: str
-    left: str
-    right: str
+    reason: str
+
+
+@dataclass(frozen=True)
+class Pairing(Generic[T]):
+    """The two outcomes of pairing: the clean matches, and the refusals."""
+
+    pairs: list[Pair[T]]
+    ambiguous: list[Ambiguous]
 
 
 def split_side(name: str) -> tuple[str, str] | None:
@@ -53,33 +86,51 @@ def split_side(name: str) -> tuple[str, str] | None:
     return base, match.group("side").upper()
 
 
-def find_pairs(names: list[str]) -> list[Pair]:
+def find_pairs(items: list[tuple[str, T]]) -> Pairing[T]:
     """Match -L names to -R names, preserving the order the left ones arrived in.
 
-    A base with two lefts and no right, or three files, is not a pair. Being
-    conservative costs the user a manual join; being loose silently welds two
-    unrelated sounds together.
+    Each item is ``(name, payload)``; only the name drives the match, so this
+    stays a pure name heuristic, and the payload rides along so a matched pair
+    hands back the two objects rather than names to resolve again.
+
+    A base with exactly one left and one right pairs. Both sides present in any
+    other count -- two lefts and a right, three files -- is *ambiguous* and
+    reported rather than resolved: being conservative costs the user a manual
+    join, and being loose silently welds two unrelated sounds together, which
+    without the mono originals is unrecoverable (ADR-0007). A lone half with no
+    opposite is neither -- it stays an unpaired mono sample, as it always has.
     """
-    lefts: dict[str, list[str]] = {}
-    rights: dict[str, list[str]] = {}
+    lefts: dict[str, list[T]] = {}
+    rights: dict[str, list[T]] = {}
     order: list[str] = []
-    for name in names:
+    for name, payload in items:
         parsed = split_side(name)
         if parsed is None:
             continue
         base, side = parsed
         bucket = lefts if side == "L" else rights
         if base not in lefts and base not in rights:
             order.append(base)
-        bucket.setdefault(base, []).append(name)
+        bucket.setdefault(base, []).append(payload)
 
-    pairs = []
+    pairs: list[Pair[T]] = []
+    ambiguous: list[Ambiguous] = []
     for base in order:
         left = lefts.get(base, [])
         right = rights.get(base, [])
         if len(left) == 1 and len(right) == 1:
             pairs.append(Pair(base=base, left=left[0], right=right[0]))
-    return pairs
+        elif left and right:
+            ambiguous.append(
+                Ambiguous(
+                    base=base,
+                    reason=(
+                        f"{len(left)} files marked -L and {len(right)} marked -R; "
+                        "refusing to guess the pair"
+                    ),
+                )
+            )
+    return Pairing(pairs=pairs, ambiguous=ambiguous)
 
 
 def interleave(left: bytes, right: bytes) -> bytes:

tests/test_stereo.py

@@ -16,6 +16,16 @@
 BACKEND = AkaiBackend()
 
 
+def _named(names):
+    """Pair each name with itself as payload.
+
+    ``find_pairs`` carries a payload beside every name; these name-only tests
+    pass the name as its own payload, so a matched pair reads back as the names
+    it joined.
+    """
+    return [(name, name) for name in names]
+
+
 @pytest.mark.parametrize(
     ("name", "expected"),
     [
@@ -67,50 +77,62 @@ def test_the_separator_is_not_optional(name):
 
 
 def test_roland_pairs_are_matched_by_base_name():
-    pairs = find_pairs(
-        [
-            "STR:Vn1 Pizz55\x7fL",
-            "STR:Solo Vla",
-            "STR:Vn1 Pizz55\x7fR",
-            "MOVIN 105 -L",
-            "MOVIN 105 -R",
-        ]
+    result = find_pairs(
+        _named(
+            [
+                "STR:Vn1 Pizz55\x7fL",
+                "STR:Solo Vla",
+                "STR:Vn1 Pizz55\x7fR",
+                "MOVIN 105 -L",
+                "MOVIN 105 -R",
+            ]
+        )
     )
-    assert [(p.base, p.left, p.right) for p in pairs] == [
+    assert [(p.base, p.left, p.right) for p in result.pairs] == [
         ("STR:Vn1 Pizz55", "STR:Vn1 Pizz55\x7fL", "STR:Vn1 Pizz55\x7fR"),
         ("MOVIN 105", "MOVIN 105 -L", "MOVIN 105 -R"),
     ]
 
 
 def test_an_unmatched_roland_half_is_not_a_pair():
-    assert find_pairs(["STR:Lone\x7fL"]) == []
-    assert find_pairs(["STR:Lone\x7fR"]) == []
+    assert find_pairs(_named(["STR:Lone\x7fL"])).pairs == []
+    assert find_pairs(_named(["STR:Lone\x7fR"])).pairs == []
 
 
-def test_ambiguous_roland_names_are_left_alone():
-    assert find_pairs(["A\x7fL", "A\x7fL", "A\x7fR"]) == []
+def test_ambiguous_roland_names_are_reported_not_paired():
+    result = find_pairs(_named(["A\x7fL", "A\x7fL", "A\x7fR"]))
+    assert result.pairs == []
+    assert [a.base for a in result.ambiguous] == ["A"]
 
 
 def test_pairs_are_matched_by_base_name():
-    pairs = find_pairs(["MOVIN 105 -L", "KICK", "MOVIN 105 -R", "SWG-T2-100-L", "SWG-T2-100-R"])
-    assert [(p.base, p.left, p.right) for p in pairs] == [
+    result = find_pairs(
+        _named(["MOVIN 105 -L", "KICK", "MOVIN 105 -R", "SWG-T2-100-L", "SWG-T2-100-R"])
+    )
+    assert [(p.base, p.left, p.right) for p in result.pairs] == [
         ("MOVIN 105", "MOVIN 105 -L", "MOVIN 105 -R"),
         ("SWG-T2-100", "SWG-T2-100-L", "SWG-T2-100-R"),
     ]
 
 
 def test_an_unmatched_half_is_not_a_pair():
-    assert find_pairs(["LONE -L"]) == []
-    assert find_pairs(["LONE -R"]) == []
+    """A lone half with no opposite stays mono -- neither a pair nor ambiguous."""
+    assert find_pairs(_named(["LONE -L"])) == find_pairs([])
+    assert find_pairs(_named(["LONE -R"])) == find_pairs([])
 
 
-def test_ambiguous_names_are_left_alone():
-    """Two lefts for one base is not a pair.
+def test_ambiguous_names_are_reported_not_paired():
+    """Two lefts for one base is not a pair; now it is reported, not dropped.
 
     Being conservative costs a manual join; being loose welds two unrelated
-    sounds together and, without the mono originals, unrecoverably.
+    sounds together and, without the mono originals, unrecoverably. Reporting
+    the refusal is what keeps a duplicate name upstream from defeating it
+    silently (issue #11).
     """
-    assert find_pairs(["A -L", "A -L", "A -R"]) == []
+    result = find_pairs(_named(["A -L", "A -L", "A -R"]))
+    assert result.pairs == []
+    assert [a.base for a in result.ambiguous] == ["A"]
+    assert "2 files marked -L and 1 marked -R" in result.ambiguous[0].reason
 
 
 # --- interleaving -------------------------------------------------------
@@ -194,3 +216,45 @@ def test_stereo_channels_match_their_mono_sources(tmp_path):
         both = w.readframes(w.getnframes())
     assert both[0::4] == left[0::2] and both[1::4] == left[1::2]
     assert both[2::4] == right[0::2] and both[3::4] == right[1::2]
+
+
+def test_two_lefts_under_one_name_are_reported_not_welded(tmp_path):
+    """Two distinct files named ``KICK -L`` and one ``KICK -R`` must not weld.
+
+    A name is not unique on these filesystems, and a name-keyed accumulator
+    would drop one left before the pairing ever saw the collision -- then join
+    whichever survived to the right and report an ordinary stereo file, the
+    silent wrong join ADR-0007 keeps the mono originals against (issue #11).
+    Nothing is lost by refusing: both lefts and the right are already on disk.
+    """
+    first = fixtures.akai_sample("KICK -L", words=64)
+    second = fixtures.akai_sample("KICK -L", words=128)
+    right = fixtures.akai_sample("KICK -R", words=100)
+    path = tmp_path / "dup.iso"
+    path.write_bytes(
+        fixtures.akai_partition(
+            [
+                (
+                    "VOL 1",
+                    [
+                        ("KICK -L", 0x73, len(first), first),
+                        ("KICK -L", 0x73, len(second), second),
+                        ("KICK -R", 0x73, len(right), right),
+                    ],
+                )
+            ]
+        )
+    )
+    out = tmp_path / "out"
+    results = list(extract_disc(FlatImage(path), BACKEND, 0, str(out)))
+
+    assert not any(isinstance(r, Joined) for r in results)
+    ambiguous = [r for r in results if isinstance(r, Skipped) and r.name == "KICK"]
+    assert len(ambiguous) == 1
+    assert "-L" in ambiguous[0].reason and "-R" in ambiguous[0].reason
+
+    volume = out / "partition-1" / "VOL 1"
+    assert (volume / "KICK -L.wav").exists()
+    assert (volume / "KICK -L_2.wav").exists()  # unique_path kept the second
+    assert (volume / "KICK -R.wav").exists()
+    assert not (volume / "stereo").exists()