Spaces:
Running
Running
File size: 5,543 Bytes
380e7c5 30ab8a7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | import threading
from conftest import FakeBody, FakeNudgeWriter, FakeSpeech, FakeVision
from focus_buddy.loop import FocusLoop
from focus_buddy.observations import Habit, Observation
WORKING = Observation()
PHONE = Observation(looking_at_phone=True, looking_at_screen=False)
NAILS = Observation(biting_nails=True)
def build(settings, observations, frames=10, nudge_writer=None):
body, speech = FakeBody(frames=frames), FakeSpeech()
loop = FocusLoop(
body=body,
vision=FakeVision(observations),
speech=speech,
settings=settings,
nudge_writer=nudge_writer,
)
return loop, body, speech
class TestSpeaking:
def test_stays_quiet_while_you_work(self, settings):
loop, _, speech = build(settings, [WORKING])
for _ in range(5):
loop.tick()
assert speech.spoken == []
def test_nudges_on_a_detected_habit(self, settings):
loop, body, speech = build(settings, [PHONE])
loop.tick()
assert len(speech.spoken) == 1
assert "phone" in speech.spoken[0].lower()
assert len(body.played) == 1
def test_cooldown_suppresses_a_second_nudge(self, settings):
settings.nudge_cooldown_s = 3600.0
loop, _, speech = build(settings, [PHONE])
loop.tick()
loop.tick()
assert len(speech.spoken) == 1
def test_no_frame_means_no_work(self, settings):
loop, _, speech = build(settings, [PHONE], frames=0)
assert loop.tick() is None
assert speech.spoken == []
class TestMemoryIntegration:
def test_counts_the_episode_it_nudged_about(self, settings):
loop, _, _ = build(settings, [NAILS])
loop.tick()
assert loop.memory.counters[Habit.BITING_NAILS] == 1
def test_summary_is_spoken_when_due(self, settings):
settings.summary_every_s = 0.0
loop, _, speech = build(settings, [WORKING])
loop.tick()
assert any("You've been working" in line for line in speech.spoken)
class TestLLMNudges:
def test_uses_a_valid_model_nudge(self, settings):
settings.use_llm_nudges = True
writer = FakeNudgeWriter("Hey, maybe set the phone aside now.")
loop, _, speech = build(settings, [PHONE], nudge_writer=writer)
loop.tick()
assert speech.spoken == ["Hey, maybe set the phone aside now."]
assert writer.calls == ["phone"]
def test_falls_back_when_the_model_talks_about_the_wrong_habit(self, settings):
settings.use_llm_nudges = True
writer = FakeNudgeWriter("Hey, you are touching your face again.")
loop, _, speech = build(settings, [PHONE], nudge_writer=writer)
loop.tick()
# Rejected, so a template about the phone is spoken instead.
assert "phone" in speech.spoken[0].lower()
assert speech.spoken[0] != writer.reply
def test_falls_back_when_the_model_raises(self, settings):
settings.use_llm_nudges = True
writer = FakeNudgeWriter(explode=True)
loop, _, speech = build(settings, [PHONE], nudge_writer=writer)
loop.tick()
assert len(speech.spoken) == 1
def test_templates_are_used_when_llm_nudges_are_off(self, settings):
writer = FakeNudgeWriter("Hey, maybe set the phone aside now.")
loop, _, _ = build(settings, [PHONE], nudge_writer=writer)
loop.tick()
assert writer.calls == []
class TestLifecycle:
def test_run_stops_when_the_event_is_set(self, settings):
loop, body, _ = build(settings, [WORKING], frames=1000)
stop = threading.Event()
original_tick = loop.tick
ticks = []
def counting_tick():
ticks.append(1)
if len(ticks) >= 3:
stop.set()
return original_tick()
loop.tick = counting_tick
loop.run(stop)
assert len(ticks) == 3
def test_run_closes_everything_it_owns(self, settings):
writer = FakeNudgeWriter()
loop, body, _ = build(settings, [WORKING], nudge_writer=writer)
stop = threading.Event()
stop.set()
loop.run(stop)
assert body.closed and writer.closed
class TestBodyOwnership:
"""A settings reload rebuilds the backends but keeps the robot connection."""
def _loop(self, owns_body):
from focus_buddy.config import Settings, SpeechEngine
from focus_buddy.loop import FocusLoop
body, vision, speech = FakeBody(), FakeVision([Observation()]), FakeSpeech()
loop = (
FocusLoop(
body=body,
vision=vision,
speech=speech,
settings=Settings(speech=SpeechEngine.NONE),
)
if owns_body
else FocusLoop(
body=body,
vision=vision,
speech=speech,
settings=Settings(speech=SpeechEngine.NONE),
owns_body=False,
)
)
return loop, body, vision
def test_by_default_the_loop_closes_the_body(self):
loop, body, vision = self._loop(True)
loop.close()
assert body.closed is True and vision.closed is True
def test_a_borrowed_body_is_left_open(self):
"""Reconnecting to the robot cost ~20s and a wobble toggle for nothing."""
loop, body, vision = self._loop(False)
loop.close()
assert body.closed is False
# The things that do depend on settings are still torn down.
assert vision.closed is True
|