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