"""Playback speed clamp and repeater speed gradients."""
|
|
from core.app_runtime.playback import (
|
PLAY_SPEED_DEFAULT,
|
PLAYLIST_ITEM_MAX,
|
clamp_play_speed,
|
daily_base_speed,
|
expand_lesson_playlist,
|
lesson_day_index,
|
mark_lesson_played_today,
|
parse_playlist,
|
playlist_from_params,
|
speeds_for_repeats,
|
)
|
from core.app_runtime.runtime import (
|
_apply_playback_speed,
|
_begin_repeater_playlist,
|
_begin_shadowing_playlist,
|
_continue_repeater_if_needed,
|
)
|
from types import SimpleNamespace
|
import asyncio
|
|
|
def test_clamp_play_speed_range():
|
assert clamp_play_speed(None) == PLAY_SPEED_DEFAULT
|
assert clamp_play_speed("nope") == PLAY_SPEED_DEFAULT
|
assert clamp_play_speed(0.3) == 0.5
|
assert clamp_play_speed(3.0) == 2.0
|
assert clamp_play_speed(0.8) == 0.8
|
assert clamp_play_speed(1.05) == 1.05
|
assert clamp_play_speed(1.0) == 1.0
|
|
|
def test_daily_speeds_use_practice_day():
|
task = {"playback": {"daily_speeds": [0.8, 0.9, 1.0, 1.05, 1.1, 1.15, 1.2]}}
|
assert daily_base_speed(task, 0) == 0.8
|
assert daily_base_speed(task, 3) == 1.05
|
assert daily_base_speed(task, 99) == 1.2
|
|
|
def test_repeat_speeds_five_passes():
|
task = {
|
"playback": {
|
"daily_speeds": [0.8, 0.9, 1.0],
|
"repeat_speeds": [0.8, 0.9, 1.0, 1.1, 1.2],
|
}
|
}
|
assert speeds_for_repeats(task, 1, day_index=0) == [0.8]
|
assert speeds_for_repeats(task, 5, day_index=0) == [0.8, 0.9, 1.0, 1.1, 1.2]
|
|
|
def test_repeat_one_does_not_use_full_curve():
|
task = {"playback": {"speed": 1.0, "repeat_speeds": [0.8, 0.9, 1.0, 1.1, 1.2]}}
|
assert speeds_for_repeats(task, 1) == [1.0]
|
|
|
def test_lesson_day_index_counts_distinct_days(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))
|
assert lesson_day_index("mac1", "english_learning", "repeater", "l1.001.mp3") == 0
|
mark_lesson_played_today("mac1", "english_learning", "repeater", "l1.001.mp3")
|
assert lesson_day_index("mac1", "english_learning", "repeater", "l1.001.mp3") == 1
|
mark_lesson_played_today("mac1", "english_learning", "repeater", "l1.001.mp3")
|
assert lesson_day_index("mac1", "english_learning", "repeater", "l1.001.mp3") == 1
|
|
|
def test_apply_playback_speed_queues_gradient():
|
task = {"playback": {"repeat_speeds": [0.8, 0.9, 1.0, 1.1, 1.2]}}
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
repeater_power_off_after=False,
|
)
|
play = _apply_playback_speed(
|
conn, task, {"name": "/english_learning/repeater/lessons/l1/001/cn/01.mp3", "repeat": 5}
|
)
|
assert play["speed"] == 0.8
|
assert play["repeat"] == 1
|
assert play["power_off_after"] is False
|
assert conn.repeater_speed_queue == [0.9, 1.0, 1.1, 1.2]
|
|
|
def test_explicit_speed_skips_gradient():
|
task = {"playback": {"repeat_speeds": [0.8, 0.9, 1.0, 1.1, 1.2]}}
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
repeater_power_off_after=False,
|
)
|
play = _apply_playback_speed(conn, task, {"name": "a.mp3", "repeat": 5, "speed": 1.05})
|
assert play["speed"] == 1.05
|
assert play["repeat"] == 5
|
assert conn.repeater_speed_queue == []
|
|
|
def test_continue_repeater_plays_next_speed(monkeypatch):
|
played = []
|
|
async def fake_play(conn, params):
|
played.append(params["speed"])
|
return None
|
|
monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
|
conn = SimpleNamespace(
|
repeater_speed_queue=[0.9, 1.0],
|
repeater_play_name="a.mp3",
|
repeater_power_off_after=False,
|
interaction_mode="PLAY_AUDIO",
|
)
|
assert asyncio.run(_continue_repeater_if_needed(conn)) is True
|
assert played == [0.9]
|
assert conn.repeater_speed_queue == [1.0]
|
assert asyncio.run(_continue_repeater_if_needed(conn)) is True
|
assert played == [0.9, 1.0]
|
assert conn.repeater_speed_queue == []
|
assert asyncio.run(_continue_repeater_if_needed(conn)) is False
|
|
|
def test_parse_playlist_from_items_and_urls():
|
items = parse_playlist(
|
{
|
"items": [
|
{
|
"url": "http://cdn.example/static/english_learning/repeater/lessons/a.mp3",
|
"repeat": 3,
|
},
|
{
|
"path": "/english_learning/repeater/lessons/b.mp3",
|
"count": 2,
|
"speed": 0.9,
|
},
|
]
|
}
|
)
|
assert items[0]["name"] == "/english_learning/repeater/lessons/a.mp3"
|
assert items[0]["repeat"] == 3
|
assert items[1]["name"] == "/english_learning/repeater/lessons/b.mp3"
|
assert items[1]["repeat"] == 2
|
assert items[1]["speed"] == 0.9
|
|
|
def test_parse_playlist_safety_cap_is_256_not_a_lesson_limit():
|
items = parse_playlist(
|
{
|
"items": [
|
{"name": f"/english_learning/shadowing/lessons/l1/001/en/{i:03d}.mp3"}
|
for i in range(1, PLAYLIST_ITEM_MAX + 4)
|
]
|
}
|
)
|
assert PLAYLIST_ITEM_MAX == 256
|
assert len(items) == PLAYLIST_ITEM_MAX
|
assert items[0]["name"].endswith("001.mp3")
|
assert items[-1]["name"].endswith("256.mp3")
|
|
|
def test_begin_repeater_queues_remaining_items():
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
repeater_power_off_after=False,
|
repeater_playlist=[],
|
)
|
task = {"playback": {"speed": 1.0}}
|
first = _begin_repeater_playlist(
|
conn,
|
task,
|
{
|
"items": [
|
{"name": "/english_learning/repeater/lessons/a.mp3", "repeat": 2, "speed": 1.0},
|
{"name": "/english_learning/repeater/lessons/b.mp3", "repeat": 1, "speed": 0.9},
|
]
|
},
|
)
|
assert first["name"].endswith("/a.mp3")
|
assert first["repeat"] == 2
|
assert first.get("finish_task") is False
|
assert len(first["items"]) == 2
|
assert first["items"][1]["name"].endswith("/b.mp3")
|
assert first["items"][1]["repeat"] == 1
|
assert conn.repeater_playlist == []
|
|
|
def test_continue_repeater_plays_next_playlist_file(monkeypatch):
|
played = []
|
|
async def fake_play(conn, params):
|
played.append((params.get("name"), params.get("repeat"), params.get("speed"), params.get("finish_task")))
|
return None
|
|
monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="/english_learning/repeater/lessons/a.mp3",
|
repeater_power_off_after=False,
|
repeater_playlist=[{"name": "/english_learning/repeater/lessons/b.mp3", "repeat": 2, "speed": 0.9}],
|
interaction_mode="PLAY_AUDIO",
|
)
|
assert asyncio.run(_continue_repeater_if_needed(conn)) is True
|
assert played == [("/english_learning/repeater/lessons/b.mp3", 2, 0.9, False)]
|
assert conn.repeater_playlist == []
|
assert asyncio.run(_continue_repeater_if_needed(conn)) is False
|
|
|
def test_begin_repeater_without_lesson_has_no_playlist():
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
repeater_power_off_after=False,
|
repeater_playlist=[],
|
)
|
first = _begin_repeater_playlist(conn, {"playback": {"speed": 1.0}}, {})
|
assert first.get("items") == []
|
assert not first.get("name")
|
|
|
def test_begin_repeater_uses_catalog_lesson():
|
from core.app_runtime.playback import TEMP_REPEATER_PLAYLIST
|
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
repeater_power_off_after=False,
|
repeater_playlist=[],
|
)
|
first = _begin_repeater_playlist(conn, {"playback": {"speed": 1.0}}, {"lesson_id": "l1.001"})
|
assert first["name"] == TEMP_REPEATER_PLAYLIST[0]["name"]
|
assert first["repeat"] == 1
|
assert first["speed"] == 1.0
|
assert first.get("finish_task") is False
|
assert len(first["items"]) == 4
|
assert first["items"][0]["repeat"] == 1
|
assert first["items"][1]["name"] == TEMP_REPEATER_PLAYLIST[1]["name"]
|
assert first["items"][1]["repeat"] == 1
|
assert first["items"][1]["speed"] == 0.8
|
assert first["items"][2]["speed"] == 0.9
|
assert first["items"][3]["repeat"] == 5
|
assert first["items"][3]["speed"] == 1.0
|
assert conn.repeater_playlist == []
|
|
|
def test_begin_shadowing_playlist_one_shot(tmp_path, monkeypatch):
|
from core.app_runtime.playback import TEMP_SHADOWING_PLAYLIST
|
|
monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
|
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="shadowing",
|
config={},
|
shadowing_clips=[],
|
shadowing_index=0,
|
shadowing_prompt="",
|
shadowing_awaiting_user=True,
|
)
|
play = _begin_shadowing_playlist(conn, {"lesson_id": "l1.001"})
|
assert play["name"] == TEMP_SHADOWING_PLAYLIST[0]["name"]
|
assert play["repeat"] == 1
|
assert play["speed"] == 0.8
|
assert play.get("finish_task") is False
|
assert play.get("listen_ratio") == 1.2
|
assert len(play["items"]) == 58
|
assert play["items"][0]["repeat"] == 1
|
assert play["items"][-1]["name"] == TEMP_SHADOWING_PLAYLIST[-1]["name"]
|
assert play["items"][-1]["speed"] == 0.8
|
assert conn.shadowing_clips.count(TEMP_SHADOWING_PLAYLIST[0]["name"]) == 1
|
assert conn.shadowing_clips[-1] == TEMP_SHADOWING_PLAYLIST[-1]["name"]
|
assert len(conn.shadowing_clips) == 58
|
assert conn.shadowing_awaiting_user is False
|
|
|
def test_expand_shadowing_playlist_one_clip_per_listen_window(tmp_path, monkeypatch):
|
from core.app_runtime.playback import (
|
TEMP_SHADOWING_PLAYLIST,
|
expand_playlist_plays,
|
)
|
|
monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
|
clips = expand_playlist_plays(expand_lesson_playlist("shadowing", "l1.001"))
|
assert clips[0] == TEMP_SHADOWING_PLAYLIST[0]["name"]
|
assert clips[-1] == TEMP_SHADOWING_PLAYLIST[-1]["name"]
|
assert len(clips) == 58
|
|
|
def test_audio_completion_timeout_scales_with_queue():
|
from core.utils.audioRateController import audio_completion_timeout_sec
|
|
assert audio_completion_timeout_sec(0) == 8.0
|
assert audio_completion_timeout_sec(82) == 8.0 + 82 * 0.06
|
assert audio_completion_timeout_sec(5000) == 180.0
|
|
|
def test_execute_named_tool_play_sends_active_task(monkeypatch):
|
from core.app_runtime.runtime import _execute_named_tool_play
|
|
captured = {}
|
|
async def fake_tool(conn, tool_name, arguments):
|
captured["tool"] = tool_name
|
captured["arguments"] = arguments
|
return None
|
|
monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool", fake_tool)
|
conn = SimpleNamespace(
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
)
|
asyncio.run(
|
_execute_named_tool_play(
|
conn,
|
{
|
"name": "/english_learning/shadowing/lessons/l1/001/en/02.mp3",
|
"repeat": 5,
|
"speed": 1.0,
|
"finish_task": True,
|
"items": [
|
{
|
"name": "/english_learning/shadowing/lessons/l1/001/en/01.mp3",
|
"repeat": 10,
|
"speed": 1.0,
|
},
|
{
|
"name": "/english_learning/shadowing/lessons/l1/001/en/02.mp3",
|
"repeat": 5,
|
"speed": 1.0,
|
},
|
],
|
},
|
)
|
)
|
assert captured["arguments"]["task_id"] == "repeater"
|
assert captured["arguments"]["app_id"] == "english_learning"
|
assert captured["arguments"]["finish_task"] is False
|
assert captured["arguments"]["force"] is True
|
assert "/01.mp3" in captured["arguments"]["items"]
|
assert "/02.mp3" in captured["arguments"]["items"]
|
|
|
def test_execute_named_tool_play_retries_duplicate_ignored(monkeypatch):
|
from core.app_runtime.runtime import _execute_named_tool_play
|
from plugins_func.register import Action, ActionResponse
|
|
calls = []
|
|
async def fake_tool(conn, tool_name, arguments):
|
calls.append(dict(arguments))
|
if len(calls) == 1:
|
return ActionResponse(
|
action=Action.RESPONSE,
|
response='{"accepted":true,"duplicate_ignored":true}',
|
)
|
return ActionResponse(
|
action=Action.RESPONSE,
|
response='{"accepted":true,"repeat":1,"power_off_after":false}',
|
)
|
|
monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool", fake_tool)
|
conn = SimpleNamespace(active_app_id="english_learning", current_task_id="shadowing")
|
notice = asyncio.run(
|
_execute_named_tool_play(
|
conn,
|
{
|
"name": "/english_learning/shadowing/lessons/l1/001/en/01.mp3",
|
"repeat": 1,
|
"speed": 1.0,
|
"finish_task": False,
|
},
|
)
|
)
|
assert notice is None
|
assert len(calls) == 2
|
assert calls[0]["task_id"] == "shadowing"
|
assert calls[0]["finish_task"] is False
|
assert calls[1]["speed"] == 1.02
|
assert calls[1]["force"] is True
|
|
|
def test_play_shadowing_clip_keeps_task_open(monkeypatch):
|
from core.app_runtime.runtime import _play_shadowing_clip
|
|
captured = {}
|
|
async def fake_play(conn, params):
|
captured["params"] = params
|
return None
|
|
monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
|
conn = SimpleNamespace(
|
interaction_mode="SHADOWING",
|
shadowing_clip_gen=0,
|
active_app_id="english_learning",
|
current_task_id="shadowing",
|
config={},
|
)
|
ok = asyncio.run(_play_shadowing_clip(conn, "/english_learning/shadowing/lessons/l1/001/en/01.mp3"))
|
assert ok is True
|
assert captured["params"]["finish_task"] is False
|
assert captured["params"]["repeat"] == 1
|
|
|
def test_lesson_end_speech_includes_chat_greeting():
|
from core.app_runtime.runtime import _lesson_end_speech
|
|
conn = SimpleNamespace(config={})
|
assert _lesson_end_speech(conn, "泛听结束") == "泛听结束。我们来聊天吧。"
|
assert _lesson_end_speech(conn, "跟读结束") == "跟读结束。我们来聊天吧。"
|
|
|
def test_expand_l1_001_repeater_and_shadowing(tmp_path, monkeypatch):
|
from core.app_runtime.playback import TEMP_REPEATER_PLAYLIST, TEMP_SHADOWING_PLAYLIST
|
|
monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
|
repeater = expand_lesson_playlist("repeater", "l1.001")
|
assert [item["name"] for item in repeater] == [item["name"] for item in TEMP_REPEATER_PLAYLIST]
|
assert repeater[3]["repeat"] == 5
|
assert "schedule" not in repeater[0]
|
|
overridden = expand_lesson_playlist("repeater", "l1.001", repeat=1)
|
assert all(item["repeat"] == 1 for item in overridden)
|
|
shadowing = expand_lesson_playlist("shadowing", "l1.001")
|
assert len(shadowing) == 58
|
assert shadowing[0]["name"] == TEMP_SHADOWING_PLAYLIST[0]["name"]
|
assert shadowing[-1]["name"].endswith("/58.mp3")
|
assert all(item["speed"] == 0.8 for item in shadowing)
|
|
|
def test_expand_catalog_story_en_passes():
|
from core.app_runtime.idiom_catalog import find_story
|
|
story = find_story("l1.002")
|
assert story is not None
|
assert story["id"] == "002"
|
items = expand_lesson_playlist(
|
"repeater",
|
"l1.002",
|
en_passes=[
|
{"speed": 0.8, "repeat": 1},
|
{"speed": 0.9, "repeat": 2},
|
{"speed": 1.0, "repeat": 5},
|
],
|
)
|
assert items[0]["name"].endswith("/l1/002/cn/01.mp3")
|
assert items[0]["repeat"] == 1
|
assert items[0]["speed"] == 1.0
|
assert [item["speed"] for item in items[1:]] == [0.8, 0.9, 1.0]
|
assert [item["repeat"] for item in items[1:]] == [1, 2, 5]
|
assert all(item["name"].endswith("/l1/002/en/01.mp3") for item in items[1:])
|
|
|
def test_expand_catalog_shadowing_sentence_files():
|
items = expand_lesson_playlist(
|
"shadowing",
|
"l1.079",
|
language="en",
|
speed=0.8,
|
)
|
assert len(items) == 30
|
assert items[0]["name"].endswith("/l1/079/en/01.mp3")
|
assert items[-1]["name"].endswith("/l1/079/en/30.mp3")
|
assert all(item["speed"] == 0.8 for item in items)
|
|
cn_items = expand_lesson_playlist(
|
"shadowing",
|
"l1.079",
|
language="cn",
|
speed=0.9,
|
)
|
assert cn_items[0]["name"].endswith("/l1/079/cn/01.mp3")
|
assert cn_items[0]["speed"] == 0.9
|
|
|
def test_expand_l1_001_custom_en_pass(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
|
items = expand_lesson_playlist(
|
"repeater",
|
"l1.001",
|
en_passes=[{"speed": 0.8, "repeat": 1}],
|
)
|
assert items[0]["name"].endswith("/l1/001/cn/01.mp3")
|
assert items[1]["name"].endswith("/l1/001/en/01.mp3")
|
|
shadowing = expand_lesson_playlist(
|
"shadowing",
|
"l1.001",
|
language="en",
|
speed=0.8,
|
sentence_count_override=58,
|
)
|
assert shadowing[0]["name"].endswith("/l1/001/en/01.mp3")
|
assert len(shadowing) == 58
|
|
|
def test_idiom_catalog_lists_all_chapters():
|
from core.app_runtime.idiom_catalog import find_story, load_catalog
|
|
data = load_catalog()
|
assert len(data["systems"]) == 7
|
assert len(data["stories"]) == 86
|
assert find_story("l1.001")["name"] == "开天辟地"
|
assert find_story("l1.001")["id"] == "001"
|
assert find_story("l3.048")["nameEn"].startswith("An ugly girl")
|
assert find_story("l6.086")["id"] == "086"
|
|
|
def test_unknown_lesson_does_not_fallback_to_temp():
|
assert expand_lesson_playlist("repeater", "l9.999") == []
|
assert expand_lesson_playlist("shadowing", "missing") == []
|
items = playlist_from_params(
|
{"default_playlist": [{"name": "/x.mp3", "repeat": 1}]},
|
{"lesson_id": "l9.999"},
|
"repeater",
|
lambda task: [{"name": "/fallback.mp3", "repeat": 1}],
|
)
|
assert items == []
|
|
|
def test_voice_without_lesson_does_not_use_temp_default():
|
from core.app_runtime.playback import default_repeater_playlist
|
|
items = playlist_from_params(None, {}, "repeater", default_repeater_playlist)
|
assert items == []
|
|
|
def test_remember_and_load_last_lesson(tmp_path, monkeypatch):
|
from core.app_runtime.playback import load_last_lesson_params, remember_last_lesson_params
|
|
monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))
|
remember_last_lesson_params("mac-1", "repeater", {"lesson_id": "l2.010", "en_passes": [{"speed": 0.8, "repeat": 1}]})
|
remember_last_lesson_params("mac-1", "shadowing", {"lesson_id": "l3.048", "speed": 0.8, "language": "en"})
|
loaded = load_last_lesson_params("mac-1", "repeater")
|
assert loaded["lesson_id"] == "l2.010"
|
assert load_last_lesson_params("mac-1", "shadowing")["lesson_id"] == "l3.048"
|
assert load_last_lesson_params("mac-1", "chat") == {}
|
assert load_last_lesson_params("other", "repeater") == {}
|
|
|
def test_begin_repeater_expands_lesson_id():
|
conn = SimpleNamespace(
|
device_id="mac",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
repeater_power_off_after=False,
|
repeater_playlist=[],
|
)
|
play = _begin_repeater_playlist(conn, {}, {"lesson_id": "l1.001", "repeat": 1})
|
assert play["name"].endswith("/l1/001/cn/01.mp3")
|
assert len(play["items"]) == 4
|
assert all(item["repeat"] == 1 for item in play["items"])
|
assert play.get("finish_task") is False
|
