70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""The dora event loop: one reading published per tick.
|
|
|
|
Every tick, changed or not. That is deliberate — the stream is also a heartbeat,
|
|
and it is what lets a consumer tell "the pad is being held still" from "this node
|
|
died". A consumer driving anything that moves needs both answers, and only one of
|
|
them can be inferred from silence.
|
|
|
|
Nothing here decides what a reading means. The pad is read in
|
|
:mod:`dora_joy.gamepad` and packed in :mod:`dora_joy.message`, and that is the
|
|
whole node.
|
|
"""
|
|
|
|
import argparse
|
|
|
|
from dora import Node
|
|
|
|
from dora_joy import message
|
|
from dora_joy.config import Config
|
|
from dora_joy.gamepad import Gamepad
|
|
from dora_joy.log import log
|
|
|
|
|
|
def main() -> None:
|
|
"""Start reading the pad and publish it until the tick closes."""
|
|
args = _parse_args()
|
|
config = Config.from_env()
|
|
|
|
# Passing the name lets the node also run standalone as a dynamic node; when
|
|
# dora spawns it, the id from the dataflow wins.
|
|
node = Node(args.name)
|
|
gamepad = Gamepad(config.joystick_index)
|
|
gamepad.start()
|
|
log(f"publishing joystick {config.joystick_index} on every tick")
|
|
|
|
present = None
|
|
|
|
try:
|
|
for event in node:
|
|
if event["type"] == "ERROR":
|
|
raise RuntimeError(event["error"])
|
|
if event["type"] == "INPUT_CLOSED" and event.get("id") == "tick":
|
|
break
|
|
if event["type"] != "INPUT" or event["id"] != "tick":
|
|
continue
|
|
|
|
sample = gamepad.read()
|
|
# A pad arriving or going away is worth one line each; the readings
|
|
# themselves are on the wire for anyone who wants them.
|
|
if sample.present != present:
|
|
present = sample.present
|
|
log("gamepad connected" if present else "no gamepad")
|
|
node.send_output("joy", message.pack(sample))
|
|
finally:
|
|
gamepad.shutdown()
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
"""Read the node name, for when dora did not spawn us."""
|
|
parser = argparse.ArgumentParser(description="Publish a gamepad's axes and buttons into a dataflow.")
|
|
parser.add_argument(
|
|
"--name",
|
|
default="joy",
|
|
help="The name of the node in the dataflow, used when running as a dynamic node.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|