render.py

"""Assemble the HTML noise report: summary, charts, and an honest methodology section.

Determinism: build_report takes `generated_at` as a preformatted string and reads only
the summary + config, so the same event log always yields byte-identical HTML (see the
snapshot test). The methodology and limitations sections are not optional — they are
written here unconditionally, and a merge-blocking test asserts their presence.
"""

from __future__ import annotations

import argparse
import dataclasses
from datetime import datetime, timedelta
from html import escape
from pathlib import Path
from typing import TYPE_CHECKING

from monitor import __version__
from monitor.config import Config
from monitor.detector import Event

from report.aggregate import (
    AmbientDay,
    Summary,
    describe_clock_anomalies,
    summarize,
    summarize_ambient,
)
from report.charts import bar_chart, heatmap

if TYPE_CHECKING:
    from collections.abc import Iterable
    from datetime import tzinfo

    from store import CalibrationEpoch, ClockAnomaly, Gap, MinuteLevel, Session

# Phrases the report-content gate checks for. Keeping them as constants makes the
# contract between the renderer and the test explicit.
METHODOLOGY_HEADING = "Methodology"
LIMITATIONS_HEADING = "Limitations"
RELATIVE_DBFS_NOTE = (
    "Levels are measured in dBFS, which is relative to digital full scale, not "
    "absolute sound pressure level (SPL) in dB. Without calibration against a "
    "reference meter, treat the numbers as relative, not absolute."
)
NO_SOURCE_NOTE = (
    "This tool measures sound levels only. It cannot prove what made a sound or "
    "where it came from; it does not record or identify any voice or source."
)
NO_CLOCK_ANOMALY_NOTE = "No clock anomalies detected during the measurement window."

#: What a peak or duration figure reads when there were no events to take it from.
NO_EVENTS_VALUE = "no events"

#: The main report's coverage statement when the record cannot support one. Said, not
#: omitted: the violations export already says this, and the main report used to print
#: nothing at all here, which reads as "the whole window was observed".
COVERAGE_UNDETERMINED_NOTE = (
    "How much of this reporting window the device actually monitored could not be "
    "determined from this record (no capture session, no recorded gap, and no measurable "
    "span of events). Do not read the counts above as covering the whole window."
)

# R5 — reader-facing "why there is deliberately no audio" note. The rationale already
# lives in docs/audits/recording-law-notes.md; this surfaces it to the neighbor / PM /
# board who reads the report, so the absence of audio reads as a privacy choice, not as
# missing data. General information, not jurisdiction-specific legal advice.
NO_AUDIO_RATIONALE = (
    "This device measures sound levels only — it never records, stores, or transmits any "
    "audio. That is a deliberate privacy choice, not missing data: each level reading is "
    "computed in memory and immediately discarded, so no speech and nothing intelligible "
    "is ever kept. There is no recording of anyone in this home or next door that could be "
    "leaked, subpoenaed, or misused. Recording a household or a neighbor can also raise "
    "consent and eavesdropping concerns under some recording laws; measuring levels only "
    "sidesteps that by never capturing content. (General information, not legal advice — "
    "check the rules where you live.)"
)

# R1 — a single, reusable plain-language "What this can and cannot prove" cover block.
# It restates limitations that already hold elsewhere in the report; it adds prominence,
# never a new claim. The same block is prepended to the report and to every exported
# artifact (the violations HTML and CSV) so the caveat travels with the file.
COVER_CAN = (
    "When sound at this device crossed a set loudness threshold, and for how long, with "
    "timestamps — an honest, time-stamped record of the pattern.",
    "How that pattern lines up with a quiet-hours window you configure.",
)
COVER_CANNOT = (
    "What made a sound, or who caused it — no audio is recorded, so there is no source "
    "attribution.",
    "Absolute loudness in dB SPL or dB(A): uncalibrated readings are relative dBFS, not "
    "the units an ordinance, lease, or HOA rule is written in.",
    "That any law, lease, or rule was broken — only the relevant authority decides that, "
    "and being within quiet hours is not the same as a violation.",
    "Anything about a place this device was not in — readings are specific to this "
    "microphone in this spot, and change if it moves.",
)
COVER_PRIVACY = (
    "By design no audio is ever recorded, stored, or transmitted, so there is nothing to "
    "leak, subpoena, or misuse. This is general information, not legal advice; verify your "
    "local rule before relying on these numbers."
)

# R3 — the no-verdict line. Any artifact that reports a quiet-hours count carries it:
# a count is a measurement, and a measurement is not a finding. Named as a constant so
# the browser port and the export gate are held to the same sentence.
NO_VERDICT_NOTE = (
    "This is a measurement, not a determination. Being within quiet hours is not the "
    "same as a violation, and only the relevant authority can decide whether a rule was "
    "broken."
)

# R2 — the headline of the uncalibrated banner. Uncalibrated readings must never get to
# look like dB(A)/SPL, in either implementation.
UNCALIBRATED_HEADLINE = "Uncalibrated — these readings are relative, not dB(A)."


def cover_text_lines() -> list[str]:
    """The cover block as plain-text lines, for the comment preamble of CSV exports."""
    lines = ["What this can and cannot prove", "", "What it can show:"]
    lines += [f"  - {x}" for x in COVER_CAN]
    lines += ["", "What it cannot prove:"]
    lines += [f"  - {x}" for x in COVER_CANNOT]
    lines += ["", COVER_PRIVACY]
    return lines


def cover_html() -> str:
    """The R1 cover block as an accessible HTML <section>. Deterministic; no new claims."""
    can = "".join(f"<li>{escape(x)}</li>" for x in COVER_CAN)
    cannot = "".join(f"<li>{escape(x)}</li>" for x in COVER_CANNOT)
    return (
        '<section class="cover" aria-label="What this report can and cannot prove">\n'
        "<h2>What this can and cannot prove</h2>\n"
        "<p><strong>What it can show:</strong></p>\n"
        f"<ul>{can}</ul>\n"
        "<p><strong>What it cannot prove:</strong></p>\n"
        f"<ul>{cannot}</ul>\n"
        f'<p class="note">{escape(COVER_PRIVACY)}</p>\n'
        "</section>"
    )


_STYLE = """
:root { color-scheme: light dark; }
body { font: 16px/1.5 system-ui, sans-serif; margin: 0; color: #111; background: #fff; }
.skip { position: absolute; left: -999px; }
.skip:focus { left: 8px; top: 8px; position: fixed; background: #fff; padding: 8px; }
main { max-width: 60rem; margin: 0 auto; padding: 1.5rem; }
h1 { font-size: 1.6rem; } h2 { font-size: 1.25rem; margin-top: 2rem; }
dl.stats { display: grid; grid-template-columns: max-content 1fr; gap: .25rem 1rem; }
dl.stats dt { font-weight: 600; }
figure.chart { margin: 1rem 0; border: 1px solid #ccc; padding: 1rem; }
figure.chart figcaption { font-weight: 600; margin-bottom: .5rem; }
table { border-collapse: collapse; margin-top: .75rem; width: 100%; }
caption { text-align: left; font-style: italic; margin-bottom: .25rem; }
th, td { border: 1px solid #bbb; padding: .25rem .5rem; text-align: left; }
.note { background: #f3f3f3; border-left: 4px solid #3b6ea5; padding: .75rem 1rem; }
section.cover { border: 1px solid #bbb; padding: .5rem 1.25rem 1rem; margin: 1rem 0; background: #fafafa; }
section.cover ul { margin: .25rem 0; }
.banner { padding: .75rem 1rem; margin: 1rem 0; border: 2px solid #b35900; background: #fff4e5; }
.banner.banner-ok { border-color: #2f6f3e; background: #eef7ef; }
:focus-visible { outline: 3px solid #3b6ea5; outline-offset: 2px; }
@media (prefers-reduced-motion: reduce) { * { animation: none !important; transition: none !important; } }
@media print {
  body { color: #000; background: #fff; }
  .skip { display: none; }
  .banner { border: 2px solid #000; }
  section.cover, figure.chart, table { break-inside: avoid; page-break-inside: avoid; }
  main { max-width: none; }
}
""".strip()


