205 lines
7.3 KiB
Python
205 lines
7.3 KiB
Python
"""Where scan results live between runs.
|
|
|
|
A directory rather than a single file, because the three kinds of content cost wildly
|
|
different amounts to produce::
|
|
|
|
<cache_dir>/
|
|
├── index.json cheap: tags and structure. Thrown away freely.
|
|
├── covers/<album_id>.jpg medium: art pulled out of an ID3 APIC frame
|
|
└── analysis/<track_key>.json expensive: minutes of DSP per track
|
|
analysis/<track_key>.beats.json
|
|
|
|
That split is the whole point. A rescan must be free to rebuild ``index.json`` without
|
|
destroying analysis, so everything expensive is keyed by a *content* key (see
|
|
:func:`~musicmouse.library.models.track_key`) rather than by album id - renaming a
|
|
folder or re-sorting a section then costs nothing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
from musicmouse.library.analysis import BeatGrid, TrackAnalysis
|
|
from musicmouse.library.models import Album, LibraryTrack
|
|
from musicmouse.library.sections import SECTIONS, AlbumKind
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
__all__ = ["Fingerprint", "LibraryCache"]
|
|
|
|
#: Bump whenever the scanner's *derivation* changes - how a title, artist or series is
|
|
#: worked out - not just when the JSON shape does. A cached entry is reused whenever its
|
|
#: files are untouched, so otherwise a change to that logic is invisible until somebody
|
|
#: edits their music folder.
|
|
_INDEX_VERSION = 4
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Fingerprint:
|
|
"""What makes a folder's cached entry still valid: its files, sizes and mtimes."""
|
|
|
|
files: tuple[tuple[str, int, int], ...]
|
|
|
|
@classmethod
|
|
def of(cls, paths: list[Path]) -> Fingerprint:
|
|
entries: list[tuple[str, int, int]] = []
|
|
for path in paths:
|
|
stat = path.stat()
|
|
entries.append((path.name, stat.st_size, int(stat.st_mtime)))
|
|
return cls(tuple(entries))
|
|
|
|
def to_json(self) -> list[list[Any]]:
|
|
return [list(entry) for entry in self.files]
|
|
|
|
@classmethod
|
|
def from_json(cls, data: list[list[Any]]) -> Fingerprint:
|
|
return cls(tuple((str(n), int(s), int(m)) for n, s, m in data))
|
|
|
|
|
|
def _write_atomic(path: Path, payload: str) -> None:
|
|
"""Write via a sibling temp file so a crash never leaves a half-written cache."""
|
|
temp = path.with_name(f"{path.name}.tmp{os.getpid()}")
|
|
temp.write_text(payload, encoding="utf-8")
|
|
temp.replace(path)
|
|
|
|
|
|
class LibraryCache:
|
|
def __init__(self, directory: Path) -> None:
|
|
self.directory = directory
|
|
self.covers = directory / "covers"
|
|
self.analysis = directory / "analysis"
|
|
|
|
def prepare(self) -> None:
|
|
for folder in (self.directory, self.covers, self.analysis):
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
# -------------------------------------------------------------------- covers
|
|
|
|
def cover_path(self, album_id: str) -> Path:
|
|
return self.covers / f"{album_id}.jpg"
|
|
|
|
def store_cover(self, album_id: str, data: bytes) -> Path:
|
|
path = self.cover_path(album_id)
|
|
path.write_bytes(data)
|
|
return path
|
|
|
|
# ------------------------------------------------------------------ analysis
|
|
|
|
def load_analysis(self, key: str) -> TrackAnalysis | None:
|
|
path = self.analysis / f"{key}.json"
|
|
try:
|
|
return TrackAnalysis.from_json(json.loads(path.read_text(encoding="utf-8")))
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
def store_analysis(self, key: str, analysis: TrackAnalysis) -> None:
|
|
_write_atomic(self.analysis / f"{key}.json", json.dumps(analysis.to_json()))
|
|
|
|
def load_beats(self, key: str) -> BeatGrid | None:
|
|
path = self.analysis / f"{key}.beats.json"
|
|
try:
|
|
return BeatGrid.from_json(json.loads(path.read_text(encoding="utf-8")))
|
|
except (OSError, ValueError, KeyError):
|
|
return None
|
|
|
|
def store_beats(self, key: str, grid: BeatGrid) -> None:
|
|
_write_atomic(self.analysis / f"{key}.beats.json", json.dumps(grid.to_json()))
|
|
|
|
# --------------------------------------------------------------------- index
|
|
|
|
def load_index(self) -> dict[str, tuple[Album, Fingerprint]]:
|
|
path = self.directory / "index.json"
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
if raw.get("version") != _INDEX_VERSION:
|
|
_log.info("Library index is from an older version; rescanning from scratch")
|
|
return {}
|
|
|
|
out: dict[str, tuple[Album, Fingerprint]] = {}
|
|
for entry in raw.get("albums", []):
|
|
try:
|
|
out[entry["id"]] = (_album_from_json(entry), Fingerprint.from_json(entry["files"]))
|
|
except (KeyError, TypeError, ValueError):
|
|
_log.debug("Dropping unreadable index entry %r", entry.get("id"))
|
|
return out
|
|
|
|
def store_index(self, albums: dict[str, tuple[Album, Fingerprint]]) -> None:
|
|
payload = {
|
|
"version": _INDEX_VERSION,
|
|
"albums": [
|
|
_album_to_json(album) | {"files": fingerprint.to_json()}
|
|
for album, fingerprint in albums.values()
|
|
],
|
|
}
|
|
self.prepare()
|
|
_write_atomic(self.directory / "index.json", json.dumps(payload, ensure_ascii=False))
|
|
|
|
|
|
# ------------------------------------------------------------------ serialisation
|
|
|
|
|
|
def _album_to_json(album: Album) -> dict[str, Any]:
|
|
return {
|
|
"id": album.id,
|
|
"section": album.section,
|
|
"kind": album.kind,
|
|
"title": album.title,
|
|
"artist": album.artist,
|
|
"series": album.series,
|
|
"figure": album.figure,
|
|
"colors": list(album.colors),
|
|
"folder": str(album.folder),
|
|
"cover": str(album.cover) if album.cover else None,
|
|
"tracks": [
|
|
{
|
|
"path": str(track.path),
|
|
"title": track.title,
|
|
"duration": track.duration,
|
|
"analysis": track.analysis.to_json() if track.analysis else None,
|
|
}
|
|
for track in album.tracks
|
|
],
|
|
}
|
|
|
|
|
|
def _album_from_json(data: dict[str, Any]) -> Album:
|
|
section = str(data["section"])
|
|
if section not in SECTIONS:
|
|
raise ValueError(f"unknown section {section!r}")
|
|
# Read back what was stored rather than re-deriving it from the section: a figure's
|
|
# kind comes from the config, so the section cannot answer for it.
|
|
kind = str(data["kind"])
|
|
if kind not in ("music", "book"):
|
|
raise ValueError(f"unknown kind {kind!r}")
|
|
red, green, blue = data["colors"]
|
|
return Album(
|
|
id=str(data["id"]),
|
|
section=section,
|
|
kind=cast("AlbumKind", kind),
|
|
title=str(data["title"]),
|
|
artist=str(data["artist"]),
|
|
series=data["series"],
|
|
figure=data["figure"],
|
|
colors=(str(red), str(green), str(blue)),
|
|
folder=Path(data["folder"]),
|
|
cover=Path(data["cover"]) if data["cover"] else None,
|
|
tracks=tuple(
|
|
LibraryTrack(
|
|
path=Path(track["path"]),
|
|
title=str(track["title"]),
|
|
duration=float(track["duration"]),
|
|
analysis=(
|
|
TrackAnalysis.from_json(track["analysis"]) if track.get("analysis") else None
|
|
),
|
|
)
|
|
for track in data["tracks"]
|
|
),
|
|
)
|