utils.py

"""
utils.py — pure logic helpers with no Streamlit dependency.

Covers:
  - Audio processing (processFile, clip extraction, randomization)
  - DataFrame builders (build_df2 … build_df5)
  - Plotly figure builders (one function per chart tab)
  - Multi-file summary DataFrame builders
"""

import io
import random
import datetime as dt
import copy
import cv2

import numpy as np
import pandas as pd
import soundfile as sf
import torch
import plotly.express as px
import plotly.graph_objects as go

import sonogram_utility as su

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
CLIP_MIN_S = 3.0
CLIP_MAX_S = 5.0
CLIP_SCAN_STEP_S = 0.5

TRANSPARENT_BG = dict(
    plot_bgcolor="rgba(0,0,0,0)",
    paper_bgcolor="rgba(0,0,0,0)",
)

# ---------------------------------------------------------------------------
# Speaker sample helpers
# ---------------------------------------------------------------------------

# ---------------------------------------------------------------------------
# Color palette — loaded from plotly_colorwheel.txt at import time.
# Edit that file to change colors; no code changes needed.
# Structure: 8 colors × 3 shades (indices 0-23) + 1 grey (index 24).
# Reserved indices:
#   0  = Single Voice  (red shade 0)
#   3  = Multi Voice   (green shade 0)
#   24 = No Voice      (grey)
# Speakers cycle through all other indices in order.
# ---------------------------------------------------------------------------

def _load_palette(path="plotly_colorwheel.txt"):
    """Parse plotly_colorwheel.txt (id, color, shade, hex table) into a list.

    Returns a list indexed by id so _PALETTE[id] gives the hex color directly.
    Falls back to hardcoded defaults if file is missing.
    """
    FALLBACK = [
        "#de2d26","#fc9272","#fee0d2",  # red    0-2
        "#e6550d","#fdae6b","#feedde",  # orange 3-5
        "#d9b300","#fdd44d","#fff7bc",  # yellow 6-8
        "#31a354","#a1d99b","#e5f5e0",  # green  9-11
        "#1a9e9e","#66c2c2","#ccecec",  # teal   12-14
        "#3182bd","#9ecae1","#deebf7",  # blue   15-17
        "#756bb1","#bcbddc","#efedf5",  # purple 18-20
        "#d4679a","#f1b6d1","#fce4f0",  # pink   21-23
        "#999DA0",                       # grey   24
    ]
    try:
        entries = {}
        with open(path, "r") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                parts = [p.strip() for p in line.split(",")]
                try:
                    idx = int(parts[0])
                    # Support both formats:
                    #   id, color, shade, hex  (4 columns, hex at index 3)
                    #   id, color_name, hex    (3 columns, hex at index 2)
                    hex_color = next(
                        (p for p in reversed(parts) if p.startswith("#")), None
                    )
                    if hex_color:
                        entries[idx] = hex_color
                except ValueError:
                    continue
        if not entries:
            return FALLBACK
        max_idx = max(entries.keys())
        palette = [entries.get(i, "#cccccc") for i in range(max_idx + 1)]
        return palette
    except FileNotFoundError:
        return FALLBACK

_PALETTE = _load_palette()
# Reserved indices — must match plotly_colorwheel.txt:
#   0  = Single Voice, 9 = Multi Voice, 24 = No Voice, 31 = Unassigned
_RESERVED = {0, 9, 24, 31}
_SPEAKER_PALETTE = [c for i, c in enumerate(_PALETTE) if i not in _RESERVED]


def colorsCSS(n, startingHue=None, pool=None):
    """Return n CSS hex colors from the speaker palette.

    Cycles if n > len(_SPEAKER_PALETTE).
    startingHue and pool accepted for backwards compatibility but ignored.
    """
    if n == 0:
        return []
    pal = _SPEAKER_PALETTE if _SPEAKER_PALETTE else _PALETTE
    return [pal[i % len(pal)] for i in range(n)]