def _fmt_seconds(seconds: float) -> str:
    if seconds < 60:
        return f"{seconds:.0f} s"
    minutes = seconds / 60
    if minutes < 60:
        return f"{minutes:.1f} min"
    return f"{minutes / 60:.1f} h"


def _conditions_html(session: Session | None) -> str:
    """Measurement-conditions paragraph from the latest capture session (lineage)."""
    if session is None:
        return (
            "<p>Measurement conditions for this report were not recorded "
            "(events predate session tracking).</p>"
        )
    mic = f", microphone {escape(session.mic_model)}" if session.mic_model else ""
    placement = f" Placement: {escape(session.placement_note)}." if session.placement_note else ""
    total = session.frames_seen + session.frames_dropped
    coverage = (
        f" Frame coverage in the most recent session was "
        f"<strong>{session.frame_coverage:.1%}</strong> "
        f"({session.frames_seen:,} of {total:,} frames processed)."
        if total
        else ""
    )
    return (
        f"<p>Captured by device <strong>{escape(session.device_label)}</strong>{mic}."
        f"{placement}{coverage}</p>"
    )


def _clock_anomalies_html(lines: list[str]) -> str:
    """Disclose clock jumps within the report window (or state that there were none)."""
    if not lines:
        return f'<p class="note">{escape(NO_CLOCK_ANOMALY_NOTE)}</p>'
    items = "".join(f"<li>{escape(line)}</li>" for line in lines)
    return (
        '<p class="note">The system clock diverged from the monotonic clock during '
        "capture, so some event timestamps near these points may be off. On hardware "
        "without a real-time clock this typically happens when the network time syncs "
        f"after boot.</p>\n<ul>{items}</ul>"
    )


AMBIENT_APPROXIMATION_NOTE = (
    "Day figures are rolled up from per-minute summaries, not from raw samples (which no "
    "longer exist past the minute they summarized). Minimum and maximum are exact; median "
    "and L90 are computed across that day's per-minute median/L90 values, an approximation "
    "of the true day-level statistic."
)


def _ambient_html(ambient_days: list[AmbientDay]) -> str:
    """EXP-01: the opt-in ambient-baseline section, or "" when the ledger has no data.

    Omitted entirely (not an empty section) when `ambient_days` is empty, mirroring how
    the event-types section disappears without any tagged events -- this keeps the
    snapshot golden untouched for every report that does not enable
    `config.ambient_ledger`.
    """
    if not ambient_days:
        return ""
    rows = "".join(
        f'<tr><th scope="row">{escape(d.day)}</th>'
        f"<td>{d.min_dbfs:.1f} dBFS</td><td>{d.median_dbfs:.1f} dBFS</td>"
        f"<td>{d.max_dbfs:.1f} dBFS</td><td>{d.l90_dbfs:.1f} dBFS</td>"
        f"<td>{d.minutes_covered}</td></tr>"
        for d in ambient_days
    )
    return (
        "\n<h2>Ambient baseline</h2>\n"
        "<p>An opt-in, per-minute summary of the room's baseline level (never audio), "
        "so an event's peak can be read against what the room normally sounds like -- "
        "and so a quiet night is distinguishable from a period the device was not "
        "capturing anything at all. L90 is the level exceeded 90% of the time that day, "
        "a standard background-noise figure (steadier than the bare minimum).</p>\n"
        f'<p class="note">{escape(AMBIENT_APPROXIMATION_NOTE)}</p>\n'
        "<table><caption>Ambient baseline by day</caption>"
        '<thead><tr><th scope="col">Day</th><th scope="col">Min</th>'
        '<th scope="col">Median</th><th scope="col">Max</th><th scope="col">L90</th>'
        '<th scope="col">Minutes covered</th></tr></thead>'
        f"<tbody>{rows}</tbody></table>"
    )


PARAM_CHANGE_NOTE = (
    "Detection settings changed during this record; each event is described under the "
    "parameters in force when it was logged."
)


def _param_key(session: Session) -> tuple[object, ...]:
    """The detection-parameter tuple that defines a parameter epoch for a session."""
    return (
        session.threshold_dbfs,
        session.min_duration_s,
        session.debounce_s,
        session.sample_rate,
        session.frame_size,
    )


def _param_epochs(sessions: list[Session]) -> list[Session]:
    """Distinct detection-parameter sets in chronological order, each represented by the
    earliest session that used it. `sessions` is expected oldest-first."""
    epochs: dict[tuple[object, ...], Session] = {}
    for s in sessions:
        epochs.setdefault(_param_key(s), s)
    return list(epochs.values())


def _methodology_html(config: Config, calib_line: str, sessions: list[Session] | None) -> str:
    """The Methodology block. With one detection-parameter set (or none recorded) this is
    the single honest paragraph, sourcing values from the session when available and
    otherwise from config. When the parameters changed across sessions, it becomes a small
    table of parameter epochs plus a disclosure that events are described under the
    settings in force when they were logged."""
    epochs = _param_epochs(sessions) if sessions else []

    if len(epochs) > 1:
        rows = "".join(
            "<tr>"
            f'<th scope="row">{escape(_epoch_since(s, config))}</th>'
            f"<td>{_fmt_threshold(s.threshold_dbfs)}</td>"
            f"<td>{_fmt_secs_param(s.min_duration_s)}</td>"
            f"<td>{_fmt_secs_param(s.debounce_s)}</td>"
            f"<td>{escape(_fmt_sampling(s.sample_rate, s.frame_size))}</td>"
            "</tr>"
            for s in epochs
        )
        return (
            "<p>Each frame of audio is read into memory, reduced to a single "
            "root-mean-square level in dBFS, and then discarded. A noise event is recorded "
            "when the level stays at or above the threshold for at least the minimum "
            "duration; brief dips shorter than the debounce do not split one event into "
            "many. For each event we store start time, duration, and peak and average "
            f"level — six numbers, no audio. {calib_line}</p>\n"
            "<table><caption>Detection-parameter epochs</caption>"
            '<thead><tr><th scope="col">Since</th><th scope="col">Threshold</th>'
            '<th scope="col">Min duration</th><th scope="col">Debounce</th>'
            '<th scope="col">Sampling</th></tr></thead>'
            f"<tbody>{rows}</tbody></table>\n"
            f"<p>{escape(PARAM_CHANGE_NOTE)}</p>"
        )

    # Single parameter set (or none recorded): source from the one epoch when present,
    # falling back to config so pre-provenance records still render truthfully.
    epoch = epochs[0] if epochs else None
    threshold = _param_or(epoch, "threshold_dbfs", config.threshold_dbfs)
    min_duration = _param_or(epoch, "min_duration_s", config.min_duration_s)
    debounce = _param_or(epoch, "debounce_s", config.debounce_s)
    sample_rate = _param_or(epoch, "sample_rate", config.sample_rate)
    frame_size = _param_or(epoch, "frame_size", config.frame_size)
    return (
        f"<p>Each ~{frame_size / sample_rate * 1000:.0f} ms frame of audio is read\n"
        "into memory, reduced to a single root-mean-square level in dBFS, and then "
        "discarded.\n"
        "A noise event is recorded when the level stays at or above\n"
        f"<strong>{threshold:.0f} dBFS</strong> for at least\n"
        f"<strong>{min_duration:.1f} s</strong>; brief dips shorter than the\n"
        f"<strong>{debounce:.1f} s</strong> debounce do not split one event into many.\n"
        "For each event we store start time, duration, and peak and average level — six "
        "numbers,\n"
        f"no audio. {calib_line}</p>"
    )


