utils.input

Controller input factory — config-driven managed input objects.

Public API::

from utils.input import InputFactory, get_factory

# In robotInit — create the factory (registers as active):
factory = InputFactory(config_path="data/inputs/controller.yaml")

# Anywhere else — fetch the active factory:
factory = get_factory()
button = factory.getButton("intake.run")
analog = factory.getAnalog("drivetrain.speed")
rumble = factory.getRumbleControl("general.rumble_left")

View Source

 1"""Controller input factory — config-driven managed input objects.
 2
 3Public API::
 4
 5    from utils.input import InputFactory, get_factory
 6
 7    # In robotInit — create the factory (registers as active):
 8    factory = InputFactory(config_path="data/inputs/controller.yaml")
 9
10    # Anywhere else — fetch the active factory:
11    factory = get_factory()
12    button = factory.getButton("intake.run")
13    analog = factory.getAnalog("drivetrain.speed")
14    rumble = factory.getRumbleControl("general.rumble_left")
15"""
16
17from utils.input.factory import InputFactory, get_factory
18from utils.input.managed_analog import ManagedAnalog
19from utils.input.managed_button import ManagedButton
20from utils.input.managed_rumble import ManagedRumble
21from utils.input.validation import ValidationIssue, validate_config
22
23__all__ = [
24    "InputFactory",
25    "get_factory",
26    "ManagedAnalog",
27    "ManagedButton",
28    "ManagedRumble",
29    "ValidationIssue",
30    "validate_config",
31]

def get_factory() -> InputFactory: View Source

66def get_factory() -> "InputFactory":
67    """Return the active InputFactory instance.
68
69    Allows any module (subsystems, commands, etc.) to fetch managed
70    inputs without having the factory passed through constructors::
71
72        from utils.input import get_factory
73        rumble = get_factory().getRumbleControl("feedback.rumble")
74
75    Raises RuntimeError if no factory has been created yet.
76    """
77    if _active_factory is None:
78        raise RuntimeError(
79            "InputFactory not initialized — create one in robotInit "
80            "before calling get_factory()")
81    return _active_factory

Return the active InputFactory instance.

Allows any module (subsystems, commands, etc.) to fetch managed inputs without having the factory passed through constructors::

from utils.input import get_factory
rumble = get_factory().getRumbleControl("feedback.rumble")

Raises RuntimeError if no factory has been created yet.

class ManagedAnalog(utils.input._nt_mapping.NtMappingMixin): View Source

 54class ManagedAnalog(NtMappingMixin):
 55    """A managed analog input with full shaping pipeline.
 56
 57    Args:
 58        action: The ActionDefinition this analog represents (may be None).
 59        accessor: Callable returning the raw axis value each cycle.
 60        default_value: Value returned when no accessor is bound.
 61    """
 62
 63    _PARAM_TYPES: dict[str, type] = {
 64        "deadband": float,
 65        "inversion": bool,
 66        "scale": float,
 67        "slew_rate": float,
 68    }
 69
 70    def __init__(
 71        self,
 72        action: ActionDefinition | None,
 73        accessor: Callable[[], float],
 74        default_value: float = 0.0,
 75    ):
 76        self._action = action
 77        self._accessor = accessor
 78        self._default_value = default_value
 79
 80        # Mutable properties — initialized from action or defaults
 81        if action is not None:
 82            self._deadband = action.deadband
 83            self._inversion = action.inversion
 84            self._scale = action.scale
 85            self._slew_rate = action.slew_rate
 86            self._trigger_mode = action.trigger_mode
 87            self._extra = action.extra or {}
 88        else:
 89            self._deadband = 0.0
 90            self._inversion = False
 91            self._scale = 1.0
 92            self._slew_rate = 0.0
 93            self._trigger_mode = EventTriggerMode.RAW
 94            self._extra = {}
 95
 96        # Binding info — set by the factory after construction.
 97        # Format: "ControllerName.input_name (channel)"
 98        self._binding_info: str = "unbound"
 99
