"""SD MP3 playback speed: clamp [0.5, 2.0], default 1.0. Gradients live on Server."""
|
|
from __future__ import annotations
|
|
import json
|
import os
|
import re
|
import time
|
from typing import Any, Dict, List, Optional
|
|
from config.config_loader import get_project_dir
|
from config.logger import setup_logging
|
|
TAG = __name__
|
logger = setup_logging()
|
|
PLAY_SPEED_MIN = 0.5
|
PLAY_SPEED_MAX = 2.0
|
PLAY_SPEED_DEFAULT = 1.0
|
|
# 小程序/定时任务下发清单之前的临时课文。路径相对 SD 卡根（/english_learning/...）。
|
TEMP_REPEATER_PLAYLIST = (
|
{
|
"name": "/english_learning/shadowing/lessons/unit01/01.mp3",
|
"repeat": 10,
|
"speed": PLAY_SPEED_DEFAULT,
|
},
|
{
|
"name": "/english_learning/shadowing/lessons/unit01/02.mp3",
|
"repeat": 5,
|
"speed": PLAY_SPEED_DEFAULT,
|
},
|
)
|
|
TEMP_SHADOWING_PLAYLIST = (
|
{
|
"name": "/english_learning/shadowing/lessons/unit01/01.mp3",
|
"repeat": 5,
|
"speed": PLAY_SPEED_DEFAULT,
|
},
|
{
|
"name": "/english_learning/shadowing/lessons/unit01/02.mp3",
|
"repeat": 10,
|
"speed": PLAY_SPEED_DEFAULT,
|
},
|
)
|
|
|
def clamp_play_speed(raw: Any) -> float:
|
try:
|
value = float(raw)
|
except (TypeError, ValueError):
|
return PLAY_SPEED_DEFAULT
|
if value != value: # NaN
|
return PLAY_SPEED_DEFAULT
|
if value < PLAY_SPEED_MIN:
|
return PLAY_SPEED_MIN
|
if value > PLAY_SPEED_MAX:
|
return PLAY_SPEED_MAX
|
return value
|
|
|
def playback_config(task: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
cfg = (task or {}).get("playback")
|
return cfg if isinstance(cfg, dict) else {}
|
|
|
def _progress_path(device_id: str) -> str:
|
safe = re.sub(r"[^A-Za-z0-9_-]+", "_", device_id or "unknown") or "unknown"
|
return os.path.join(get_project_dir(), "data", "playback_progress", f"{safe}.json")
|
|
|
def _load_progress(device_id: str) -> Dict[str, Any]:
|
path = _progress_path(device_id)
|
if not os.path.isfile(path):
|
return {}
|
try:
|
with open(path, "r", encoding="utf-8") as handle:
|
data = json.load(handle)
|
return data if isinstance(data, dict) else {}
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"读取播放进度失败: {exc}")
|
return {}
|
|
|
def _save_progress(device_id: str, data: Dict[str, Any]) -> None:
|
path = _progress_path(device_id)
|
try:
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
with open(path, "w", encoding="utf-8") as handle:
|
json.dump(data, handle, ensure_ascii=False, indent=2)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"保存播放进度失败: {exc}")
|
|
|
def _lesson_key(app_id: str, task_id: str) -> str:
|
return f"{app_id or ''}/{task_id or ''}"
|
|
|
def lesson_day_index(
|
device_id: str,
|
app_id: str,
|
task_id: str,
|
lesson: str = "",
|
) -> int:
|
"""0-based count of distinct calendar days this lesson has already been played."""
|
data = _load_progress(device_id)
|
bucket = data.get(_lesson_key(app_id, task_id)) or {}
|
lessons = bucket.get("lessons") if isinstance(bucket, dict) else {}
|
key = lesson or "_default"
|
days = lessons.get(key) if isinstance(lessons, dict) else None
|
if not isinstance(days, list):
|
return 0
|
return len([d for d in days if d])
|
|
|
def mark_lesson_played_today(
|
device_id: str,
|
app_id: str,
|
task_id: str,
|
lesson: str = "",
|
) -> None:
|
today = time.strftime("%Y-%m-%d")
|
data = _load_progress(device_id)
|
key = _lesson_key(app_id, task_id)
|
bucket = data.get(key)
|
if not isinstance(bucket, dict):
|
bucket = {}
|
lessons = bucket.get("lessons")
|
if not isinstance(lessons, dict):
|
lessons = {}
|
lesson_key = lesson or "_default"
|
days = lessons.get(lesson_key)
|
if not isinstance(days, list):
|
days = []
|
if today not in days:
|
days.append(today)
|
lessons[lesson_key] = days
|
bucket["lessons"] = lessons
|
data[key] = bucket
|
_save_progress(device_id, data)
|
|
|
def daily_base_speed(task: Optional[Dict[str, Any]], day_index: int = 0) -> float:
|
cfg = playback_config(task)
|
daily = cfg.get("daily_speeds")
|
if isinstance(daily, list) and daily:
|
idx = max(0, int(day_index or 0))
|
if idx >= len(daily):
|
idx = len(daily) - 1
|
return clamp_play_speed(daily[idx])
|
if cfg.get("speed") is not None:
|
return clamp_play_speed(cfg.get("speed"))
|
return PLAY_SPEED_DEFAULT
|
|
|
def speeds_for_repeats(
|
task: Optional[Dict[str, Any]],
|
repeat_n: int = 1,
|
day_index: int = 0,
|
) -> List[float]:
|
"""One speed per pass. repeat=1 uses daily curve; repeat>1 samples repeat_speeds if set."""
|
n = 1
|
try:
|
n = int(repeat_n or 1)
|
except (TypeError, ValueError):
|
n = 1
|
if n < 1:
|
n = 1
|
base = daily_base_speed(task, day_index)
|
curve = playback_config(task).get("repeat_speeds")
|
if n == 1 or not isinstance(curve, list) or not curve:
|
return [base] * n
|
values = [clamp_play_speed(item) for item in curve]
|
if n == len(values):
|
return values
|
if n == 1:
|
return [values[0]]
|
out: List[float] = []
|
last = len(values) - 1
|
for i in range(n):
|
pos = (i / (n - 1)) * last
|
lo = int(pos)
|
hi = min(lo + 1, last)
|
frac = pos - lo
|
out.append(clamp_play_speed(values[lo] * (1.0 - frac) + values[hi] * frac))
|
return out
|
|
|
def resolve_play_speed(
|
task: Optional[Dict[str, Any]],
|
day_index: int = 0,
|
explicit: Any = None,
|
) -> float:
|
if explicit is not None:
|
return clamp_play_speed(explicit)
|
return daily_base_speed(task, day_index)
|
|
|
def normalize_media_name(raw: Any) -> str:
|
"""SD files[].name, MQTT path, or HTTP url → /app/task/lessons/... form."""
|
name = str(raw or "").strip().replace("\\", "/")
|
if not name:
|
return ""
|
for marker in ("/english_learning/", "english_learning/"):
|
idx = name.find(marker)
|
if idx >= 0:
|
rest = name[idx:]
|
return rest if rest.startswith("/") else "/" + rest
|
if "://" in name:
|
path = name.split("://", 1)[1]
|
_, _, rest = path.partition("/")
|
if rest:
|
return "/" + rest.lstrip("/")
|
return ""
|
if not name.startswith("/"):
|
name = "/" + name.lstrip("/")
|
return name
|
|
|
def _item_repeat(raw: Dict[str, Any]) -> int:
|
for key in ("repeat", "count", "play_count"):
|
if raw.get(key) is None:
|
continue
|
try:
|
value = int(raw.get(key))
|
except (TypeError, ValueError):
|
continue
|
return max(1, value)
|
return 1
|
|
|
def playlist_item_from_value(raw: Any) -> Optional[Dict[str, Any]]:
|
if isinstance(raw, str):
|
name = normalize_media_name(raw)
|
if not name:
|
return None
|
return {"name": name, "repeat": 1}
|
if not isinstance(raw, dict):
|
return None
|
name = normalize_media_name(
|
raw.get("name")
|
or raw.get("path")
|
or raw.get("target_path")
|
or raw.get("url")
|
or ""
|
)
|
if not name:
|
return None
|
item: Dict[str, Any] = {"name": name, "repeat": _item_repeat(raw)}
|
if raw.get("speed") is not None:
|
item["speed"] = clamp_play_speed(raw.get("speed"))
|
if raw.get("power_off_after") is not None:
|
item["power_off_after"] = bool(raw.get("power_off_after"))
|
return item
|
|
|
def parse_playlist(params: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
"""Repeater/shadowing playlist. Each item is {name, repeat, optional speed}."""
|
params = params or {}
|
raw_items = params.get("items") or params.get("playlist") or params.get("clips")
|
if isinstance(raw_items, list) and raw_items:
|
items = []
|
for raw in raw_items:
|
item = playlist_item_from_value(raw)
|
if item:
|
items.append(item)
|
return items
|
item = playlist_item_from_value(
|
{
|
"name": params.get("name") or params.get("path") or params.get("url") or "",
|
"repeat": params.get("repeat") or params.get("count") or 1,
|
"speed": params.get("speed"),
|
"power_off_after": params.get("power_off_after"),
|
}
|
)
|
if item and item["name"]:
|
return [item]
|
return []
|
|
|
def expand_playlist_plays(items: Optional[List[Dict[str, Any]]]) -> List[str]:
|
"""Flatten {name, repeat} into one path per play. Shadowing uses this so each pass has its own listen window."""
|
clips: List[str] = []
|
for item in items or []:
|
if not isinstance(item, dict):
|
continue
|
name = str(item.get("name") or "")
|
if not name:
|
continue
|
clips.extend([name] * _item_repeat(item))
|
return clips
|
|
|
def default_repeater_playlist(task: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
|
"""No downlink yet: task.default_playlist, else the unit01 01×10 / 02×5 stub."""
|
raw = (task or {}).get("default_playlist")
|
if isinstance(raw, list) and raw:
|
items = parse_playlist({"items": raw})
|
if items:
|
return items
|
return [dict(item) for item in TEMP_REPEATER_PLAYLIST]
|
|
|
def default_shadowing_playlist(task: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
|
"""No downlink yet: task.default_playlist, else the unit01 01×5 / 02×10 stub."""
|
raw = (task or {}).get("default_playlist")
|
if isinstance(raw, list) and raw:
|
items = parse_playlist({"items": raw})
|
if items:
|
return items
|
return [dict(item) for item in TEMP_SHADOWING_PLAYLIST]
|
