Skip to content

Postprocessing API

Relating the experiment's recordings to each other, once the pipeline has run over each of them on its own.

Speaker Attribution

body_eye_sync.postprocessing.attribution

Who was speaking, decided by whose glasses video microphone heard them loudest.

AttributionCancelled

Bases: Exception

Raised when a caller's progress callback asks for the pass to stop.

Source code in src/body_eye_sync/postprocessing/attribution.py 
36
37
class AttributionCancelled(Exception):
    """Raised when a caller's ``progress`` callback asks for the pass to stop."""

Levels dataclass

How loud every recording is compared to its own quiet baseline.

Source code in src/body_eye_sync/postprocessing/attribution.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@dataclass
class Levels:
    """How loud every recording is compared to its own quiet baseline."""

    ids: list[str]
    times: np.ndarray
    above_floor: np.ndarray
    live_above_floor_db: float

    @cached_property
    def live(self) -> np.ndarray:
        """Which recordings are carrying speech at each moment."""
        return self.above_floor > self.live_above_floor_db

    @cached_property
    def loudest(self) -> np.ndarray:
        """The row of the loudest live recording at each moment, or ``-1`` if unclear."""
        if not self.ids or self.times.size == 0:
            return np.empty(0, dtype=int)
        return np.where(self.live.any(axis=0), self.above_floor.argmax(axis=0), -1)

    def _window(self, start: float, end: float) -> slice:
        """The columns covering ``[start, end)``, on the evenly spaced time grid."""
        stop = max(end, start + _HOP_SECONDS)
        return slice(
            int(np.searchsorted(self.times, start, side="left")),
            int(np.searchsorted(self.times, stop, side="left")),
        )

    def share(
        self,
        name: str,
        start: float,
        end: float,
    ) -> float:
        """How much of ``[start, end)`` one recording is the loudest for, ignoring silent stretches."""
        if name not in self.ids:
            return 0.0
        winners = self.loudest[self._window(start, end)]
        winners = winners[winners >= 0]
        if winners.size == 0:
            return 0.0
        return float((winners == self.ids.index(name)).mean())

    def live_share(
        self,
        name: str,
        start: float,
        end: float,
    ) -> float:
        """How much of ``[start, end)`` one recording is live for.

        Unlike :meth:`share` this doesn't depend on the other recordings: a
        microphone can be live while another is louder, which is what two people
        talking at once looks like.
        """
        if name not in self.ids:
            return 0.0
        window = self._window(start, end)
        if window.start >= window.stop:
            return 0.0
        return float(self.live[self.ids.index(name)][window].mean())

live cached property

Which recordings are carrying speech at each moment.

loudest cached property

The row of the loudest live recording at each moment, or -1 if unclear.

live_share(name, start, end)

How much of [start, end) one recording is live for.

Unlike :meth:share this doesn't depend on the other recordings: a microphone can be live while another is louder, which is what two people talking at once looks like.

Source code in src/body_eye_sync/postprocessing/attribution.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def live_share(
    self,
    name: str,
    start: float,
    end: float,
) -> float:
    """How much of ``[start, end)`` one recording is live for.

    Unlike :meth:`share` this doesn't depend on the other recordings: a
    microphone can be live while another is louder, which is what two people
    talking at once looks like.
    """
    if name not in self.ids:
        return 0.0
    window = self._window(start, end)
    if window.start >= window.stop:
        return 0.0
    return float(self.live[self.ids.index(name)][window].mean())

share(name, start, end)

How much of [start, end) one recording is the loudest for, ignoring silent stretches.

Source code in src/body_eye_sync/postprocessing/attribution.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def share(
    self,
    name: str,
    start: float,
    end: float,
) -> float:
    """How much of ``[start, end)`` one recording is the loudest for, ignoring silent stretches."""
    if name not in self.ids:
        return 0.0
    winners = self.loudest[self._window(start, end)]
    winners = winners[winners >= 0]
    if winners.size == 0:
        return 0.0
    return float((winners == self.ids.index(name)).mean())

