Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions docs/magicbot.rst
Original file line number Diff line number Diff line change
Expand Up @@ -51,3 +51,39 @@ State machines
:exclude-members: members
:undoc-members:
:show-inheritance:

Commands
~~~~~~~~

.. automodule:: magicbot.commands
:members:
:undoc-members:
:show-inheritance:

``magicbot.commands.CommandRunner`` is a MagicBot component that executes
stored ``commands2.Command`` instances using keep-alive semantics.

Call ``self.cmd.run(command)`` each loop that a stored command instance should
remain active. If a command is not re-requested on a later loop, it is ended as
interrupted when it is interruptible.

Important usage constraints:

- construct command instances once and store them on the robot/component
- do not create a new command inside ``execute()`` or a state method each loop
- components, state machines, and robot code may all call ``run(command)``
- multiple different commands may run concurrently
- this is not ``commands2.CommandScheduler`` and does not perform full
scheduler-style requirement arbitration

Example::

class Shooter:
cmd: magicbot.commands.CommandRunner

def setup(self):
self.fire = FireCommand(...)

def execute(self):
if should_fire():
self.cmd.run(self.fire)
75 changes: 75 additions & 0 deletions magicbot/commands.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
from __future__ import annotations

from typing import Callable

import commands2


class CommandRunner:
"""
A MagicBot component that executes ``commands2.Command`` instances using
keep-alive semantics.

Call :meth:`run` to start a stored command instance. Once started, a
command continues running on each later loop until it reports finished.

Multiple different commands may be active concurrently. Repeated
:meth:`run` calls for the same command instance in a single loop are
deduplicated.

This component intentionally does not expose or depend on
``commands2.CommandScheduler``.
"""

robot: object

def __init__(self) -> None:
self._requested: dict[commands2.Command, None] = {}
self._active: dict[commands2.Command, None] = {}

def run(self, command: commands2.Command) -> None:
"""
Request that a stored command instance start running.
"""
self._requested.setdefault(command, None)

def _discard(self, command: commands2.Command) -> None:
self._requested.pop(command, None)
self._active.pop(command, None)

def _call(self, command: commands2.Command, func: Callable[[], object]) -> bool:
try:
func()
except Exception:
self._discard(command)
self.robot.onException()

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like the only reason the robot is injected here is for onException? I think it'd be reasonable to reimplement it here instead of injecting the entire robot and creating a reference cycle.

return False
return True

def execute(self) -> None:
new_commands = [
command for command in self._requested if command not in self._active
]
for command in new_commands:
if self._call(command, command.initialize):
self._active[command] = None

commands_to_execute = [
command for command in self._requested if command in self._active
]
commands_to_execute.extend(
command for command in self._active if command not in self._requested
)

finished: list[commands2.Command] = []
for command in commands_to_execute:
if not self._call(command, command.execute):
continue
if command.isFinished():
if self._call(command, lambda command=command: command.end(False)):
finished.append(command)

for command in finished:
self._active.pop(command, None)

self._requested.clear()
2 changes: 1 addition & 1 deletion magicbot/magicrobot.py
Original file line number Diff line number Diff line change
Expand Up @@ -692,7 +692,7 @@ def _create_components(self) -> None:
self._components = components

def _collect_injectables(self) -> dict[str, Any]:
injectables = {}
injectables = {"robot": self}
cls = type(self)

for n in dir(self):
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ classifiers = [
dependencies = [
"toposort",
"wpilib>=2026.1.1,<2027",
"robotpy-commands-v2>=2026.1.1,<2027",
]

[[project.authors]]
Expand Down
Loading
Loading