def _param_or(session: Session | None, attr: str, fallback: float | int) -> float | int:
    if session is None:
        return fallback
    value = getattr(session, attr)
    return fallback if value is None else value


def _epoch_since(session: Session, config: Config) -> str:
    return datetime.fromtimestamp(session.started_at, tz=config.tzinfo()).date().isoformat()


def _fmt_threshold(value: float | None) -> str:
    return "—" if value is None else f"{value:.0f} dBFS"


def _fmt_secs_param(value: float | None) -> str:
    return "—" if value is None else f"{value:.1f} s"


def _fmt_sampling(sample_rate: int | None, frame_size: int | None) -> str:
    if sample_rate is None or frame_size is None:
        return "—"
    return f"{sample_rate} Hz / {frame_size}-sample frames"


def _epoch_index_at(history: list[CalibrationEpoch], ts: float) -> int | None:
    """Index into `history` (ascending by effective_from) of the epoch in force at `ts`.

    A timestamp before the first epoch resolves to that first epoch (epoch 0 covers all
    historical rows); None only when the history is empty.
    """
    if not history:
        return None
    chosen = 0
    for i, epoch in enumerate(history):
        if epoch.effective_from <= ts:
            chosen = i
        else:
            break
    return chosen


def _offset_at(history: list[CalibrationEpoch], ts: float) -> float:
    """The calibration offset in force at `ts`, or 0.0 when no calibration exists."""
    idx = _epoch_index_at(history, ts)
    return 0.0 if idx is None else history[idx].offset


def _epochs_covering(
    history: list[CalibrationEpoch], events: list[Event]
) -> list[CalibrationEpoch]:
    """The subset of epochs that are in force for at least one of `events`, in order."""
    if not history or not events:
        return []
    used = {idx for ev in events if (idx := _epoch_index_at(history, ev.start)) is not None}
    return [history[i] for i in sorted(used)]


def _apply_offset(event: Event, offset: float) -> Event:
    """Return the event with its stored raw levels shifted by a calibration offset."""
    if offset == 0.0:
        return event
    return dataclasses.replace(
        event, peak_level=event.peak_level + offset, avg_level=event.avg_level + offset
    )


def _per_event_offsets(
    events: list[Event], history: list[CalibrationEpoch], config: Config
) -> list[float]:
    """The calibration offset applied to each event at render time (parallel list).

    This is the single resolver every rendered artifact must go through — the HTML
    report and the CSV/violations exports all adjust levels with exactly these values,
    so no two artifacts generated from the same log can disagree numerically.
    Attribution is by event *start*: an event that straddles a recalibration uses the
    epoch in force when it began. With no calibration history at all, the config's
    bootstrap offset applies uniformly (the deprecated-but-supported fallback).
    """
    if history:
        return [_offset_at(history, ev.start) for ev in events]
    return [config.calibration_offset] * len(events)


# --- Calibration basis: was the offset a row carries in force when the row was measured?
#
# `_epoch_index_at` resolves a timestamp before the first calibration epoch to that first
# epoch ("epoch 0 covers all historical rows"). That is a deliberate design choice: it
# keeps every level in a report on one scale, and re-rendering a date range yields the same
# numbers before and after a recalibration (ADR-0003). What was missing was any disclosure
# that it happened. "Calibrate once, after a week or two of logging" is the ordinary way
# this tool gets used, and until this was added that was exactly the case with no caveat:
# a single-epoch report rendered the earlier events as calibrated SPL estimates, and the
# exports stamped every row with the same offset, on the strength of a measurement taken
# days after those events — with no marker of any kind. The basis travels with every row
# and the report names the count, so a reader can tell a reading the calibration vouched
# for from one it was extended back over.

#: The epoch whose offset this row carries was in force when the row was measured.
CALIBRATION_IN_FORCE = "in-force"
#: The row predates the first calibration; the first epoch's offset was applied to it
#: retroactively. The calibration postdates the measurement.
CALIBRATION_BACK_APPLIED = "back-applied"
#: No calibration history exists; the deprecated config bootstrap offset applies uniformly.
CALIBRATION_BOOTSTRAP = "bootstrap-config"
#: No calibration at all: the row is raw, relative dBFS.
CALIBRATION_NONE = "none"
#: The caller did not say. Exports default to this rather than guessing a basis.
CALIBRATION_UNSTATED = "unstated"


def _basis_at(history: list[CalibrationEpoch], ts: float, config: Config) -> str:
    if history:
        return CALIBRATION_BACK_APPLIED if ts < history[0].effective_from else CALIBRATION_IN_FORCE
    return CALIBRATION_BOOTSTRAP if config.calibration_offset != 0.0 else CALIBRATION_NONE


def _per_event_basis(
    events: list[Event], history: list[CalibrationEpoch], config: Config
) -> list[str]:
    """For each event (parallel to `_per_event_offsets`), whether its offset was in force
    when it was measured, back-applied from a later calibration, or not a calibration at
    all. Attribution is by event start, the same rule the offsets use."""
    return [_basis_at(history, ev.start, config) for ev in events]


@dataclasses.dataclass(frozen=True)
class BackApplied:
    """How much of a report's data predates its first calibration.

    `count` events (of `total`) and `minutes` ambient-ledger minutes started before the
    first epoch took effect and carry its offset retroactively. `None` from
    `back_applied_summary` means nothing was back-applied — including the legacy case
    where the first epoch is the v2->v3 migration's epoch 0 at `effective_from = 0`,
    which genuinely does cover everything and has its own caveat (ADR-0003).
    """

    count: int
    total: int
    minutes: int
    effective_from: float
    offset: float
    reference_instrument: str | None


def back_applied_summary(
    events: list[Event],
    history: list[CalibrationEpoch],
    minutes: list[MinuteLevel] | None = None,
) -> BackApplied | None:
    if not history:
        return None
    first = history[0]
    count = sum(1 for ev in events if ev.start < first.effective_from)
    minute_count = sum(1 for m in (minutes or []) if m.minute_start < first.effective_from)
    if count == 0 and minute_count == 0:
        return None
    return BackApplied(
        count=count,
        total=len(events),
        minutes=minute_count,
        effective_from=first.effective_from,
        offset=first.offset,
        reference_instrument=first.reference_instrument,
    )


