"""Playback speed clamp and repeater speed gradients."""
|
|
from core.app_runtime.playback import (
|
PLAY_SPEED_DEFAULT,
|
clamp_play_speed,
|
daily_base_speed,
|
lesson_day_index,
|
mark_lesson_played_today,
|
parse_playlist,
|
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", "unit01.mp3") == 0
|
mark_lesson_played_today("mac1", "english_learning", "repeater", "unit01.mp3")
|
assert lesson_day_index("mac1", "english_learning", "repeater", "unit01.mp3") == 1
|
mark_lesson_played_today("mac1", "english_learning", "repeater", "unit01.mp3")
|
assert lesson_day_index("mac1", "english_learning", "repeater", "unit01.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/unit01/full.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_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_uses_temp_unit01_playlist():
|
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}}, {})
|
assert first["name"] == TEMP_REPEATER_PLAYLIST[0]["name"]
|
assert first["repeat"] == 10
|
assert first.get("finish_task") is False
|
assert len(first["items"]) == 2
|
assert first["items"][0]["repeat"] == 10
|
assert first["items"][1]["name"] == TEMP_REPEATER_PLAYLIST[1]["name"]
|
assert first["items"][1]["repeat"] == 5
|
assert conn.repeater_playlist == []
|
|
|
def test_begin_shadowing_playlist_one_shot():
|
from core.app_runtime.playback import TEMP_SHADOWING_PLAYLIST
|
|
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, {})
|
assert play["name"] == TEMP_SHADOWING_PLAYLIST[0]["name"]
|
assert play["repeat"] == 5
|
assert play.get("finish_task") is False
|
assert play.get("listen_ratio") == 1.2
|
assert len(play["items"]) == 2
|
assert play["items"][0]["repeat"] == 5
|
assert play["items"][1]["name"] == TEMP_SHADOWING_PLAYLIST[1]["name"]
|
assert play["items"][1]["repeat"] == 10
|
assert conn.shadowing_clips.count(TEMP_SHADOWING_PLAYLIST[0]["name"]) == 5
|
assert conn.shadowing_clips.count(TEMP_SHADOWING_PLAYLIST[1]["name"]) == 10
|
assert len(conn.shadowing_clips) == 15
|
assert conn.shadowing_awaiting_user is False
|
|
|
def test_expand_shadowing_playlist_one_clip_per_listen_window():
|
from core.app_runtime.playback import (
|
TEMP_SHADOWING_PLAYLIST,
|
default_shadowing_playlist,
|
expand_playlist_plays,
|
)
|
|
clips = expand_playlist_plays(default_shadowing_playlist())
|
assert clips.count(TEMP_SHADOWING_PLAYLIST[0]["name"]) == 5
|
assert clips.count(TEMP_SHADOWING_PLAYLIST[1]["name"]) == 10
|
assert len(clips) == 15
|
|
|
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/unit01/02.mp3",
|
"repeat": 5,
|
"speed": 1.0,
|
"finish_task": True,
|
"items": [
|
{
|
"name": "/english_learning/shadowing/lessons/unit01/01.mp3",
|
"repeat": 10,
|
"speed": 1.0,
|
},
|
{
|
"name": "/english_learning/shadowing/lessons/unit01/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/unit01/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/unit01/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, "跟读结束") == "跟读结束。我们来聊天吧。"
|
