ISR/isrweb/audiofile.py

"""What an audio file is, and what its name means.

Two clusters that share a theme — facts about the files on disk — and share
the property of being touched only when a container format misbehaves:

* the filename clock (recordings are named for their *start* time, and that
  name is the only reliable time anchor the archive has), and
* byte-level format handling: the 44-byte PCM header for clips, duration
  probing, and the live-header patching that lets a browser show a duration
  for a file still being recorded.

Imports nothing from the rest of the app.
"""

import struct
import wave
from datetime import datetime, timedelta
from pathlib import Path

try:
    import soundfile as sf
    SOUNDFILE_AVAILABLE = True
except ImportError:
    SOUNDFILE_AVAILABLE = False

AUDIO_EXTENSIONS = {'.wav', '.mp3', '.ogg', '.flac', '.aac', '.opus'}

# Recording filenames encode the start time as this strftime format (kept in
# sync with isr.FILENAME_FORMAT). It is the authoritative recording start and
# the only reliable clock anchor — mtime drifts to the last write, so cut clip
# names and the displayed date are both derived from this.
FILENAME_FORMAT = '%Y%m%d_%H%M%S'

MIME_TYPES = {
    '.wav':  'audio/wav',
    '.mp3':  'audio/mpeg',
    '.ogg':  'audio/ogg',
    '.flac': 'audio/flac',
    '.aac':  'audio/aac',
    '.opus': 'audio/ogg',
}

# ---------------------------------------------------------------------------
# The filename clock
# ---------------------------------------------------------------------------

def _recording_start(stem: str):
    """Parse the recording start time encoded in a filename stem.

    Returns a datetime, or None if the stem is not in FILENAME_FORMAT
    (e.g. a manually renamed file). strptime ignores any extension because
    callers pass Path.stem.
    """
    try:
        return datetime.strptime(stem, FILENAME_FORMAT)
    except ValueError:
        return None


def _cut_filename(stem: str, ext: str, start: float, end: float) -> str:
    """Name a cut by the real wall-clock span it covers.

    For a recording that started at 22:00:00, a 22:31:30→22:32:30 slice
    (start=1890, end=1950) becomes ``20260523_22-31-30_22-32-30.flac``.
    Falls back to the source stem plus second offsets when the filename is
    not in FILENAME_FORMAT (e.g. a manually renamed recording).
    """
    started = _recording_start(stem)
    if started is None:
        return f'{stem}_cut_{int(start)}s-{int(end)}s{ext}'
    cut_start = started + timedelta(seconds=start)
    cut_end   = started + timedelta(seconds=end)
    return f'{cut_start:%Y%m%d}_{cut_start:%H-%M-%S}_{cut_end:%H-%M-%S}{ext}'


def _live_wav_header(path: Path, size: int):
    """Return the WAV header (through the 'data' chunk header) with RIFF and
    data sizes rewritten to match the current file size, or None.

    While a WAV file is still being recorded its header claims ~0 frames, so
    browsers show no duration and refuse to seek. Serving a header patched to
    the bytes recorded so far fixes both; the patch is the same length as the
    original header, so all byte offsets and Range math stay valid.
    """
    try:
        with open(path, 'rb') as fh:
            hdr = fh.read(512)
        if len(hdr) < 44 or hdr[:4] != b'RIFF' or hdr[8:12] != b'WAVE':
            return None
        pos = 12
        while pos + 8 <= len(hdr):
            chunk_id   = hdr[pos:pos + 4]
            chunk_size = int.from_bytes(hdr[pos + 4:pos + 8], 'little')
            if chunk_id == b'data':
                data_off = pos + 8
                patched = bytearray(hdr[:data_off])
                patched[4:8] = (size - 8).to_bytes(4, 'little')
                patched[pos + 4:pos + 8] = (size - data_off).to_bytes(4, 'little')
                return bytes(patched)
            pos += 8 + chunk_size + (chunk_size & 1)
        return None
    except Exception:
        return None


# CRC-8 (poly 0x07) used by FLAC frame headers
_CRC8_TABLE = []
for _i in range(256):
    _c = _i
    for _ in range(8):
        _c = ((_c << 1) ^ 0x07) & 0xFF if _c & 0x80 else (_c << 1) & 0xFF
    _CRC8_TABLE.append(_c)

_FLAC_BLOCKSIZES = {1: 192, 2: 576, 3: 1152, 4: 2304, 5: 4608, 8: 256, 9: 512,
                    10: 1024, 11: 2048, 12: 4096, 13: 8192, 14: 16384, 15: 32768}


def _crc8(data: bytes) -> int:
    crc = 0
    for b in data:
        crc = _CRC8_TABLE[crc ^ b]
    return crc