100        self._init_nt_mapping()
101
102        self._pipeline: Callable[[float], float] = lambda x: x
103        self._slew_limiter = None
104        self._rebuild_pipeline()
105        self._rebuild_slew_limiter()
106
107    # --- Properties with pipeline rebuild ---
108
109    @property
110    def deadband(self) -> float:
111        return self._deadband
112
113    @deadband.setter
114    def deadband(self, value: float) -> None:
115        if self._deadband != value:
116            self._deadband = value
117            self._rebuild_pipeline()
118
119    @property
120    def inversion(self) -> bool:
121        return self._inversion
122
123    @inversion.setter
124    def inversion(self, value: bool) -> None:
125        if self._inversion != value:
126            self._inversion = value
127            self._rebuild_pipeline()
128
129    @property
130    def scale(self) -> float:
131        return self._scale
132
133    @scale.setter
134    def scale(self, value: float) -> None:
135        if self._scale != value:
136            self._scale = value
137            self._rebuild_pipeline()
138
139    @property
140    def slew_rate(self) -> float:
141        """Max output change rate in units/sec.  0 = disabled.
142
143        Applies symmetrically (same limit up and down) unless
144        ``extra["negative_slew_rate"]`` is set on the action.
145        """
146        return self._slew_rate
147
148    @slew_rate.setter
149    def slew_rate(self, value: float) -> None:
150        if self._slew_rate != value:
151            self._slew_rate = value
152            self._rebuild_slew_limiter()
153
154    @property
155    def action(self) -> ActionDefinition | None:
156        return self._action
157
158    # --- Value access ---
159
160    def get(self) -> float:
161        """Return the shaped value through the full pipeline + slew limit."""
162        shaped = self._pipeline(self._accessor())
163        if self._slew_limiter is not None:
164            shaped = self._slew_limiter.calculate(shaped)
165        self._last_slew_output = shaped
166        return shaped
167
168    def getRaw(self) -> float:
169        """Return the unmodified raw HID value."""
170        return self._accessor()
171
172    def __call__(self) -> float:
173        """Alias for get() — makes ManagedAnalog a Callable[[], float]."""
174        return self.get()
175
176    # --- Internal ---
177
178    def __str__(self) -> str:
179        name = self._action.qualified_name if self._action else "unbound"
180        mode = self._trigger_mode.value
181        parts = [f"ManagedAnalog('{name}'"]
182        parts.append(f"bind={self._binding_info}")
183        parts.append(f"mode={mode}")
184        parts.append(f"pipeline={self._pipeline}")
185        if self._slew_rate > 0:
186            neg = self._extra.get("negative_slew_rate", -self._slew_rate)
187            parts.append(f"slew_rate={self._slew_rate}/{neg}")
188        parts_str = ", ".join(parts)
189        return f"{parts_str})"
190
191    def _rebuild_pipeline(self) -> None:
192        """Rebuild the shaping closure from current properties."""
193        self._pipeline = build_shaping_pipeline(
194            inversion=self._inversion,
195            deadband=self._deadband,
196            trigger_mode=self._trigger_mode,
197            scale=self._scale,
198            extra=self._extra,
199            action_name=(
200                self._action.qualified_name if self._action else ""),
201        )
202
203    def _rebuild_slew_limiter(self) -> None:
204        """Recreate the SlewRateLimiter when rate changes.
205
206        SlewRateLimiter rate limits are constructor params, so a new
207        instance is needed when the rate changes.  Setting slew_rate
208        to 0 disables the limiter.
209
210        The negative (decreasing) rate defaults to ``-slew_rate`` but
211        can be overridden via ``extra["negative_slew_rate"]`` in the
212        action's YAML config for asymmetric limiting.
213
214        Preserves the last output value so changing the rate at runtime
215        doesn't cause a transient glitch (the new limiter starts from
216        where the old one left off rather than resetting to 0).
217        """
218        if self._slew_rate <= 0:
219            self._slew_limiter = None
220            return
221        try:
222            from wpimath.filter import SlewRateLimiter
223            neg_rate = self._extra.get(
224                "negative_slew_rate", -self._slew_rate)
225            initial = getattr(self, '_last_slew_output', 0.0)
226            self._slew_limiter = SlewRateLimiter(
227                self._slew_rate, neg_rate, initial)
228        except ImportError:
229            # Graceful degradation in test environments
230            self._slew_limiter = None
231
232    def _rebind(self, accessor: Callable[[], float]) -> None:
233        """Swap the raw accessor for dynamic remapping."""
234        self._accessor = accessor

A managed analog input with full shaping pipeline.

Args: action: The ActionDefinition this analog represents (may be None). accessor: Callable returning the raw axis value each cycle. default_value: Value returned when no accessor is bound.

ManagedAnalog( action: utils.controller.ActionDefinition | None, accessor: Callable[[], float], default_value: float = 0.0) View Source

 70    def __init__(
 71        self,
 72        action: ActionDefinition | None,
 73        accessor: Callable[[], float],
 74        default_value: float = 0.0,
 75    ):
 76        self._action = action
 77        self._accessor = accessor
 78        self._default_value = default_value
 79
 80        # Mutable properties — initialized from action or defaults
 81        if action is not None:
 82            self._deadband = action.deadband
 83            self._inversion = action.inversion
 84            self._scale = action.scale
 85            self._slew_rate = action.slew_rate
 86            self._trigger_mode = action.trigger_mode
 87            self._extra = action.extra or {}
 88        else:
 89            self._deadband = 0.0
 90            self._inversion = False
 91            self._scale = 1.0
 92            self._slew_rate = 0.0
 93            self._trigger_mode = EventTriggerMode.RAW
 94            self._extra = {}
 95
 96        # Binding info — set by the factory after construction.
 97        # Format: "ControllerName.input_name (channel)"
 98        self._binding_info: str = "unbound"
 99