def back_applied_sentence(ba: BackApplied, *, tz: tzinfo) -> str:
    """The disclosure, in words a reader can act on: how many rows, which date, and what
    the retroactive application assumes."""
    when = datetime.fromtimestamp(ba.effective_from, tz=tz).strftime("%Y-%m-%d %H:%M %Z")
    against = f" against {ba.reference_instrument}" if ba.reference_instrument else ""
    parts = []
    if ba.count:
        parts.append(f"{ba.count} of {ba.total} events")
    if ba.minutes:
        parts.append(f"{ba.minutes} ambient-ledger minutes")
    what = " and ".join(parts)
    return (
        f"{what} were recorded before the first calibration, which was taken on {when}"
        f"{against}. The offset measured then ({ba.offset:+.1f} dB) has been applied to "
        "them retroactively so every level here is on one scale, but for those readings "
        "the calibration postdates the measurement: it assumes the microphone, its gain, "
        "and its placement were unchanged in between, and nothing in this record can "
        "confirm that. Each exported row says whether its offset was in force when it "
        "was measured or back-applied."
    )


def _apply_offset_minute(minute: MinuteLevel, offset: float) -> MinuteLevel:
    """Return an ambient-ledger minute (EXP-01) with its four scalars shifted by a
    calibration offset. A dB offset is additive, so shifting min/median/max/L90 by the
    same amount preserves their relative shape exactly."""
    if offset == 0.0:
        return minute
    return dataclasses.replace(
        minute,
        min_dbfs=minute.min_dbfs + offset,
        median_dbfs=minute.median_dbfs + offset,
        max_dbfs=minute.max_dbfs + offset,
        l90_dbfs=minute.l90_dbfs + offset,
    )


def _per_minute_offsets(
    minutes: list[MinuteLevel], history: list[CalibrationEpoch], config: Config
) -> list[float]:
    """The calibration offset applied to each ambient-ledger minute (EXP-01), the same
    resolution rule `_per_event_offsets` uses for events, attributed by `minute_start`."""
    if history:
        return [_offset_at(history, m.minute_start) for m in minutes]
    return [config.calibration_offset] * len(minutes)


def _fmt_effective_from(effective_from: float, tz: tzinfo) -> str:
    """Human label for an epoch boundary; epoch 0 is the start of the record."""
    if effective_from <= 0.0:
        return "start of record"
    return datetime.fromtimestamp(effective_from, tz=tz).strftime("%Y-%m-%d %H:%M %Z")


def _calibration_epochs_html(
    epochs: list[CalibrationEpoch], *, tz: tzinfo, back_applied_line: str = ""
) -> str:
    """A per-epoch offsets table plus a disclosure line, for a multi-epoch window.

    `back_applied_line` (already HTML-escaped) is the pre-calibration disclosure; the
    multi-epoch path needs it too, since rows before the *first* epoch are back-applied
    here exactly as they are on the single-epoch path.
    """
    rows = "".join(
        f'<tr><th scope="row">{escape(_fmt_effective_from(e.effective_from, tz))}</th>'
        f"<td>{e.offset:+.1f} dB</td>"
        f"<td>{escape(e.reference_instrument) if e.reference_instrument else '—'}</td>"
        f"<td>{escape(e.note) if e.note else '—'}</td></tr>"
        for e in epochs
    )
    return (
        "\n<p>This reporting window spans <strong>more than one calibration epoch</strong>. "
        "Each event's level is adjusted by the offset that was in force when it began, "
        "so recalibrating does not rewrite stored numbers. One caveat: events stored by "
        "versions of this tool from before the calibration history existed had any "
        "then-configured offset already included in their stored levels; if that offset "
        "was nonzero, those older rows render over-adjusted here. The database records "
        "when that upgrade happened, so affected rows are identifiable. "
        "The offsets applied are:"
        "</p>\n"
        + (f"<p>{back_applied_line}</p>\n" if back_applied_line else "")
        + "<table><caption>Calibration offsets by epoch</caption>"
        '<thead><tr><th scope="col">Effective from</th><th scope="col">Offset</th>'
        '<th scope="col">Reference instrument</th><th scope="col">Note</th></tr></thead>'
        f"<tbody>{rows}</tbody></table>"
    )