def extract_clip_bytes(waveform, sample_rate, seg_start, seg_end):
    """Return WAV bytes for the loudest SAMPLE_MIN–SAMPLE_MAX window in [seg_start, seg_end]."""
    total_samples = waveform.shape[-1]
    seg_start_s = int(seg_start * sample_rate)
    seg_end_s   = min(int(seg_end * sample_rate), total_samples)

    seg_dur      = (seg_end_s - seg_start_s) / sample_rate
    clip_dur     = min(max(min(seg_dur, CLIP_MAX_S), CLIP_MIN_S), seg_dur)
    clip_samples = int(clip_dur * sample_rate)

    best_start  = seg_start_s
    best_rms    = -1.0
    step_samples = int(CLIP_SCAN_STEP_S * sample_rate)

    pos = seg_start_s
    while pos + clip_samples <= seg_end_s:
        window = waveform[:, pos: pos + clip_samples].float()
        rms = float(window.pow(2).mean().sqrt())
        if rms > best_rms:
            best_rms   = rms
            best_start = pos
        pos += step_samples

    clip_np = waveform[:, best_start: best_start + clip_samples].numpy().T
    buf = io.BytesIO()
    sf.write(buf, clip_np, sample_rate, format="WAV", subtype="PCM_16")
    buf.seek(0)
    return buf.read()


def build_speaker_clips(annotations, waveform, sample_rate):
    """Return (samples_dict, segments_dict) for all speakers in annotations.

    samples_dict  : {speaker: wav_bytes}
    segments_dict : {speaker: [(start, end), ...]}
    """
    clips    = {}
    segments = {}

    for speaker in annotations.labels():
        speaker_segments = [
            seg for seg, _, label in annotations.itertracks(yield_label=True)
            if label == speaker
        ]
        if not speaker_segments:
            continue

        segments[speaker] = [(s.start, s.end) for s in speaker_segments]
        longest = max(speaker_segments, key=lambda s: s.duration)
        clips[speaker] = extract_clip_bytes(waveform, sample_rate, longest.start, longest.end)

    return clips, segments


def get_randomized_clip(waveform, sample_rate, segments):
    """Return WAV bytes for a random 3–5 s audio sample drawn from a random segment.

    segments : [(start, end), ...]  (all segments for one speaker)
    """
    durations  = [max(e - s, 0.01) for s, e in segments]
    total_dur  = sum(durations)
    rand_val   = random.random() * total_dur
    cumulative = 0.0
    chosen_start, chosen_end = segments[0]
    for (seg_s, seg_e), dur in zip(segments, durations):
        cumulative += dur
        if rand_val <= cumulative:
            chosen_start, chosen_end = seg_s, seg_e
            break

    seg_dur  = chosen_end - chosen_start
    clip_dur = min(max(min(seg_dur, CLIP_MAX_S), CLIP_MIN_S), seg_dur)
    max_offset = max(seg_dur - clip_dur, 0.0)
    offset     = random.uniform(0.0, max_offset)
    clip_start = chosen_start + offset
    clip_end   = clip_start + clip_dur

    return extract_clip_bytes(waveform, sample_rate, clip_start, clip_end)


# ---------------------------------------------------------------------------
# DataFrame builders  (called from analyze() in state.py)
# ---------------------------------------------------------------------------

def build_df3(noVoice, oneVoice, multiVoice):
    """Voice category totals DataFrame."""
    return pd.DataFrame({
        "values": [su.sumTimes(noVoice), su.sumTimes(oneVoice), su.sumTimes(multiVoice)],
        "names":  ["No Voice", "Single Voice", "Multi Voice"],
    })


def build_df4(speakerNames, categorySelections, categoryNames, currAnnotation):
    """Speaker-to-category time DataFrame. Returns (df4, nameList, valueList, extraNames, extraValues)."""
    nameList   = list(categoryNames)
    valueList  = [0.0] * len(nameList)
    extraNames : list = []
    extraValues: list = []

    for sp in speakerNames:
        found = False
        for i, _ in enumerate(nameList):
            if sp in categorySelections[i]:
                valueList[i] += su.sumTimes(currAnnotation.subset([sp]))
                found = True
                break
        if not found:
            extraNames.append(sp)
            extraValues.append(su.sumTimes(currAnnotation.subset([sp])))

    if extraNames:
        pairs = sorted(zip(extraNames, extraValues), key=lambda p: p[0])
        extraNames, extraValues = map(list, zip(*pairs))
    else:
        extraNames, extraValues = [], []

    df4 = pd.DataFrame({"values": valueList + extraValues, "names": nameList + extraNames})
    return df4, nameList, valueList, extraNames, extraValues


