cristhian
04f0d57068
Two sources want one arm and the arm cannot take both. Putting the decision in one place means there is exactly one answer to "why did it not move" rather than one answer per input path. The model is ported from the monolithic repo's RobotService lease rather than replaced with last-writer-wins, because the console is already built on it: its jog pad is gated on owning the arm and it draws an owner chip. A simpler scheme would mean rewriting the page to match the node instead of the other way round. The gamepad takes the arm implicitly, on the first nudge of a stick, so nobody has to reach for a browser to use the thing in their hands — and gives it back after sitting untouched, so a pad left on the bench cannot lock the browser out forever. The web takes and releases explicitly, and may preempt: an operator who can see the arm is better placed than this code to decide they should have it.
107 lines
2.5 KiB
Python
107 lines
2.5 KiB
Python
"""The lease: who may drive the arm, and how it changes hands."""
|
|
|
|
from dora_teleop.arbiter import GAMEPAD, WEB, Arbiter
|
|
|
|
|
|
def build(idle_release_s=2.0):
|
|
return Arbiter(idle_release_s)
|
|
|
|
|
|
def test_a_free_arm_goes_to_whoever_asks():
|
|
arbiter = build()
|
|
|
|
assert arbiter.take(WEB, now := 100.0) is True
|
|
assert arbiter.owner == WEB
|
|
assert arbiter.may_drive(WEB)
|
|
assert now == 100.0
|
|
|
|
|
|
def test_a_held_arm_is_not_handed_over_for_the_asking():
|
|
arbiter = build()
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
assert arbiter.take(WEB, 101.0) is False
|
|
assert arbiter.owner == GAMEPAD
|
|
|
|
|
|
def test_a_non_owner_may_not_drive():
|
|
arbiter = build()
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
assert arbiter.may_drive(WEB) is False
|
|
|
|
|
|
def test_preempting_takes_a_held_arm():
|
|
# The console offers this deliberately: an operator who can see the arm is
|
|
# better placed than this code to decide they should have it.
|
|
arbiter = build()
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
assert arbiter.take(WEB, 101.0, preempt=True) is True
|
|
assert arbiter.owner == WEB
|
|
|
|
|
|
def test_taking_an_arm_you_already_hold_is_not_a_conflict():
|
|
arbiter = build()
|
|
arbiter.take(WEB, 100.0)
|
|
|
|
assert arbiter.take(WEB, 101.0) is True
|
|
|
|
|
|
def test_releasing_frees_the_arm():
|
|
arbiter = build()
|
|
arbiter.take(WEB, 100.0)
|
|
|
|
arbiter.release(WEB)
|
|
|
|
assert arbiter.owner is None
|
|
|
|
|
|
def test_releasing_an_arm_somebody_else_holds_does_nothing():
|
|
arbiter = build()
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
arbiter.release(WEB)
|
|
|
|
assert arbiter.owner == GAMEPAD
|
|
|
|
|
|
def test_an_idle_gamepad_hands_the_arm_back():
|
|
# A pad left on the bench would otherwise lock the browser out forever.
|
|
arbiter = build(idle_release_s=2.0)
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
arbiter.expire_idle_gamepad(102.0)
|
|
|
|
assert arbiter.owner is None
|
|
|
|
|
|
def test_a_gamepad_still_being_used_keeps_the_arm():
|
|
arbiter = build(idle_release_s=2.0)
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
arbiter.note_activity(GAMEPAD, 101.5)
|
|
arbiter.expire_idle_gamepad(102.0)
|
|
|
|
assert arbiter.owner == GAMEPAD
|
|
|
|
|
|
def test_the_web_never_expires_on_its_own():
|
|
# It took the arm by pressing a button and gives it back by pressing another.
|
|
arbiter = build(idle_release_s=2.0)
|
|
arbiter.take(WEB, 100.0)
|
|
|
|
arbiter.expire_idle_gamepad(999.0)
|
|
|
|
assert arbiter.owner == WEB
|
|
|
|
|
|
def test_activity_from_a_non_owner_does_not_extend_anyone_else():
|
|
arbiter = build(idle_release_s=2.0)
|
|
arbiter.take(GAMEPAD, 100.0)
|
|
|
|
arbiter.note_activity(WEB, 101.9)
|
|
arbiter.expire_idle_gamepad(102.0)
|
|
|
|
assert arbiter.owner is None
|