attribute_segments(transcripts, levels, timelines, settings, words=None, *, progress=None)

Give each transcribed segment to the wearer whose microphone won it.

Source code in src/body_eye_sync/postprocessing/attribution.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def attribute_segments(
    transcripts: dict[str, pd.DataFrame],
    levels: Levels,
    timelines: dict[str, Timeline],
    settings: SpeechPostProcessingSettings,
    words: dict[str, pd.DataFrame | None] | None = None,
    *,
    progress: Progress | None = None,
) -> pd.DataFrame:
    """Give each transcribed segment to the wearer whose microphone won it."""
    words = words or {}
    spoken = _spoken_words(words, timelines)

    segments: list[_Segment] = []
    for name, transcript in transcripts.items():
        if transcript.empty or name not in levels.ids:
            continue
        transcript = _split_attribution_segments(
            transcript,
            words.get(name),
            settings,
        )
        bounds = _on_experiment_clock(transcript, timelines.get(name, Timeline()))
        for (start, end), segment_id, text in zip(
            bounds, transcript["segment_id"], transcript["text"]
        ):
            segments.append(
                _Segment(
                    name,
                    int(segment_id),
                    start,
                    end,
                    str(text),
                    levels.share(name, start, end),
                    levels.live_share(name, start, end),
                )
            )

    # only include segments which were the loudest or live for more than the ownership share threshold
    spoke = [
        segment
        for segment in segments
        if segment.share > settings.ownership_share
        or segment.live > settings.ownership_share
    ]
    rows = sorted(
        (
            (
                turn.start,
                turn.end,
                turn.name,
                turn.segment_id,
                turn.text,
            )
            for turn in _accept(spoke, spoken, settings, progress)
        ),
        key=lambda row: (row[0], row[1], row[2]),
    )
    table = pd.DataFrame(
        [(index, *row) for index, row in enumerate(rows)], columns=TURN_COLUMNS
    )
    return table.astype(
        {
            "turn_id": int,
            "start": float,
            "end": float,
            "speaker": str,
            "source_segment_id": int,
            "text": str,
        }
    )

measure_levels(loudness, timelines, settings)

Put every recording's measured loudness on the experiment clock.

Source code in src/body_eye_sync/postprocessing/attribution.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def measure_levels(
    loudness: dict[str, pd.DataFrame],
    timelines: dict[str, Timeline],
    settings: SpeechPostProcessingSettings,
) -> Levels:
    """Put every recording's measured loudness on the experiment clock."""
    measured: dict[str, tuple[np.ndarray, np.ndarray]] = {}
    for name, table in loudness.items():
        if table is None or table.empty:
            continue
        levels = table["level_db"].to_numpy(dtype=float)
        timeline = timelines.get(name, Timeline())
        measured[name] = (
            timeline.to_experiment_times(table["time"].to_numpy(dtype=float)),
            levels,
        )
    if not measured:
        return Levels(
            [],
            np.empty(0),
            np.empty((0, 0)),
            settings.live_above_floor_db,
        )

    start = max(times[0] for times, _ in measured.values())
    end = min(times[-1] for times, _ in measured.values())
    if end <= start:
        # the recordings do not overlap, so there is nothing to compare
        return Levels(
            [],
            np.empty(0),
            np.empty((0, 0)),
            settings.live_above_floor_db,
        )

    grid = np.arange(start, end, _HOP_SECONDS)
    ids = sorted(measured)
    rows = [np.interp(grid, measured[name][0], measured[name][1]) for name in ids]
    matrix = np.vstack(rows)
    floors = np.percentile(matrix, settings.floor_percentile, axis=1, keepdims=True)
    return Levels(
        ids,
        grid,
        matrix - floors,
        settings.live_above_floor_db,
    )