def build_report(
    summary: Summary,
    *,
    config: Config,
    generated_at: str,
    calibration_offset: float | None = None,
    calibration_note: str | None = None,
    calibration_epochs: list[CalibrationEpoch] | None = None,
    session: Session | None = None,
    sessions: list[Session] | None = None,
    unmonitored: set[tuple[str, int]] | None = None,
    monitored_hours: float | None = None,
    wall_clock_hours: float | None = None,
    clock_anomaly_lines: list[str] | None = None,
    ambient_days: list[AmbientDay] | None = None,
    back_applied: BackApplied | None = None,
    title: str = "Olive's Bark Logger — Noise Report",
) -> str:
    """Render the full report as a single self-contained HTML string.

    `back_applied` (see `back_applied_summary`) says how many rows predate the first
    calibration and carry its offset retroactively; when given, the disclosure is
    rendered in the calibration banner and the methodology line on *both* the
    single-offset and the multi-epoch paths, because a report with one epoch is the
    ordinary case and was the one with no caveat at all.

    When `calibration_epochs` holds more than one epoch, a per-epoch offsets table and a
    recalibration disclosure are rendered; otherwise the single-offset path is used and
    `calibration_offset`/`calibration_note` describe the one offset in force.
    `ambient_days` (EXP-01) is opt-in and normally empty; when empty the ambient-baseline
    section is omitted entirely rather than rendered blank, so a report generated without
    the ledger enabled is byte-identical to what it would have rendered before EXP-01.
    """
    offset = config.calibration_offset if calibration_offset is None else calibration_offset
    note = config.calibration_note if calibration_note is None else calibration_note
    epochs = calibration_epochs or []
    multi_epoch = len(epochs) > 1
    calibrated = offset != 0.0
    conditions_html = _conditions_html(session)
    clock_html = _clock_anomalies_html(clock_anomaly_lines or [])
    ambient_html = _ambient_html(ambient_days or [])

    hour_chart = bar_chart(
        chart_id="by-hour",
        title="Events by hour of day",
        labels=[f"{h:02d}" for h in range(24)],
        values=[float(summary.by_hour.get(h, 0)) for h in range(24)],
        value_caption="events",
    )
    day_labels = list(summary.by_day.keys())
    day_chart = bar_chart(
        chart_id="by-day",
        title="Events by day",
        labels=day_labels if day_labels else ["(no data)"],
        values=[float(v) for v in summary.by_day.values()] if day_labels else [0.0],
        value_caption="events",
    )

    if summary.by_day_hour:
        heat_days = list(summary.by_day_hour.keys())
        heat_grid = [[summary.by_day_hour[d][h] for h in range(24)] for d in heat_days]
        unmon_note = (
            " Hours the device was not listening are hatched and labeled "
            '"not monitored" in the table, so absence of data is never read as quiet.'
            if unmonitored
            else ""
        )
        calendar_section = (
            "\n<h2>Calendar heatmap</h2>\n"
            "<p>Each cell is the number of sound-level events that began in that hour, by "
            "day and hour of day. Every calendar day in the reporting window has a row, "
            "including days with no events, so a quiet day is visible as a quiet day "
            "rather than missing. Darker cells saw more events; the count is printed in "
            "every non-empty cell and repeated in the data table below, so the pattern "
            "does not depend on color. These are event counts only — never audio."
            f"{unmon_note}</p>\n"
            + heatmap(
                chart_id="calendar",
                title="Events by day and hour",
                day_labels=heat_days,
                grid=heat_grid,
                unmonitored=unmonitored,
            )
        )
    else:
        calendar_section = (
            "\n<h2>Calendar heatmap</h2>\n<p>No events have been logged yet, so there is "
            "no calendar to show.</p>"
        )

    quiet_window = config.quiet_hours.label()

    # With no events there is no peak to report. `summarize` returns 0.0 for the empty
    # case, and 0.0 dBFS is digital full scale — the loudest reading the device can
    # produce — so printing it would state that a silent log hit maximum loudness.
    # Absence is written as absence.
    if summary.event_count:
        longest = _fmt_seconds(summary.longest_event_seconds)
        loudest = f"{summary.loudest_peak_dbfs:.1f} dBFS"
        mean_peak = f"{summary.mean_peak_dbfs:.1f} dBFS"
    else:
        longest = loudest = mean_peak = NO_EVENTS_VALUE
    stats = {
        "Total events": str(summary.event_count),
        "Total loud time": _fmt_seconds(summary.total_loud_seconds),
        "Longest event": longest,
        "Loudest peak": loudest,
        "Mean peak": mean_peak,
        f"Events during quiet hours ({quiet_window})": str(summary.quiet_hours_event_count),
        "Loud time during quiet hours (pro-rated)": _fmt_seconds(summary.quiet_hours_loud_seconds),
        "Loud time during quiet hours (start-attributed)": _fmt_seconds(
            summary.quiet_hours_loud_seconds_start_attributed
        ),
    }
    stats_html = "".join(f"<dt>{escape(k)}</dt><dd>{escape(v)}</dd>" for k, v in stats.items())

    back_applied_line = (
        escape(back_applied_sentence(back_applied, tz=config.tzinfo()))
        if back_applied is not None
        else ""
    )
    if multi_epoch:
        calib_line = (
            "Levels are adjusted for calibration at render time from an append-only "
            "history; because this window spans more than one calibration epoch, each "
            "event uses the offset in force when it was measured (see the table below)."
        )
        calib_epochs_section = _calibration_epochs_html(
            epochs, tz=config.tzinfo(), back_applied_line=back_applied_line
        )
    else:
        calib_line = (
            f"A calibration offset of {offset:+.1f} dB is applied "
            f"({escape(note)}). Readings approximate SPL but remain estimates."
            if calibrated
            else f"No calibration offset is applied ({escape(note)})."
        )
        calib_epochs_section = ""
    if back_applied_line:
        calib_line = f"{calib_line} {back_applied_line}"

    # R2 — unmissable calibration-honesty banner. Uncalibrated readings must never get to
    # look like dB(A)/SPL; when calibrated, the reference-instrument provenance (carried in
    # the calibration note) is surfaced prominently rather than buried in methodology.
    if calibrated:
        banner_html = (
            '<aside class="banner banner-ok" role="note" aria-label="Calibration status">\n'
            f"<strong>Calibrated.</strong> An offset of {offset:+.1f} dB is applied "
            f"({escape(note)}). Readings approximate sound level (SPL) but remain estimates "
            "affected by microphone, placement, and room acoustics.\n"
            + (
                f"<p><strong>Calibration postdates some readings.</strong> {back_applied_line}</p>\n"
                if back_applied_line
                else ""
            )
            + "</aside>"
        )
    else:
        banner_html = (
            '<aside class="banner" role="note" aria-label="Calibration status">\n'
            f"<strong>{UNCALIBRATED_HEADLINE}</strong> "
            "Levels are relative dBFS, not absolute sound level in dB(A) or dB SPL. Do not "
            "read them as the decibel numbers an ordinance or lease specifies; only their "
            "pattern relative to each other on this device is meaningful. Run "
            "<code>olive-calibrate</code> against a reference meter to estimate SPL (still "
            "an estimate, not a Class 1/2 sound-level-meter reading).\n"
            "</aside>"
        )

    # R3 — quiet-hours duration rollup. Ordinances/CC&Rs commonly key on accumulated
    # duration in a day; this totals detected loud time within the configured window, per
    # day, WITHOUT rendering a verdict. The no-verdict framing is mandatory.
    if summary.quiet_hours_loud_seconds_by_day:
        rollup_rows = "".join(
            f'<tr><th scope="row">{escape(day)}</th><td>{_fmt_seconds(secs)}</td></tr>'
            for day, secs in summary.quiet_hours_loud_seconds_by_day.items()
        )
        rollup_section = (
            "\n<h2>Quiet-hours duration rollup</h2>\n"
            "<p>Detected loud time within the quiet-hours window, totaled per day and "
            "attributed by each event's start time. Some ordinances and CC&amp;Rs key on "
            "accumulated duration in a day — figures of around <strong>30 minutes "
            "continuous</strong> or <strong>60 minutes intermittent</strong> are sometimes "
            "cited — but the threshold, the unit, and the definition vary by jurisdiction.</p>\n"
            f'<div class="note"><p>{NO_VERDICT_NOTE} Compare these durations against your '
            "own local ordinance, lease, or HOA rule.</p></div>\n"
            "<table><caption>Loud time within quiet hours, per day</caption>"
            '<thead><tr><th scope="col">Day</th>'
            '<th scope="col">Loud time within quiet hours</th></tr></thead>'
            f"<tbody>{rollup_rows}</tbody></table>"
        )
    else:
        rollup_section = (
            "\n<h2>Quiet-hours duration rollup</h2>\n"
            "<p>No events fell within the quiet-hours window, so there is nothing to roll "
            "up. Being within quiet hours is not the same as a violation in any case.</p>"
        )

    methodology_html = _methodology_html(config, calib_line, sessions)

    if monitored_hours is not None and wall_clock_hours is not None:
        coverage_html = (
            f"<p>Over this reporting window the device monitored "
            f"{monitored_hours:.1f} of {wall_clock_hours:.1f} wall-clock hours; the "
            "remainder is shown as not monitored rather than quiet.</p>\n"
        )
    else:
        coverage_html = f'<p class="note">{escape(COVERAGE_UNDETERMINED_NOTE)}</p>\n'

    tags_section = ""
    if summary.by_tag:
        rows = "".join(
            f'<tr><th scope="row">{escape(tag)}</th><td>{count}</td></tr>'
            for tag, count in summary.by_tag.items()
        )
        tags_section = (
            "\n<h2>Event types (coarse hint)</h2>\n"
            "<p>A crude, on-device classification of each event as bark-like or ambient, "
            "from sound shape only. It is a hint, not a fact, and it cannot identify a "
            "source.</p>\n"
            "<table><caption>Events by coarse type</caption>"
            '<thead><tr><th scope="col">Type</th><th scope="col">Events</th></tr></thead>'
            f"<tbody>{rows}</tbody></table>"
        )

    return f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{escape(title)}</title>