100        self._init_nt_mapping()
101
102        self._pipeline: Callable[[float], float] = lambda x: x
103        self._slew_limiter = None
104        self._rebuild_pipeline()
105        self._rebuild_slew_limiter()

deadband: float View Source

109    @property
110    def deadband(self) -> float:
111        return self._deadband

inversion: bool View Source

119    @property
120    def inversion(self) -> bool:
121        return self._inversion

scale: float View Source

129    @property
130    def scale(self) -> float:
131        return self._scale

slew_rate: float View Source

139    @property
140    def slew_rate(self) -> float:
141        """Max output change rate in units/sec.  0 = disabled.
142
143        Applies symmetrically (same limit up and down) unless
144        ``extra["negative_slew_rate"]`` is set on the action.
145        """
146        return self._slew_rate

Max output change rate in units/sec. 0 = disabled.

Applies symmetrically (same limit up and down) unless extra["negative_slew_rate"] is set on the action.

action: utils.controller.ActionDefinition | None View Source

154    @property
155    def action(self) -> ActionDefinition | None:
156        return self._action

def get(self) -> float: View Source

160    def get(self) -> float:
161        """Return the shaped value through the full pipeline + slew limit."""
162        shaped = self._pipeline(self._accessor())
163        if self._slew_limiter is not None:
164            shaped = self._slew_limiter.calculate(shaped)
165        self._last_slew_output = shaped
166        return shaped

Return the shaped value through the full pipeline + slew limit.

def getRaw(self) -> float: View Source

168    def getRaw(self) -> float:
169        """Return the unmodified raw HID value."""
170        return self._accessor()

Return the unmodified raw HID value.

class ManagedRumble: View Source

14class ManagedRumble:
15    """A managed rumble output channel.
16
17    Args:
18        action: The ActionDefinition this output represents (may be None).
19        setter: Callable that sets the rumble intensity (0.0 to 1.0).
20    """
21
22    def __init__(
23        self,
24        action: ActionDefinition | None,
25        setter: Callable[[float], None],
26    ):
27        self._action = action
28        self._setter = setter
29        self._stop_time: float | None = None
30        self._current_value: float = 0.0
31
32        # Binding info — set by the factory after construction.
33        # Format: "ControllerName.input_name (channel)"
34        self._binding_info: str = "unbound"
35
36    @property
37    def action(self) -> ActionDefinition | None:
38        return self._action
39
40    def set(self, value: float, timeout: float = 0.0) -> None:
41        """Set rumble intensity.
42
43        Args:
44            value: Intensity from 0.0 to 1.0.
45            timeout: If > 0, auto-stop after this many seconds.
46        """
47        self._current_value = value
48        self._setter(value)
49        if timeout > 0:
50            self._stop_time = time.monotonic() + timeout
51        else:
52            self._stop_time = None
53
54    def stop(self) -> None:
55        """Stop the rumble immediately."""
56        self._current_value = 0.0
57        self._setter(0.0)
58        self._stop_time = None
59
60    def update(self) -> None:
61        """Check timeouts — call once per robot cycle."""
62        if (self._stop_time is not None
63                and time.monotonic() >= self._stop_time):
64            self.stop()
65
66    def __str__(self) -> str:
67        name = self._action.qualified_name if self._action else "unbound"
68        return f"ManagedRumble('{name}', bind={self._binding_info})"
69
70    def _rebind(self, setter: Callable[[float], None]) -> None:
71        """Swap the output setter for dynamic remapping."""
72        self._setter = setter

A managed rumble output channel.

Args: action: The ActionDefinition this output represents (may be None). setter: Callable that sets the rumble intensity (0.0 to 1.0).

ManagedRumble( action: utils.controller.ActionDefinition | None, setter: Callable[[float], NoneType]) View Source

22    def __init__(
23        self,
24        action: ActionDefinition | None,
25        setter: Callable[[float], None],
26    ):
27        self._action = action
28        self._setter = setter
29        self._stop_time: float | None = None
30        self._current_value: float = 0.0
31
32        # Binding info — set by the factory after construction.
33        # Format: "ControllerName.input_name (channel)"
34        self._binding_info: str = "unbound"

action: utils.controller.ActionDefinition | None View Source

36    @property
37    def action(self) -> ActionDefinition | None:
38        return self._action

def set(self, value: float, timeout: float = 0.0) -> None: View Source

40    def set(self, value: float, timeout: float = 0.0) -> None:
41        """Set rumble intensity.
42
43        Args:
44            value: Intensity from 0.0 to 1.0.
45            timeout: If > 0, auto-stop after this many seconds.
46        """
47        self._current_value = value
48        self._setter(value)
49        if timeout > 0:
50            self._stop_time = time.monotonic() + timeout
51        else:
52            self._stop_time = None

