"""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('