def build_df5(oneVoice, multiVoice, sumNoVoice, sumOneVoice, sumMultiVoice, currTotalTime):
    """Hierarchical voice-category DataFrame for sunburst / treemap."""
    speakerList,      timeList      = su.sumTimesPerSpeaker(oneVoice)
    multiSpeakerList, multiTimeList = su.sumMultiTimesPerSpeaker(multiVoice)

    speakerList      = list(speakerList)      if speakerList      else []
    timeList         = list(timeList)         if timeList         else []
    multiSpeakerList = list(multiSpeakerList) if multiSpeakerList else []
    multiTimeList    = list(multiTimeList)    if multiTimeList    else []

    # Filter out None/empty speaker labels — these come from annotationToNoiseList
    # when a time window has no identified speaker. Plotly silently drops or
    # crashes on None ids, causing the entire chart to disappear.
    ov_pairs = [(s, t) for s, t in zip(speakerList, timeList)
                if s is not None and str(s).strip() != ""]
    speakerList = [p[0] for p in ov_pairs]
    timeList    = [p[1] for p in ov_pairs]

    mv_pairs = [(s, t) for s, t in zip(multiSpeakerList, multiTimeList)
                if s is not None and str(s).strip() != ""]
    multiSpeakerList = [p[0] for p in mv_pairs]
    multiTimeList    = [p[1] for p in mv_pairs]

    safeTotalTime  = currTotalTime if currTotalTime > 0 else 1
    safeOneVoice   = sumOneVoice   if sumOneVoice   > 0 else 1
    summativeMulti = sum(multiTimeList) if multiTimeList else 1

    base = [sumNoVoice / safeTotalTime, sumOneVoice / safeTotalTime, sumMultiVoice / safeTotalTime]

    # Normalize child times so they sum exactly to their parent's total.
    # This guarantees branchvalues="total" is satisfied regardless of whether
    # the classifier's segment durations perfectly match sumOneVoice/sumMultiVoice.
    sumTimeList = sum(timeList) if timeList else 0
    if sumTimeList > 0:
        normTimeList = [t / sumTimeList * sumOneVoice for t in timeList]
    else:
        normTimeList = timeList

    sumMultiTimeList = sum(multiTimeList) if multiTimeList else 0
    if sumMultiTimeList > 0:
        normMultiTimeList = [t / sumMultiTimeList * sumMultiVoice for t in multiTimeList]
    else:
        normMultiTimeList = multiTimeList

    timeStrings      = su.timeToString(normTimeList)      if normTimeList      else []
    multiTimeStrings = su.timeToString(normMultiTimeList) if normMultiTimeList else []
    if isinstance(timeStrings, str):
        timeStrings = [timeStrings]
    if isinstance(multiTimeStrings, str):
        multiTimeStrings = [multiTimeStrings]

    n_ov = len(speakerList)
    n_mv = len(multiSpeakerList)

    return pd.DataFrame({
        "ids":    ["NV", "OV", "MV"] + [f"OV_{i}" for i in range(n_ov)] + [f"MV_{i}" for i in range(n_mv)],
        "labels": ["No Voice", "Single Voice", "Multi Voice"] + speakerList + multiSpeakerList,
        "parents":      ["", "", ""] + ["OV"] * n_ov + ["MV"] * n_mv,
        "parentNames":  ["Total", "Total", "Total"] + ["Single Voice"] * n_ov + ["Multi Voice"] * n_mv,
        "values":       [sumNoVoice, sumOneVoice, sumMultiVoice] + normTimeList + normMultiTimeList,
        "valueStrings": [
            su.timeToString(sumNoVoice),
            su.timeToString(sumOneVoice),
            su.timeToString(sumMultiVoice),
        ] + timeStrings + multiTimeStrings,
        "percentiles": [b * 100 for b in base]
            + [t / safeTotalTime * 100 for t in normTimeList]
            + [t / safeTotalTime * 100 for t in normMultiTimeList],
        "parentPercentiles": [b * 100 for b in base]
            + [t / safeOneVoice * 100 for t in normTimeList]
            + [t / summativeMulti * 100 for t in normMultiTimeList],
    })


def build_df2(df4_names, df4_values, currTotalTime):
    """Time-spoken DataFrame (raw seconds) used by the bar chart tab."""
    return pd.DataFrame({
        "values": list(df4_values),
        "names":  df4_names,
    })


# ---------------------------------------------------------------------------
# Plotly figure builders
# ---------------------------------------------------------------------------