Set rumble intensity.

Args: value: Intensity from 0.0 to 1.0. timeout: If > 0, auto-stop after this many seconds.

def stop(self) -> None: View Source

54    def stop(self) -> None:
55        """Stop the rumble immediately."""
56        self._current_value = 0.0
57        self._setter(0.0)
58        self._stop_time = None

Stop the rumble immediately.

def update(self) -> None: View Source

60    def update(self) -> None:
61        """Check timeouts — call once per robot cycle."""
62        if (self._stop_time is not None
63                and time.monotonic() >= self._stop_time):
64            self.stop()

Check timeouts — call once per robot cycle.

@dataclass

class ValidationIssue: View Source

21@dataclass
22class ValidationIssue:
23    """A single validation finding."""
24    level: str       # "error" or "warning"
25    message: str
26    context: str = ""
27
28    def __str__(self) -> str:
29        prefix = f"[{self.level.upper()}]"
30        if self.context:
31            return f"{prefix} {self.context}: {self.message}"
32        return f"{prefix} {self.message}"

A single validation finding.

ValidationIssue(level: str, message: str, context: str = '')

level: str

message: str

context: str = ''

def validate_config( config: utils.controller.FullConfig) -> list[ValidationIssue]: View Source

142def validate_config(config: FullConfig) -> list[ValidationIssue]:
143    """Validate a full controller configuration.
144
145    Returns a list of issues (errors and warnings).
146    An empty list means the config is valid.
147    """
148    issues: list[ValidationIssue] = []
149
150    # Validate each action definition
151    for action in config.actions.values():
152        issues.extend(_validate_action(action))
153
154    # Validate controller bindings
155    for port, ctrl in config.controllers.items():
156        ctrl_ctx = f"controller {port} ({ctrl.name or 'unnamed'})"
157
158        for input_name, action_names in ctrl.bindings.items():
159            # Check input name is recognized
160            if input_name not in ALL_INPUT_NAMES:
161                issues.append(ValidationIssue(
162                    "error",
163                    f"Unknown input name '{input_name}'",
164                    ctrl_ctx))
165
166            category = get_input_category(input_name)
167
168            for action_name in action_names:
169                # Check action exists
170                if action_name not in config.actions:
171                    issues.append(ValidationIssue(
172                        "error",
173                        f"Binding references unknown action "
174                        f"'{action_name}'",
175                        f"{ctrl_ctx} / {input_name}"))
176                    continue
177
178                action = config.actions[action_name]
179
180                # Type consistency: action input_type vs physical input
181                if (action.input_type == InputType.ANALOG
182                        and category == "button"):
183                    issues.append(ValidationIssue(
184                        "warning",
185                        f"Analog action '{action_name}' bound to "
186                        f"button input '{input_name}'",
187                        ctrl_ctx))
188
189                if (action.input_type == InputType.BUTTON
190                        and category == "axis"):
191                    issues.append(ValidationIssue(
192                        "warning",
193                        f"Button action '{action_name}' bound to "
194                        f"axis input '{input_name}'",
195                        ctrl_ctx))
196
197                if (action.input_type == InputType.BOOLEAN_TRIGGER
198                        and category != "axis"):
199                    issues.append(ValidationIssue(
200                        "error",
201                        f"BOOLEAN_TRIGGER action '{action_name}' must "
202                        f"be bound to an axis input, got '{input_name}'",
203                        ctrl_ctx))
204
205                if (action.input_type == InputType.VIRTUAL_ANALOG
206                        and category not in ("button", "pov")):
207                    issues.append(ValidationIssue(
208                        "error",
209                        f"VIRTUAL_ANALOG action '{action_name}' must "
210                        f"be bound to a button/POV input, "
211                        f"got '{input_name}'",
212                        ctrl_ctx))
213
214                if (action.input_type == InputType.OUTPUT
215                        and category != "output"):
216                    issues.append(ValidationIssue(
217                        "warning",
218                        f"Output action '{action_name}' bound to "
219                        f"non-output input '{input_name}'",
220                        ctrl_ctx))
221
222    # Warn about empty action definitions
223    for action in config.actions.values():
224        if (not action.description
225                and action.input_type == InputType.BUTTON
226                and action.trigger_mode == EventTriggerMode.ON_TRUE
227                and action.deadband == 0.0
228                and action.scale == 1.0
229                and not action.extra):
230            issues.append(ValidationIssue(
231                "warning",
232                "Action has all default values (possibly empty definition)",
233                action.qualified_name))
234
235    return issues

Validate a full controller configuration.

Returns a list of issues (errors and warnings). An empty list means the config is valid.