def _flac_coded_number(buf: bytes, pos: int):
    """Decode the UTF-8-style frame/sample number; returns (value, next_pos)."""
    b0 = buf[pos]
    if b0 < 0x80:
        return b0, pos + 1
    n, mask = 0, 0x40
    while b0 & mask:
        n += 1
        mask >>= 1
    if n < 1 or n > 6:  # 10xxxxxx is not a valid leading byte
        return None
    val = b0 & (mask - 1)
    for i in range(1, n + 1):
        c = buf[pos + i]
        if c & 0xC0 != 0x80:
            return None
        val = (val << 6) | (c & 0x3F)
    return val, pos + 1 + n


def _flac_frame_samples(buf: bytes, pos: int, fixed_bs: int):
    """If a valid FLAC frame header starts at pos, return the stream sample
    count through the end of that frame, else None. Validity is confirmed by
    the header's CRC-8, so false sync matches in compressed data are rejected."""
    try:
        variable = buf[pos + 1] & 0x01
        bs_code  = buf[pos + 2] >> 4
        sr_code  = buf[pos + 2] & 0x0F
        if bs_code == 0 or sr_code == 15 or buf[pos + 3] & 0x01:
            return None
        if (buf[pos + 3] >> 4) > 10:  # reserved channel assignment
            return None
        coded = _flac_coded_number(buf, pos + 4)
        if coded is None:
            return None
        val, p = coded
        bs = _FLAC_BLOCKSIZES.get(bs_code)
        if bs_code == 6:
            bs = buf[p] + 1
            p += 1
        elif bs_code == 7:
            bs = int.from_bytes(buf[p:p + 2], 'big') + 1
            p += 2
        if sr_code == 12:
            p += 1
        elif sr_code in (13, 14):
            p += 2
        if _crc8(buf[pos:p]) != buf[p]:
            return None
        if variable:           # val is the frame's starting sample number
            return val + (bs or 0)
        return val * (fixed_bs or bs or 4096) + (bs or 0)
    except IndexError:
        return None


def _live_flac_header(path: Path, size: int):
    """Return the first 26 bytes of a FLAC file with STREAMINFO total_samples
    patched to the samples recorded so far, or None.

    Like _live_wav_header, but FLAC duration cannot be derived from the byte
    count (variable compression). Instead the sample count is parsed out of
    the last frame header in the file tail — each FLAC frame carries its own
    frame/sample number.
    """
    try:
        with open(path, 'rb') as fh:
            head = fh.read(42)
            if len(head) < 42 or head[:4] != b'fLaC':
                return None
            # STREAMINFO must be the first metadata block
            if head[4] & 0x7F != 0 or int.from_bytes(head[5:8], 'big') != 34:
                return None
            fixed_bs = int.from_bytes(head[8:10], 'big')

            tail_len = min(size, 65536)
            fh.seek(size - tail_len)
            buf = fh.read(tail_len)

        samples = None
        for i in range(len(buf) - 20, -1, -1):
            if buf[i] == 0xFF and (buf[i + 1] & 0xFC) == 0xF8:
                samples = _flac_frame_samples(buf, i, fixed_bs)
                if samples:
                    break
        if not samples:
            return None

        # Bytes 18-25 hold: sample rate (20 bits) | channels-1 (3) |
        # bps-1 (5) | total_samples (36). Replace only the low 36 bits.
        field = int.from_bytes(head[18:26], 'big')
        field = (field & ~((1 << 36) - 1)) | min(samples, (1 << 36) - 1)
        patched = bytearray(head[:26])
        patched[18:26] = field.to_bytes(8, 'big')
        return bytes(patched)
    except Exception:
        return None


def _wav_header(n_frames: int, channels: int, framerate: int, sampwidth: int) -> bytes:
    """Standard 44-byte PCM WAV header for a clip of known length."""
    data_len  = n_frames * channels * sampwidth
    byte_rate = framerate * channels * sampwidth
    return (b'RIFF' + (36 + data_len).to_bytes(4, 'little') + b'WAVE'
            + b'fmt ' + (16).to_bytes(4, 'little')
            + struct.pack('<HHIIHH', 1, channels, framerate, byte_rate,
                          channels * sampwidth, sampwidth * 8)
            + b'data' + data_len.to_bytes(4, 'little'))


def _get_audio_duration(path: Path):
    """Return duration in seconds for any supported audio file, or None."""
    ext = path.suffix.lower()
    if ext == '.wav':
        try:
            with wave.open(str(path), 'rb') as wf:
                return round(wf.getnframes() / wf.getframerate(), 2)
        except Exception:
            return None
    if SOUNDFILE_AVAILABLE and ext in ('.flac', '.ogg', '.opus'):
        try:
            with sf.SoundFile(path) as f:
                return round(len(f) / f.samplerate, 2)
        except Exception:
            return None
    return None