def _save_fig(fig, *paths):
    """Try to write fig to each path; silently skip on failure."""
    for path in paths:
        try:
            fig.write_image(path)
        except Exception:
            pass


def build_fig_pie1(df3, catTypeColors):
    """Voice category pie chart."""
    fig = go.Figure()
    fig.update_layout(
        title_text="Percentage of each voice category",
        colorway=catTypeColors,
        **TRANSPARENT_BG,
    )
    fig.add_trace(go.Pie(values=df3["values"], labels=df3["names"], sort=False))
    return fig


def build_fig_pie2(df4, speakerNames, speaker_color_map, catColors, get_display_name_fn, currFile):
    """Speaker / category pie chart."""
    df4 = df4.copy()
    df4["names"] = df4["names"].apply(lambda s: get_display_name_fn(s, currFile))
    colors = [speaker_color_map.get(n, _SPEAKER_PALETTE[i % len(_SPEAKER_PALETTE)])
              for i, n in enumerate(df4["names"])]
    fig = go.Figure()
    fig.update_layout(title_text="Percentage of speakers per role", **TRANSPARENT_BG)
    fig.add_trace(go.Pie(values=df4["values"], labels=df4["names"],
                         marker_colors=colors, sort=False))
    return fig


def _voice_color_map(df5_labels, speaker_color_map):
    """Build the label->color map for sunburst/treemap charts.

    Single Voice → _PALETTE[0]  (red shade 0, reserved)
    Multi Voice  → _PALETTE[9]  (green shade 0, reserved)
    No Voice     → _PALETTE[-1] (grey, reserved)
    Speakers     → looked up from speaker_color_map for cross-chart consistency
    """
    top_labels = ["No Voice", "Single Voice", "Multi Voice"]
    color_map = {
        "Single Voice": _PALETTE[0],
        "Multi Voice":  _PALETTE[9],
        "No Voice":     _PALETTE[24],
        "Unassigned":   _PALETTE[31] if len(_PALETTE) > 31 else "#777a7d",
    }
    speaker_labels = [l for l in df5_labels if l not in top_labels]
    for i, lbl in enumerate(speaker_labels):
        color_map[lbl] = speaker_color_map.get(
            lbl, _SPEAKER_PALETTE[i % len(_SPEAKER_PALETTE)]
        )
    return color_map


def build_fig_sunburst(df5, catTypeColors, speaker_color_map, get_display_name_fn, currFile):
    """Sunburst voice-category chart."""
    df5 = df5.copy()
    df5["labels"]      = df5["labels"].apply(lambda s: get_display_name_fn(s, currFile))
    df5["parentNames"] = df5["parentNames"].apply(lambda s: get_display_name_fn(s, currFile))

    color_map = _voice_color_map(df5["labels"], speaker_color_map)

    fig = px.sunburst(
        df5,
        branchvalues="total",
        names="labels", ids="ids", parents="parents",
        values="percentiles",
        custom_data=["labels", "valueStrings", "percentiles", "parentNames", "parentPercentiles"],
        color="labels",
        title="Percentage of each voice category with speakers (Combination)",
        color_discrete_map=color_map,
    )
    fig.update_traces(hovertemplate="<br>".join([
        "<b>%{customdata[0]}</b>",
        "Duration: %{customdata[1]}s",
        "Percentage of Total: %{customdata[2]:.2f}%",
        "Parent: %{customdata[3]}",
        "Percentage of Parent: %{customdata[4]:.2f}%",
    ]))
    fig.update_layout(**TRANSPARENT_BG)
    return fig


