mrleo1nidandClaude Opus 5.5
ef29b85320
- migration 00006: track features and embeddings (pgvector, HNSW), analysis queue filled by a trigger on new tracks, wave sessions and impressions, recommender profiles, model versions - analyzer basic-1 (ADR-014): ffmpeg decode, three 30 s excerpts, 256-d timbre/rhythm embedding, tempo by onset autocorrelation, key, loudness, energy, danceability, acousticness, intro length; worker with SKIP LOCKED claims, retries and re-analysis on version change - wave engine v1: candidates (session/taste similarity, fresh, liked artists, rediscover, popular, exploration, seed), filters (dislikes, played in 3 h, already served), heuristic score, activity and mood fit from contexts.yaml with relaxing thresholds, MMR with one track per artist, exploration share, session vector updated by reactions and a change of course after three skips - POST /internal/wave/batch with the catalog index cached in memory and the session kept in Redis Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Профили занятий и настроений: соответствие трека контексту (context_fit)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from importlib import resources
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import yaml
|
|
from numpy.typing import NDArray
|
|
from pydantic import BaseModel, model_validator
|
|
|
|
FEATURES = ("bpm", "energy", "danceability", "valence", "acousticness")
|
|
|
|
|
|
class Rule(BaseModel):
|
|
feature: str
|
|
min: float | None = None
|
|
max: float | None = None
|
|
sigma: float
|
|
half_time: bool = False
|
|
|
|
@model_validator(mode="after")
|
|
def check(self) -> Rule:
|
|
if self.feature not in FEATURES:
|
|
raise ValueError(f"unknown feature {self.feature!r}; known: {', '.join(FEATURES)}")
|
|
if self.min is None and self.max is None:
|
|
raise ValueError("rule needs min or max")
|
|
if self.sigma <= 0:
|
|
raise ValueError("sigma must be positive")
|
|
return self
|
|
|
|
|
|
class Context(BaseModel):
|
|
icon: str
|
|
rules: list[Rule]
|
|
|
|
|
|
class Contexts(BaseModel):
|
|
activities: dict[str, Context]
|
|
moods: dict[str, Context]
|
|
|
|
|
|
def load_contexts(path: str = "") -> Contexts:
|
|
"""contexts.yaml из пути или встроенный; ошибка в файле — ошибка запуска (Pydantic)."""
|
|
raw = (
|
|
Path(path).read_text("utf-8")
|
|
if path
|
|
else resources.files(__package__).joinpath("contexts.yaml").read_text("utf-8")
|
|
)
|
|
return Contexts.model_validate(yaml.safe_load(raw))
|
|
|
|
|
|
def _bell(x: NDArray[np.float64], rule: Rule) -> NDArray[np.float64]:
|
|
lo = -np.inf if rule.min is None else rule.min
|
|
hi = np.inf if rule.max is None else rule.max
|
|
dist = np.where(x < lo, lo - x, np.where(x > hi, x - hi, 0.0))
|
|
return np.asarray(np.exp(-0.5 * (dist / rule.sigma) ** 2))
|
|
|
|
|
|
def fit(features: dict[str, NDArray[np.float64]], rules: list[Rule]) -> NDArray[np.float64]:
|
|
"""Произведение «колоколов» по правилам; NaN (признака нет) — нейтрально (1).
|
|
half_time: для темпа засчитывается и половинный (бег под 85 BPM — тот же шаг, что 170)."""
|
|
n = len(next(iter(features.values())))
|
|
out = np.ones(n)
|
|
for rule in rules:
|
|
x = features[rule.feature]
|
|
known = ~np.isnan(x)
|
|
b = _bell(np.nan_to_num(x), rule)
|
|
if rule.half_time:
|
|
b = np.maximum(b, _bell(np.nan_to_num(x) * 2, rule))
|
|
out *= np.where(known, b, 1.0)
|
|
return out
|