<style>{_STYLE}</style>
</head>
<body>
<a class="skip" href="#main">Skip to report</a>
<main id="main">
<h1>{escape(title)}</h1>
<p>Generated {escape(generated_at)}. This report summarizes sound-level <em>events</em> —
when sound crossed a threshold and for how long. No audio was recorded, stored, or
transmitted to produce it.</p>

{cover_html()}

{banner_html}

<h2>Summary</h2>
<dl class="stats">{stats_html}</dl>

<h2>Measurement conditions</h2>
{conditions_html}
{clock_html}

<h2>Distributions</h2>
{hour_chart}
{day_chart}
{calendar_section}{ambient_html}

{tags_section}
<h2>Quiet hours</h2>
<p>Quiet-hours window: <strong>{quiet_window}</strong> in time zone
<strong>{escape(config.tz)}</strong> (daylight-saving aware). Of {summary.event_count}
total events, <strong>{summary.quiet_hours_event_count}</strong> started within quiet hours,
and <strong>{_fmt_seconds(summary.quiet_hours_loud_seconds)}</strong> of loud time fell
inside the window
(vs. {_fmt_seconds(summary.quiet_hours_loud_seconds_start_attributed)} if counted whole by
start time). Event counts are attributed by start time, since a count cannot be split; loud
seconds are pro-rated across the quiet-window boundary, so an event that begins before the
window and ends inside it contributes only the seconds actually within quiet hours.</p>
{rollup_section}

<h2>Why there is deliberately no audio</h2>
<div class="note"><p>{escape(NO_AUDIO_RATIONALE)}</p></div>