def build_fig_sunburst_single(df5, speaker_color_map, get_display_name_fn, currFile):
    """Sunburst showing only Single Voice speakers."""
    df5 = df5.copy()
    df5["labels"]      = df5["labels"].apply(lambda s: get_display_name_fn(s, currFile))
    df5["parentNames"] = df5["parentNames"].apply(lambda s: get_display_name_fn(s, currFile))

    # Keep only the Single Voice parent row and its children
    keep_ids = {"OV"} | {row["ids"] for _, row in df5.iterrows()
                         if row["parents"] == "OV"}
    df5 = df5[df5["ids"].isin(keep_ids)].copy()
    # Re-root: Single Voice becomes the top-level (parent = "")
    df5.loc[df5["ids"] == "OV", "parents"] = ""

    color_map = {lbl: speaker_color_map.get(lbl, _SPEAKER_PALETTE[i % len(_SPEAKER_PALETTE)])
                 for i, lbl in enumerate(df5["labels"])}
    color_map["Single Voice"] = _PALETTE[0]

    fig = px.sunburst(
        df5,
        branchvalues="total",
        names="labels", ids="ids", parents="parents",
        values="percentiles",
        custom_data=["labels", "valueStrings", "percentiles", "parentNames", "parentPercentiles"],
        color="labels",
        title="Percentage of each voice category with speakers (Single Voice)",
        color_discrete_map=color_map,
    )
    fig.update_traces(hovertemplate="<br>".join([
        "<b>%{customdata[0]}</b>",
        "Duration: %{customdata[1]}s",
        "Percentage of Total: %{customdata[2]:.2f}%",
        "Parent: %{customdata[3]}",
        "Percentage of Parent: %{customdata[4]:.2f}%",
    ]))
    fig.update_layout(**TRANSPARENT_BG, font_color="#323236")
    return fig


def build_fig_sunburst_multi(df5, speaker_color_map, get_display_name_fn, currFile):
    """Sunburst showing only Multi Voice speakers."""
    df5 = df5.copy()
    df5["labels"]      = df5["labels"].apply(lambda s: get_display_name_fn(s, currFile))
    df5["parentNames"] = df5["parentNames"].apply(lambda s: get_display_name_fn(s, currFile))

    # Keep only the Multi Voice parent row and its children
    keep_ids = {"MV"} | {row["ids"] for _, row in df5.iterrows()
                         if row["parents"] == "MV"}
    df5 = df5[df5["ids"].isin(keep_ids)].copy()
    if df5.empty:
        return None
    # Re-root: Multi Voice becomes the top-level (parent = "")
    df5.loc[df5["ids"] == "MV", "parents"] = ""

    color_map = {lbl: speaker_color_map.get(lbl, _SPEAKER_PALETTE[i % len(_SPEAKER_PALETTE)])
                 for i, lbl in enumerate(df5["labels"])}
    color_map["Multi Voice"] = _PALETTE[9]

    fig = px.sunburst(
        df5,
        branchvalues="total",
        names="labels", ids="ids", parents="parents",
        values="percentiles",
        custom_data=["labels", "valueStrings", "percentiles", "parentNames", "parentPercentiles"],
        color="labels",
        title="Percentage of each voice category with speakers (Multiple Voices)",
        color_discrete_map=color_map,
    )
    fig.update_traces(hovertemplate="<br>".join([
        "<b>%{customdata[0]}</b>",
        "Duration: %{customdata[1]}s",
        "Percentage of Total: %{customdata[2]:.2f}%",
        "Parent: %{customdata[3]}",
        "Percentage of Parent: %{customdata[4]:.2f}%",
    ]))
    fig.update_layout(**TRANSPARENT_BG, font_color="#323236")
    return fig


def build_fig_treemap(df5, catTypeColors, speaker_color_map, get_display_name_fn, currFile):
    """Treemap voice-category chart."""
    df5 = df5.copy()
    df5["labels"]      = df5["labels"].apply(lambda s: get_display_name_fn(s, currFile))
    df5["parentNames"] = df5["parentNames"].apply(lambda s: get_display_name_fn(s, currFile))

    color_map = _voice_color_map(df5["labels"], speaker_color_map)

    fig = px.treemap(
        df5,
        branchvalues="total",
        names="labels", parents="parents", ids="ids",
        values="percentiles",
        custom_data=["labels", "valueStrings", "percentiles", "parentNames", "parentPercentiles"],
        color="labels",
        title="Division of speakers in each voice category",
        color_discrete_map=color_map,
    )
    fig.update_traces(hovertemplate="<br>".join([
        "<b>%{customdata[0]}</b>",
        "Duration: %{customdata[1]}s",
        "Percentage of Total: %{customdata[2]:.2f}%",
        "Parent: %{customdata[3]}",
        "Percentage of Parent: %{customdata[4]:.2f}%",
    ]))
    fig.update_layout(**TRANSPARENT_BG)
    return fig