<h2>{METHODOLOGY_HEADING}</h2>
{methodology_html}
{coverage_html}{calib_epochs_section}
<h2>{LIMITATIONS_HEADING}</h2>
<div class="note">
<p>{escape(RELATIVE_DBFS_NOTE)}</p>
<p>{escape(NO_SOURCE_NOTE)}</p>
<p>Microphone placement and room acoustics affect every reading; an event count
reflects this device in this spot, not an absolute fact about the building. These
numbers are offered to inform, not to manufacture a case.</p>
</div>
</main>
</body>
</html>
"""


Span = tuple[float, float]


def _merge_spans(spans: Iterable[Span]) -> list[Span]:
    """Union of half-open intervals: sorted, non-overlapping, empty ones dropped.

    Merging matters for correctness, not tidiness: coverage arithmetic subtracts these
    from a window, and two overlapping intervals summed independently would subtract the
    shared seconds twice.
    """
    merged: list[Span] = []
    for lo, hi in sorted(s for s in spans if s[1] > s[0]):
        if merged and lo <= merged[-1][1]:
            if hi > merged[-1][1]:
                merged[-1] = (merged[-1][0], hi)
        else:
            merged.append((lo, hi))
    return merged


def _subtract_spans(base: list[Span], holes: list[Span]) -> list[Span]:
    """``base`` minus ``holes``. Both must already be merged (see `_merge_spans`)."""
    out: list[Span] = []
    for lo, hi in base:
        cursor = lo
        for h_lo, h_hi in holes:
            if h_hi <= cursor:
                continue
            if h_lo >= hi:
                break
            if h_lo > cursor:
                out.append((cursor, h_lo))
            cursor = h_hi
            if cursor >= hi:
                break
        if cursor < hi:
            out.append((cursor, hi))
    return out


def _clip_spans(spans: Iterable[Span], window: Span) -> list[Span]:
    """Every span trimmed to ``window``, merged."""
    win_start, win_end = window
    return _merge_spans((max(lo, win_start), min(hi, win_end)) for lo, hi in spans)


def _span_seconds(spans: Iterable[Span]) -> float:
    return sum(hi - lo for lo, hi in spans)


def _session_end(session: Session) -> float:
    """The last moment this session can be *shown* to have been capturing.

    ``ended_at`` when the run recorded one. When it did not, the run died before its
    shutdown path could write one — a crash, a power cut, a SIGKILL — and the frame
    counters are the only evidence left. They are checkpointed on a wall-clock cadence
    (``checkpoint_interval_s``), so the start time plus the duration of the frames
    actually accounted for is the last instant the record vouches for. Running such a
    session to the end of the window instead would hand a dead monitor credit for every
    hour it was dead, which is exactly the defect this module is being corrected for; a
    live monitor loses at most one checkpoint interval, which errs toward claiming less.

    A session with neither an end nor usable framing metadata (a legacy row from before
    those columns existed) vouches for nothing past its own start. Its events still
    count: `on_air_spans` unions them in separately.

    The rule itself lives on the store's `Session.last_vouched_at`, because retention
    applies the same one when deciding whether a session row is old enough to delete:
    a session is never credited as listening for longer than it is kept, or kept for
    longer than it is credited.
    """
    return session.last_vouched_at


def on_air_spans(events: list[Event], sessions: list[Session], window: Span) -> list[Span] | None:
    """Stretches of ``window`` when the device is known to have been listening.

    Derived from the capture-session ledger, because the gap ledger cannot answer this:
    a gap row is written *by the running monitor* (``resilient_source`` reporting a
    device error), so the most ordinary outage of all — the monitor not running, after a
    stop, a reboot, a crash, or a power cut — leaves no gap behind. It does leave two
    session rows with a hole between them, and that hole is what this reads.

    Each session runs from its start to `_session_end`. Each event's own span is unioned
    in as well: a logged event is proof the device was listening at that moment, whatever
    the session rows do or do not say.

    Returns None — "cannot be determined", not "fully covered" — for a log with no
    session rows at all (written before session tracking existed). Callers fall back to
    the older whole-window-minus-recorded-gaps figure there rather than inventing an
    outage the record cannot support.
    """
    if not sessions:
        return None
    spans: list[Span] = [(s.started_at, _session_end(s)) for s in sessions]
    spans += [(e.start, e.end) for e in events]
    return _clip_spans(spans, window)


def off_air_spans(
    events: list[Event], gaps: list[Gap], sessions: list[Session]
) -> list[Span] | None:
    """Stretches of the reporting window with no monitor running at all, or None.

    None means the record cannot say (no sessions, or no determinable window). These are
    reported separately from the gap ledger because they are a different fact about a
    different kind of outage, and a reader handed "20% coverage" alongside "no monitoring
    gaps were recorded" deserves to see where the other 80% went.
    """
    window = _coverage_window(events, gaps, sessions)
    if window is None:
        return None
    on_air = on_air_spans(events, sessions, window)
    if on_air is None:
        return None
    return _subtract_spans([window], on_air)


def _fill_window_days(
    by_day_hour: dict[str, dict[int, int]], window: Span, tz: tzinfo
) -> dict[str, dict[int, int]]:
    """`by_day_hour` with a zero row for every calendar day (in `tz`) the window touches.

    Days are keyed by ISO date like `summarize` keys them; existing rows are kept as-is
    and the result is sorted by day so the heatmap reads in calendar order.
    """
    filled = dict(by_day_hour)
    day = datetime.fromtimestamp(window[0], tz=tz).date()
    last = datetime.fromtimestamp(window[1], tz=tz).date()
    while day <= last:
        filled.setdefault(day.isoformat(), {h: 0 for h in range(24)})
        day += timedelta(days=1)
    return dict(sorted(filled.items()))


def _unmonitored_buckets(
    spans: list[Span], day_hour: dict[str, dict[int, int]], tz: tzinfo
) -> set[tuple[str, int]]:
    """Which (day_label, hour) heatmap cells the unmonitored ``spans`` cover.

    Only cells that appear in the heatmap (days that had events) and that hold zero
    events are marked, so a partly-monitored hour with a real event still shows its count.
    A span is expanded hour by hour; each touched hour whose cell is empty becomes
    'not monitored'.
    """
    buckets: set[tuple[str, int]] = set()
    for start, end in spans:
        # Walk each clock hour the span touches. Step by 3600 s from its start.
        t = start
        while t < end:
            dt = datetime.fromtimestamp(t, tz=tz)
            day = dt.date().isoformat()
            hour = dt.hour
            if day in day_hour and day_hour[day].get(hour, 0) == 0:
                buckets.add((day, hour))
            t += 3600.0
    return buckets


def _coverage_window(
    events: list[Event], gaps: list[Gap], sessions: list[Session]
) -> tuple[float, float] | None:
    """The observed reporting span as (start_unix, end_unix), or None if undeterminable.

    The span runs from the earliest observed moment to the latest across events, gaps,
    and every capture session — not just the most recent one, or a log whose monitor was
    restarted would report a window starting at whichever event happened to come first.
    This is the single definition of "the window" that every coverage figure is measured
    against, so the main report and the violations export cannot disagree about what they
    are covering.
    """
    starts: list[float] = [e.start for e in events]
    ends: list[float] = [e.end for e in events]
    for g in gaps:
        starts.append(g.start)
        ends.append(g.end)
    for s in sessions:
        starts.append(s.started_at)
        if s.ended_at is not None:
            ends.append(s.ended_at)
    if not starts or not ends:
        return None
    win_start, win_end = min(starts), max(ends)
    if win_end - win_start <= 0:
        return None
    return win_start, win_end


def _coverage_hours(
    events: list[Event], gaps: list[Gap], sessions: list[Session]
) -> tuple[float, float] | None:
    """Monitored vs wall-clock hours over the reporting window, or None if undeterminable.

    The window is `_coverage_window()`. Monitored time is the session ledger's on-air
    stretches (`on_air_spans`) minus any recorded gap inside them: time with no monitor
    running counts as unmonitored even though no gap row exists for it, which is the
    whole point — the device cannot write down an outage it was not running for.

    Only a log with no session rows at all falls back to treating the window as monitored
    apart from recorded gaps, because such a record genuinely cannot say more.
    """
    window = _coverage_window(events, gaps, sessions)
    if window is None:
        return None
    win_start, win_end = window
    span = win_end - win_start
    holes = _clip_spans(((g.start, g.end) for g in gaps), window)
    on_air = on_air_spans(events, sessions, window)
    if on_air is None:
        monitored = max(0.0, span - _span_seconds(holes))
    else:
        monitored = _span_seconds(_subtract_spans(on_air, holes))
    return monitored / 3600.0, span / 3600.0


def generate_report_from_db(
    db_path: str,
    config: Config,
    *,
    generated_at: str,
) -> str:
    """Read events from the store and render the report. Deterministic given inputs."""
    from store import EventStore

    anomalies: list[ClockAnomaly]
    with EventStore(db_path) as store:
        events = store.events()  # raw dBFS levels; calibration is applied here, at render
        history = store.calibration_history()
        session = store.latest_session()
        anomalies = store.clock_anomalies()
        gaps = store.gaps()
        sessions = store.sessions()
        minute_levels = store.minute_levels()  # EXP-01, opt-in; usually empty

    # Which epochs actually cover the events in this window? Only those drive the choice
    # between the single-offset path and the multi-epoch disclosure. Levels are always
    # adjusted through the shared per-event resolver, the same one the exports use.
    tz = config.tzinfo()
    epochs_in_window = _epochs_covering(history, events)
    offsets = _per_event_offsets(events, history, config)
    adjusted = [_apply_offset(ev, off) for ev, off in zip(events, offsets)]
    summary = summarize(adjusted, quiet_hours=config.quiet_hours, tz=tz)
    # Rows that predate the first calibration carry its offset retroactively (by
    # design); the report must say so, on every path, or "Calibrated." overclaims.
    back_applied = back_applied_summary(events, history, minute_levels)

    # Ambient baseline (EXP-01): same render-time calibration resolution as events, then
    # rolled up to one row per day. Empty when the ledger was never enabled.
    minute_offsets = _per_minute_offsets(minute_levels, history, config)
    adjusted_minutes = [
        _apply_offset_minute(m, off) for m, off in zip(minute_levels, minute_offsets)
    ]
    ambient_days = summarize_ambient(adjusted_minutes, tz=tz)

    # Monitoring-gap honesty: unmonitored heatmap buckets plus monitored-vs-wall-clock
    # coverage, rendered on both the single-offset and multi-epoch paths. "Unmonitored"
    # is recorded gaps *and* time with no monitor running (off_air_spans) — the heatmap's
    # hatched third state used to be reachable only from a device-error gap, so a night
    # the monitor spent switched off rendered as a row of quiet zeros.
    # Every calendar day the window covers gets a heatmap row, not only the days that
    # had events. Without this a quiet monitored day and a day the monitor was off
    # both simply vanished from the calendar — indistinguishable from each other and
    # from the days that were never in the window. With the row present, a quiet day
    # shows its zeros and an off-air day is hatched "not monitored" below.
    window = _coverage_window(events, gaps, sessions)
    if window is not None:
        summary = dataclasses.replace(
            summary, by_day_hour=_fill_window_days(summary.by_day_hour, window, tz)
        )
    unmonitored_spans = [(g.start, g.end) for g in gaps] + (
        off_air_spans(events, gaps, sessions) or []
    )
    unmonitored = (
        _unmonitored_buckets(_merge_spans(unmonitored_spans), summary.by_day_hour, tz)
        if unmonitored_spans
        else None
    )
    coverage = _coverage_hours(events, gaps, sessions)
    monitored_hours, wall_clock_hours = coverage if coverage else (None, None)

    if len(epochs_in_window) > 1:
        latest = epochs_in_window[-1]
        return build_report(
            summary,
            config=config,
            generated_at=generated_at,
            calibration_offset=latest.offset,
            calibration_note=latest.note or config.calibration_note,
            calibration_epochs=epochs_in_window,
            session=session,
            sessions=sessions,
            unmonitored=unmonitored,
            monitored_hours=monitored_hours,
            wall_clock_hours=wall_clock_hours,
            clock_anomaly_lines=describe_clock_anomalies(anomalies, tz=tz),
            ambient_days=ambient_days,
            back_applied=back_applied,
        )

    # Single-offset path: the whole window is under one calibration (or none -> config).
    if epochs_in_window:
        epoch = epochs_in_window[0]
        offset, note = epoch.offset, epoch.note or config.calibration_note
    elif history:
        offset, note = history[-1].offset, history[-1].note or config.calibration_note
    else:
        offset, note = config.calibration_offset, config.calibration_note
    return build_report(
        summary,
        config=config,
        generated_at=generated_at,
        calibration_offset=offset,
        calibration_note=note,
        session=session,
        sessions=sessions,
        unmonitored=unmonitored,
        monitored_hours=monitored_hours,
        wall_clock_hours=wall_clock_hours,
        clock_anomaly_lines=describe_clock_anomalies(anomalies, tz=tz),
        ambient_days=ambient_days,
        back_applied=back_applied,
    )


def _write_tagged_pdf_cli(html: str, path: Path, *, violation: bool = False) -> bool:
    """Write one optional PDF, returning False after an actionable user-facing error."""
    from report.pdf_export import (
        PdfExportUnavailable,
        TaggedPdfGenerationError,
        write_tagged_pdf,
    )

    try:
        n = write_tagged_pdf(html, path)
    except (PdfExportUnavailable, TaggedPdfGenerationError) as exc:
        print(f"Skipped {path}: {exc}")
        return False
    suffix = (
        "structural tags only; not a verified PDF/UA conformance claim"
        if violation
        else "structural tags only; not a verified PDF/UA conformance claim, see "
        "docs/adr/0004-weasyprint-for-tagged-pdf-a-export.md"
    )
    print(f"Wrote {path} ({n} bytes, tagged PDF/A-3a -- {suffix}).")
    return True


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="olive-report",
        description="Generate an accessible HTML noise report from the event log.",
    )
    parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
    parser.add_argument("--config", type=Path, default=None, help="path to JSON config")
    parser.add_argument("--db", type=str, default=None, help="path to SQLite event log")
    parser.add_argument("--out", type=Path, default=Path("report.html"), help="output HTML path")
    parser.add_argument(
        "--generated-at",
        type=str,
        default=None,
        help="timestamp string for the report header (default: now, UTC)",
    )
    parser.add_argument(
        "--csv", type=Path, default=None, help="also export the event log to this CSV path"
    )
    parser.add_argument(
        "--violations-csv",
        type=Path,
        default=None,
        help="export every event flagged within/outside quiet hours to this CSV path",
    )
    parser.add_argument(
        "--violations-html",
        type=Path,
        default=None,
        help="render a standalone honest quiet-hours violation report to this HTML path",
    )
    parser.add_argument(
        "--pdf",
        type=Path,
        default=None,
        help=(
            "also export a tagged PDF/A-3a of the main report to this path (needs the "
            "'pdf' extra: pip install -e '.[pdf]'; see "
            "docs/adr/0004-weasyprint-for-tagged-pdf-a-export.md -- structural tags "
            "only, not a verified PDF/UA conformance claim)"
        ),
    )
    parser.add_argument(
        "--violations-pdf",
        type=Path,
        default=None,
        help="also export a tagged PDF/A-3a of the quiet-hours violation report to this path",
    )
    args = parser.parse_args(argv)

    config = Config.load(args.config)
    db_path = args.db or config.db_path
    generated_at = args.generated_at or datetime.now(config.tzinfo()).strftime("%Y-%m-%d %H:%M %Z")

    html = generate_report_from_db(db_path, config, generated_at=generated_at)
    args.out.write_text(html, encoding="utf-8")
    print(f"Wrote {args.out} ({len(html)} bytes).")

    if args.pdf is not None and not _write_tagged_pdf_cli(html, args.pdf):
        return 1

    if (
        args.csv is not None
        or args.violations_csv is not None
        or args.violations_html is not None
        or args.violations_pdf is not None
    ):
        from store import EventStore

        with EventStore(db_path) as store:
            raw_events = store.events()  # raw dBFS; calibration is applied below, at render
            history = store.calibration_history()
            gaps = store.gaps()
            # The exports state their own monitoring coverage, computed from the same
            # (events, gaps, sessions) triple the main report uses — so the handed-over
            # document cannot claim a window the device did not observe. The full session
            # list, not just the latest: the stretches between runs are exactly the time
            # no monitor was listening, and no gap row exists for them.
            export_sessions = store.sessions()
        # Exports must agree numerically with the main report: the same render-time,
        # per-epoch calibration is applied to every exported artifact, and each CSV row
        # records the offset it received (raw = value - offset). The calibrated flag is
        # derived from the store's history, never from the deprecated config field.
        offsets = _per_event_offsets(raw_events, history, config)
        # Parallel to `offsets`: whether each row's offset was in force when it was
        # measured or back-applied from a later calibration. Every export carries it per
        # row, and the violations HTML states the count, so a row the calibration did
        # not vouch for cannot pass as one it did.
        basis = _per_event_basis(raw_events, history, config)
        back_applied = back_applied_summary(raw_events, history)
        back_applied_note = (
            back_applied_sentence(back_applied, tz=config.tzinfo())
            if back_applied is not None
            else None
        )
        events = [_apply_offset(ev, off) for ev, off in zip(raw_events, offsets)]
        multi_epoch = len(set(offsets)) > 1
        if events:
            calibrated = all(off != 0.0 for off in offsets)
        else:  # nothing to adjust; disclose the calibration in force (store, then config)
            calibrated = (history[-1].offset if history else config.calibration_offset) != 0.0

        if args.csv is not None:
            from report.export import events_to_csv

            rows = events_to_csv(
                events,
                args.csv,
                tz=config.tzinfo(),
                offsets_db=offsets,
                basis=basis,
                gaps=gaps,
            )
            print(f"Wrote {args.csv} ({rows} rows).")

        if (
            args.violations_csv is not None
            or args.violations_html is not None
            or args.violations_pdf is not None
        ):
            from report.violations import (
                build_violation_report_html,
                compute_violations,
                violations_to_csv,
            )

            if args.violations_csv is not None:
                rows = violations_to_csv(
                    events,
                    args.violations_csv,
                    quiet_hours=config.quiet_hours,
                    tz=config.tzinfo(),
                    tz_name=config.tz,
                    offsets_db=offsets,
                    basis=basis,
                    gaps=gaps,
                    sessions=export_sessions,
                )
                print(f"Wrote {args.violations_csv} ({rows} rows).")
            if args.violations_html is not None or args.violations_pdf is not None:
                report = compute_violations(
                    events,
                    quiet_hours=config.quiet_hours,
                    tz=config.tzinfo(),
                    tz_name=config.tz,
                    offsets_db=offsets,
                    basis=basis,
                    gaps=gaps,
                    sessions=export_sessions,
                )
                vhtml = build_violation_report_html(
                    report,
                    threshold_dbfs=config.threshold_dbfs,
                    min_duration_s=config.min_duration_s,
                    generated_at=generated_at,
                    calibrated=calibrated,
                    multi_epoch=multi_epoch,
                    back_applied_note=back_applied_note,
                )
                if args.violations_html is not None:
                    args.violations_html.write_text(vhtml, encoding="utf-8")
                    print(f"Wrote {args.violations_html} ({len(vhtml)} bytes).")
                if args.violations_pdf is not None and not _write_tagged_pdf_cli(
                    vhtml, args.violations_pdf, violation=True
                ):
                    return 1
    return 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())