def build_fig_timeline(speakers_dataFrame, currTotalTime, speaker_color_map, get_display_name_fn, currFile, mv_intervals=None):
    """Gantt-style speaker timeline with optional multi-voice vertical shading."""
    df = speakers_dataFrame.copy()
    df["Resource"] = df["Resource"].apply(lambda s: get_display_name_fn(s, currFile))

    base = dt.datetime.combine(dt.date.today(), dt.time.min)

    def to_audio_dt(s):
        if isinstance(s, (dt.datetime, pd.Timestamp)):
            midnight = s.replace(hour=0, minute=0, second=0, microsecond=0)
            seconds  = (s - midnight).total_seconds()
        else:
            seconds = float(s)
        return base + dt.timedelta(seconds=seconds)

    df["Start"]  = df["Start"].apply(to_audio_dt)
    df["Finish"] = df["Finish"].apply(to_audio_dt)

    fig = px.timeline(
        df, x_start="Start", x_end="Finish", y="Resource", color="Resource",
        title="Timeline of audio with speakers",
        color_discrete_map=speaker_color_map,
    )
    fig.update_yaxes(autorange=True)

    # Add vertical shading for multi-voice intervals.
    # Note: add_vrect does not support hover tooltips in Plotly — this is a
    # known Plotly limitation. Hover on these bands cannot be added without
    # introducing extra traces that corrupt the y-axis.
    for start_s, end_s in (mv_intervals or []):
        fig.add_vrect(
            x0=base + dt.timedelta(seconds=start_s),
            x1=base + dt.timedelta(seconds=end_s),
            fillcolor="#64A377",
            opacity=0.25,
            layer="below",
            line_width=0,
        )

    h = int(currTotalTime // 3600)
    m = int(currTotalTime %  3600 // 60)
    s = int(currTotalTime %  60)
    ms= int(currTotalTime * 1_000_000 % 1_000_000)
    time_max = dt.time(h, m, s, ms)

    fig.update_layout(
        xaxis_tickformatstops=[
            dict(dtickrange=[None, 1000], value="%H:%M:%S.%L"),
            dict(dtickrange=[1000, None], value="%H:%M:%S"),
        ],
        xaxis=dict(range=[
            dt.datetime.combine(dt.date.today(), dt.time.min),
            dt.datetime.combine(dt.date.today(), time_max),
        ]),
        xaxis_title="Time",
        yaxis_title=None,
        showlegend=False,
        yaxis={"showticklabels": True},
        **TRANSPARENT_BG,
    )
    return fig


def _seconds_to_hhmmss(seconds):
    """Convert a float seconds value to a hh:mm:ss.ss string."""
    seconds = float(seconds)
    h  = int(seconds // 3600)
    m  = int((seconds % 3600) // 60)
    s  = seconds % 60
    return f"{h:02d}:{m:02d}:{s:05.2f}"


def _darken_hex(hex_color, factor=0.55):
    """Return a darker version of a hex color by reducing brightness."""
    import colorsys
    h = hex_color.lstrip('#')
    r, g, b = int(h[0:2],16)/255, int(h[2:4],16)/255, int(h[4:6],16)/255
    hue, sat, val = colorsys.rgb_to_hsv(r, g, b)
    val = max(val * factor, 0.0)
    r2, g2, b2 = colorsys.hsv_to_rgb(hue, sat, val)
    return f"#{int(r2*255):02X}{int(g2*255):02X}{int(b2*255):02X}"


def build_fig_bar(df2, speakerNames, catColors, speaker_color_map, get_display_name_fn, currFile, mv_per_speaker=None):
    """Horizontal bar chart — time spoken per speaker (hh:mm:ss.ss).
    Each bar has a darker overlay showing the speaker's multi-voice portion.
    """
    mv_per_speaker = mv_per_speaker or {}
    df2 = df2.copy()
    df2 = df2[df2["names"].isin(speakerNames)]

    raw_to_display   = {sp: get_display_name_fn(sp, currFile) for sp in df2["names"]}
    df2["display"]   = df2["names"].map(raw_to_display)
    df2["mv_secs"]   = df2["names"].map(lambda sp: mv_per_speaker.get(sp, 0.0))
    df2["mv_secs"]   = df2["mv_secs"].clip(upper=df2["values"])  # cap at total time
    df2["sv_secs"]   = (df2["values"] - df2["mv_secs"]).clip(lower=0)
    df2["time_label"]    = df2["values"].apply(_seconds_to_hhmmss)
    df2["mv_time_label"] = df2["mv_secs"].apply(_seconds_to_hhmmss)

    disp_color_map = {raw_to_display[sp]: speaker_color_map.get(raw_to_display[sp], "#aaaaaa")
                      for sp in df2["names"]}
    disp_dark_map  = {d: _darken_hex(c) for d, c in disp_color_map.items()}

    # Sort descending so highest speaker is added first → top of chart,
    # SPEAKER_001 added last → bottom, matching Timeline order.
    df2 = df2.sort_values("names", ascending=False).reset_index(drop=True)

    fig = go.Figure()
    for _, row in df2.iterrows():
        col = disp_color_map.get(row["display"], "#aaaaaa")
        fig.add_trace(go.Bar(
            x=[row["sv_secs"]], y=[row["display"]], orientation="h",
            marker_color=col, showlegend=False,
            customdata=[[row["display"], row["time_label"], row["mv_time_label"]]],
            hovertemplate="<b>%{customdata[0]}</b><br>Total: %{customdata[1]}<br>Multi Voice: %{customdata[2]}<extra></extra>",
        ))
    for _, row in df2.iterrows():
        if row["mv_secs"] <= 0:
            continue
        dark = disp_dark_map.get(row["display"], "#555555")
        fig.add_trace(go.Bar(
            x=[row["mv_secs"]], y=[row["display"]], orientation="h",
            marker_color=dark, showlegend=False,
            customdata=[[row["display"], row["time_label"], row["mv_time_label"]]],
            hovertemplate="<b>%{customdata[0]}</b><br>Total: %{customdata[1]}<br>Multi Voice: %{customdata[2]}<extra></extra>",
        ))

    fig.update_layout(
        barmode="stack",
        title="Time spoken by each speaker",
        xaxis_title="Time Spoken",
        yaxis_title=None,
        showlegend=False,
        yaxis={"showticklabels": True},
        xaxis={"showticklabels": False},
        **TRANSPARENT_BG,
    )
    return fig


# ---------------------------------------------------------------------------
# Multi-file summary DataFrames
# ---------------------------------------------------------------------------

def build_multifile_category_df(validNames, results, summaries, categories, categorySelect,
                                speakerRenames=None):
    """Build df6 (category breakdown per file) for the multi-file expander.

    Uses su.sumTimes() per speaker (same as the single-file charts) so that:
    - Each speaker's time = union of their segments (overlaps within one speaker
      are merged by get_timeline().duration())
    - Multiple speakers in the same role are subset-unioned before summing so
      cross-speaker overlaps within a role are counted only once
    - Values are proportions (0-1) of the file's total duration

    speakerRenames: {filename: {raw_sp: display_name}} — applied to unassigned
    speaker column headers.
    """
    speakerRenames = speakerRenames or {}
    df6_dict      = {"files": validNames}
    allCategories = copy.deepcopy(categories)

    # First pass: discover unassigned speaker columns across all files
    for fn in validNames:
        currAnnotation, _ = results[fn]
        prefix = fn + ": "
        assigned = {
            t[len(prefix):]
            for tokens in categorySelect
            for t in tokens
            if t.startswith(prefix)
        }
        renames = speakerRenames.get(fn, {})
        for sp in currAnnotation.labels():
            if sp not in assigned:
                display = renames.get(sp, sp)
                if display not in allCategories:
                    allCategories.append(display)
                df6_dict.setdefault(display, [])

    for category in categories:
        df6_dict.setdefault(category, [])

    # Second pass: compute proportions per file
    for row_idx, fn in enumerate(validNames):
        currAnnotation, totalSeconds = results[fn]
        safe_total = max(totalSeconds, 1)
        prefix = fn + ": "
        renames = speakerRenames.get(fn, {})

        # Track which allCategories columns get a value this row
        filled = set()

        # For each role: union all assigned speakers into one subset, then sum.
        # su.sumTimes uses get_timeline(False).duration() which merges overlaps.
        for i, category in enumerate(categories):
            assigned_sps = [
                t[len(prefix):]
                for t in categorySelect[i]
                if t.startswith(prefix)
            ] if i < len(categorySelect) else []
            valid_sps = [sp for sp in assigned_sps if sp in currAnnotation.labels()]
            if valid_sps:
                val = su.sumTimes(currAnnotation.subset(valid_sps)) / safe_total
            else:
                val = 0.0
            df6_dict[category].append(min(val, 1.0))
            filled.add(category)

        # For unassigned speakers: each gets their own column
        assigned_all = {
            t[len(prefix):]
            for tokens in categorySelect
            for t in tokens
            if t.startswith(prefix)
        }
        unassigned = [sp for sp in currAnnotation.labels() if sp not in assigned_all]
        for sp in unassigned:
            display = renames.get(sp, sp)
            val = su.sumTimes(currAnnotation.subset([sp])) / safe_total
            df6_dict[display].append(min(val, 1.0))
            filled.add(display)

        # Fill 0 for every allCategories column not touched this row
        for category in allCategories:
            if category not in filled:
                df6_dict[category].append(0)

    # Normalize each file's row so values sum to 100
    df6 = pd.DataFrame(df6_dict)
    value_cols = [c for c in df6.columns if c != "files"]
    row_sums = df6[value_cols].sum(axis=1).replace(0, 1)
    df6[value_cols] = df6[value_cols].div(row_sums, axis=0) * 100
    return df6, allCategories


def build_multifile_role_voice_df(validNames, results, summaries, categories,
                                  categorySelect, speakerRenames=None):
    """Build df8: per-file proportions split by role for single voice, plus
    Multi Voice and No Voice.

    Single Voice time is broken down into each role and an Unassigned bucket
    (speakers in single-voice segments that haven't been assigned to any role).
    Multi Voice and No Voice come from df5 percentiles (0-100 scale) converted
    to 0-1 proportions.

    This is the combination of df6 (role proportions) and df7 (voice categories)
    where Single Voice is replaced by its constituent roles.
    """
    speakerRenames = speakerRenames or {}
    col_names = list(categories) + ["Unassigned", "Multi Voice", "No Voice"]
    df8_dict  = {"files": validNames}
    for col in col_names:
        df8_dict[col] = []

    for fn in validNames:
        currAnnotation, totalSeconds = results[fn]
        safe_total = max(totalSeconds, 1)
        prefix     = fn + ": "
        renames    = speakerRenames.get(fn, {})

        # Role proportions — same logic as build_multifile_category_df
        assigned_all = set()
        for i, category in enumerate(categories):
            assigned_sps = [
                t[len(prefix):]
                for t in (categorySelect[i] if i < len(categorySelect) else [])
                if t.startswith(prefix)
            ]
            valid_sps = [sp for sp in assigned_sps if sp in currAnnotation.labels()]
            assigned_all.update(valid_sps)
            if valid_sps:
                val = su.sumTimes(currAnnotation.subset(valid_sps)) / safe_total
            else:
                val = 0.0
            df8_dict[category].append(min(val, 1.0))

        # Unassigned speakers
        unassigned_sps = [sp for sp in currAnnotation.labels() if sp not in assigned_all]
        if unassigned_sps:
            val = su.sumTimes(currAnnotation.subset(unassigned_sps)) / safe_total
        else:
            val = 0.0
        df8_dict["Unassigned"].append(min(val, 1.0))

        # Multi Voice and No Voice from df5 percentiles (0-100 → 0-1)
        partial = summaries[fn]["df5"]
        df8_dict["No Voice"].append(partial["percentiles"][0] / 100)
        df8_dict["Multi Voice"].append(partial["percentiles"][2] / 100)

    # Normalize each file's row so values sum to 100
    df8 = pd.DataFrame(df8_dict)
    row_sums = df8[col_names].sum(axis=1).replace(0, 1)
    df8[col_names] = df8[col_names].div(row_sums, axis=0) * 100
    return df8, col_names


def build_multifile_voice_df(validNames, summaries):
    """Build df7 (no/one/multi voice percentages per file) for the multi-file expander.
    Values are normalized to sum to 100 per file.
    """
    voiceNames = ["No Voice", "Single Voice", "Multi Voice"]
    df7_dict   = {"files": validNames}
    for name in voiceNames:
        df7_dict[name] = []

    for fn in validNames:
        partial = summaries[fn]["df5"]
        for i, name in enumerate(voiceNames):
            df7_dict[name].append(partial["percentiles"][i])

    df7 = pd.DataFrame(df7_dict)
    row_sums = df7[voiceNames].sum(axis=1).replace(0, 1)
    df7[voiceNames] = df7[voiceNames].div(row_sums, axis=0) * 100
    return df7, voiceNames