pamoja.kit

Idiomatic helper-math facade.

The helpers are named for the goal rather than the technique, with the real algorithm one layer down: smooth a noisy reading, hold a value with a PID, warn before a tank runs dry, notice when a tracked point leaves its area, and drive a robot: chassis kinematics, an arm, odometry, waypoint guidance, and the safety gate every motion command passes through.

They are synchronous and allocation-free in the core, so this module re-exports most of the generated classes rather than wrapping them. What it adds is a Coordinate for the geo helpers, so a fix travels as one value instead of a pair of loose floats, and the Boundary, Edge, and Elbow enums for the crossing states a Geofence and a Trigger report and the branch a TwoLinkArm solves for.

A reading that is not a finite number, such as the NaN a failed sensor reports, is ignored by every helper that keeps state, so one bad reading cannot poison an average or an integral. Anomaly flags such a reading instead, and the motion helpers stop or hold rather than move on one.

  1"""Idiomatic helper-math facade.
  2
  3The helpers are named for the goal rather than the technique, with the real
  4algorithm one layer down: smooth a noisy reading, hold a value with a PID, warn
  5before a tank runs dry, notice when a tracked point leaves its area, and drive a
  6robot: chassis kinematics, an arm, odometry, waypoint guidance, and the safety
  7gate every motion command passes through.
  8
  9They are synchronous and allocation-free in the core, so this module re-exports
 10most of the generated classes rather than wrapping them. What it adds is a
 11:class:`Coordinate` for the geo helpers, so a fix travels as one value instead of
 12a pair of loose floats, and the :class:`Boundary`, :class:`Edge`, and
 13:class:`Elbow` enums for the crossing states a :class:`Geofence` and a
 14:class:`Trigger` report and the branch a :class:`TwoLinkArm` solves for.
 15
 16A reading that is not a finite number, such as the NaN a failed sensor reports,
 17is ignored by every helper that keeps state, so one bad reading cannot poison an
 18average or an integral. :class:`Anomaly` flags such a reading instead, and the
 19motion helpers stop or hold rather than move on one.
 20"""
 21
 22from __future__ import annotations
 23
 24import enum
 25from typing import NamedTuple
 26
 27from pamoja._native import Anomaly, Median, Trend, Window
 28from pamoja._native import window_capacity as _window_capacity
 29from pamoja._native import Calibration, Debounce, Depletion, Kalman, Pid, Ramp, Smoother, Surge, Thermostat
 30from pamoja._native import Complementary, Tilt, dew_point, tilt_from_accel
 31from pamoja._native import (
 32    celsius_to_fahrenheit,
 33    celsius_to_kelvin,
 34    fahrenheit_to_celsius,
 35    hectopascals_to_pascals,
 36    kelvin_to_celsius,
 37    kilopascals_to_pascals,
 38    pascals_to_hectopascals,
 39    pascals_to_kilopascals,
 40    pascals_to_psi,
 41    percent_to_ratio,
 42    psi_to_pascals,
 43    ratio_to_percent,
 44)
 45from pamoja._native import (
 46    Ackermann,
 47    DhParameters,
 48    DiffDrive,
 49    EStop,
 50    Esc,
 51    Guidance,
 52    Limits,
 53    Mecanum,
 54    Odometry,
 55    Pose,
 56    Quadrature,
 57    QuadratureScale,
 58    SafetyGate,
 59    ServoMap,
 60    SkidSteer,
 61    Transform,
 62    Twist,
 63    TwoLinkArm,
 64    Watchdog,
 65    WaypointFollower,
 66    WheelSpeeds,
 67    forward_kinematics,
 68    obstacle_stop,
 69)
 70from pamoja._native import Geofence as _NativeGeofence
 71from pamoja._native import Trigger as _NativeTrigger
 72from pamoja._native import bearing_between as _bearing_between
 73from pamoja._native import deadband
 74from pamoja._native import distance_between as _distance_between
 75
 76#: How many readings a windowed helper keeps.
 77WINDOW_CAPACITY = _window_capacity()
 78
 79__all__ = [
 80    "Window",
 81    "WINDOW_CAPACITY",
 82    "Trend",
 83    "Median",
 84    "Anomaly",
 85    "Boundary",
 86    "Calibration",
 87    "Complementary",
 88    "Coordinate",
 89    "Debounce",
 90    "Depletion",
 91    "Geofence",
 92    "Kalman",
 93    "Pid",
 94    "Ramp",
 95    "Smoother",
 96    "Surge",
 97    "Thermostat",
 98    "Tilt",
 99    "Trigger",
100    "Edge",
101    "bearing_between",
102    "deadband",
103    "dew_point",
104    "distance_between",
105    "tilt_from_accel",
106    "celsius_to_fahrenheit",
107    "celsius_to_kelvin",
108    "fahrenheit_to_celsius",
109    "hectopascals_to_pascals",
110    "kelvin_to_celsius",
111    "kilopascals_to_pascals",
112    "pascals_to_hectopascals",
113    "pascals_to_kilopascals",
114    "pascals_to_psi",
115    "percent_to_ratio",
116    "psi_to_pascals",
117    "ratio_to_percent",
118    "Ackermann",
119    "DhParameters",
120    "DiffDrive",
121    "EStop",
122    "Elbow",
123    "Esc",
124    "Guidance",
125    "Limits",
126    "Mecanum",
127    "Odometry",
128    "Pose",
129    "Quadrature",
130    "QuadratureScale",
131    "SafetyGate",
132    "ServoMap",
133    "SkidSteer",
134    "Transform",
135    "Twist",
136    "TwoLinkArm",
137    "Watchdog",
138    "WaypointFollower",
139    "WheelSpeeds",
140    "forward_kinematics",
141    "obstacle_stop",
142]
143
144
145class Elbow(str, enum.Enum):
146    """Which way a :class:`TwoLinkArm`'s elbow bends; both reach the same point."""
147
148    #: The elbow angle is positive, counter-clockwise.
149    UP = "up"
150    #: The elbow angle is negative, clockwise.
151    DOWN = "down"
152
153
154class Edge(str, enum.Enum):
155    """What a :class:`Trigger` reports when a reading changes its state."""
156
157    #: The reading just crossed the line: the condition became true.
158    SET = "set"
159    #: The reading just came back past the release band: the condition stopped holding.
160    CLEARED = "cleared"
161
162
163class Trigger:
164    """Fires once when a reading crosses a line, and not again until it has come back.
165
166    A trigger is a threshold with hysteresis that reports its edges: ``update``
167    answers :attr:`Edge.SET` on the reading that crosses the line,
168    :attr:`Edge.CLEARED` on the one that comes back past the release band, and
169    ``None`` for every reading in between, so an alert is sent once rather than on
170    every reading while the condition holds.
171
172    Example::
173
174        dry = Trigger.below(20.0, 2.0)
175        dry.update(25.0)  # None
176        dry.update(19.0)  # Edge.SET
177        dry.update(21.0)  # None, still inside the release band
178        dry.update(22.5)  # Edge.CLEARED
179    """
180
181    __slots__ = ("_native",)
182
183    def __init__(self, native: _NativeTrigger) -> None:
184        """Wrap a generated trigger; use :meth:`above` or :meth:`below` instead."""
185        self._native = native
186
187    @classmethod
188    def above(cls, threshold: float, hysteresis: float) -> Trigger:
189        """Create a trigger that fires when a reading rises above the line.
190
191        :param threshold: The line to watch.
192        :param hysteresis: How far below the line the reading must fall to clear.
193        :returns: A trigger that has not fired.
194        """
195        return cls(_NativeTrigger.above(threshold, hysteresis))
196
197    @classmethod
198    def below(cls, threshold: float, hysteresis: float) -> Trigger:
199        """Create a trigger that fires when a reading falls below the line.
200
201        :param threshold: The line to watch.
202        :param hysteresis: How far above the line the reading must rise to clear.
203        :returns: A trigger that has not fired.
204        """
205        return cls(_NativeTrigger.below(threshold, hysteresis))
206
207    def update(self, reading: float) -> Edge | None:
208        """Feed a reading in and report the edge it caused.
209
210        :param reading: The latest reading. One that is not a finite number is ignored.
211        :returns: The edge, or ``None`` while nothing changed.
212        """
213        edge = self._native.update(reading)
214        return None if edge is None else Edge(edge)
215
216    @property
217    def is_set(self) -> bool:
218        """Whether the condition currently holds."""
219        return self._native.is_set
220
221    @property
222    def threshold(self) -> float:
223        """The line the trigger watches."""
224        return self._native.threshold
225
226    @property
227    def hysteresis(self) -> float:
228        """The release band on the far side of the line."""
229        return self._native.hysteresis
230
231    @property
232    def watches_above(self) -> bool:
233        """Whether the trigger watches a rising reading, as :meth:`above` makes it."""
234        return self._native.watches_above
235
236
237class Coordinate(NamedTuple):
238    """A latitude and longitude in degrees."""
239
240    #: Degrees north of the equator, negative for south.
241    latitude: float
242    #: Degrees east of the prime meridian, negative for west.
243    longitude: float
244
245
246class Boundary(str, enum.Enum):
247    """Where a fix sits relative to a :class:`Geofence`, including a crossing."""
248
249    #: The fix is inside the fence and was inside before, or is the first fix inside.
250    INSIDE = "Inside"
251    #: The fix is outside the fence and was outside before, or is the first fix outside.
252    OUTSIDE = "Outside"
253    #: The fix just crossed from inside to outside: the moment to raise a breach alert.
254    EXITED = "Exited"
255    #: The fix just crossed from outside back inside.
256    ENTERED = "Entered"
257
258
259class Geofence:
260    """Keeps a tracked point inside an area, and notices when it leaves.
261
262    A fence is a center and a radius; feeding it successive fixes reports whether
263    each is inside or outside and, crucially, the single fix that crossed, so an
264    alert fires once on the crossing rather than on every fix while away.
265
266    Example::
267
268        pen = Geofence(Coordinate(-1.2921, 36.8219), 50.0)
269        pen.update(Coordinate(-1.2921, 36.8219))  # Boundary.INSIDE
270        pen.update(Coordinate(-1.2930, 36.8219))  # Boundary.EXITED
271    """
272
273    __slots__ = ("_native",)
274
275    def __init__(self, center: Coordinate, radius_m: float) -> None:
276        """Create a circular fence around a center fix.
277
278        :param center: The center of the fence.
279        :param radius_m: The fence radius, in meters.
280        """
281        self._native = _NativeGeofence(center.latitude, center.longitude, radius_m)
282
283    def update(self, point: Coordinate) -> Boundary:
284        """Feed a fix in and report where it sits, including a single crossing.
285
286        :param point: The latest fix.
287        :returns: The boundary state for this fix.
288        """
289        return Boundary(self._native.update(point.latitude, point.longitude))
290
291    def contains(self, point: Coordinate) -> bool:
292        """Report whether a fix lies inside, without recording a crossing.
293
294        :param point: The fix to test.
295        :returns: ``True`` if the fix is inside the fence.
296        """
297        return self._native.contains(point.latitude, point.longitude)
298
299
300def distance_between(origin: Coordinate, destination: Coordinate) -> float:
301    """Return the great-circle distance between two coordinates, in meters.
302
303    :param origin: The coordinate to measure from.
304    :param destination: The coordinate to measure to.
305    :returns: The distance in meters.
306    """
307    return _distance_between(
308        origin.latitude, origin.longitude, destination.latitude, destination.longitude
309    )
310
311
312def bearing_between(origin: Coordinate, destination: Coordinate) -> float:
313    """Return the initial bearing from one coordinate to another, in degrees.
314
315    :param origin: The coordinate to measure from.
316    :param destination: The coordinate to measure to.
317    :returns: The bearing in degrees, clockwise from north.
318    """
319    return _bearing_between(
320        origin.latitude, origin.longitude, destination.latitude, destination.longitude
321    )
class Window:

A rolling window of the most recent readings, with the stats over them.

def push(self, /, reading):

Adds a reading, dropping the oldest once the window is full.

def latest(self, /):

The most recent reading, or None while the window is empty.

def oldest(self, /):

The oldest reading still held, or None while the window is empty.

def mean(self, /):

The mean of the readings, or None while the window is empty.

def min(self, /):

The smallest reading, or None while the window is empty.

def max(self, /):

The largest reading, or None while the window is empty.

def range(self, /):

The spread between the smallest and largest readings, or None while empty.

def variance(self, /):

The population variance of the readings, 0 for one reading, or None while empty.

is_full

Whether the window holds as many readings as it keeps.

capacity

How many readings the window holds before it starts dropping.

WINDOW_CAPACITY = 32
class Trend:

Fits a line through recent readings, so a slow drift is visible before it matters.

def push(self, /, reading):

Adds a reading.

slope

The fitted slope in units per reading, or None without two readings.

A positive slope is a rising signal.

capacity

How many readings the estimator keeps.

class Median:

Rejects a single wild reading, where an average would let it pull the answer.

def update(self, /, reading):

Folds a reading in and returns the median of the window.

value

The current median, or None before the first reading.

capacity

How many readings the filter keeps.

class Anomaly:

Flags a reading that stands out from the ones before it.

def check(self, /, reading):

Folds a reading in and reports whether it stands out.

Nothing is flagged before two readings are held; from the third on, a reading can be, and one that is not a finite number always is.

capacity

How many readings the baseline keeps.

class Boundary(builtins.str, enum.Enum):
247class Boundary(str, enum.Enum):
248    """Where a fix sits relative to a :class:`Geofence`, including a crossing."""
249
250    #: The fix is inside the fence and was inside before, or is the first fix inside.
251    INSIDE = "Inside"
252    #: The fix is outside the fence and was outside before, or is the first fix outside.
253    OUTSIDE = "Outside"
254    #: The fix just crossed from inside to outside: the moment to raise a breach alert.
255    EXITED = "Exited"
256    #: The fix just crossed from outside back inside.
257    ENTERED = "Entered"

Where a fix sits relative to a Geofence, including a crossing.

INSIDE = <Boundary.INSIDE: 'Inside'>
OUTSIDE = <Boundary.OUTSIDE: 'Outside'>
EXITED = <Boundary.EXITED: 'Exited'>
ENTERED = <Boundary.ENTERED: 'Entered'>
class Calibration:

Turns a raw sensor count into the units the reading is actually in.

def linear(scale, offset):

Creates a calibration applying raw * scale + offset.

def two_point(raw_low, value_low, raw_high, value_high):

Creates a calibration fitted through two known reference points.

def apply(self, /, raw):

Converts a raw reading into calibrated units.

class Complementary:

Fuses a drifting rate, such as a gyroscope's, with a noisy absolute reading, such as an accelerometer's tilt, into one steady estimate.

def update(self, /, rate, absolute, dt):

Fuses a rate and an absolute reading over dt and returns the estimate. If any of the three is not a finite number, the update is ignored.

estimate

The current estimate.

class Coordinate(typing.NamedTuple):
238class Coordinate(NamedTuple):
239    """A latitude and longitude in degrees."""
240
241    #: Degrees north of the equator, negative for south.
242    latitude: float
243    #: Degrees east of the prime meridian, negative for west.
244    longitude: float

A latitude and longitude in degrees.

Coordinate(latitude: float, longitude: float)

Create new instance of Coordinate(latitude, longitude)

latitude: float

Alias for field number 0

longitude: float

Alias for field number 1

class Debounce:

Stops a flickering input from acting until it has settled.

def update(self, /, raw):

Feeds a raw reading in and returns the settled state.

state

The settled state.

class Depletion:

Warns before a falling level runs out, by projecting its rate of fall.

def update(self, /, level):

Records a level and returns the samples left before the threshold.

Returns 0 once the level is at or below the threshold, the first reading included, and None while the level is steady or rising, or on a first reading above it, when no rate of fall is known yet.

class Geofence:
260class Geofence:
261    """Keeps a tracked point inside an area, and notices when it leaves.
262
263    A fence is a center and a radius; feeding it successive fixes reports whether
264    each is inside or outside and, crucially, the single fix that crossed, so an
265    alert fires once on the crossing rather than on every fix while away.
266
267    Example::
268
269        pen = Geofence(Coordinate(-1.2921, 36.8219), 50.0)
270        pen.update(Coordinate(-1.2921, 36.8219))  # Boundary.INSIDE
271        pen.update(Coordinate(-1.2930, 36.8219))  # Boundary.EXITED
272    """
273
274    __slots__ = ("_native",)
275
276    def __init__(self, center: Coordinate, radius_m: float) -> None:
277        """Create a circular fence around a center fix.
278
279        :param center: The center of the fence.
280        :param radius_m: The fence radius, in meters.
281        """
282        self._native = _NativeGeofence(center.latitude, center.longitude, radius_m)
283
284    def update(self, point: Coordinate) -> Boundary:
285        """Feed a fix in and report where it sits, including a single crossing.
286
287        :param point: The latest fix.
288        :returns: The boundary state for this fix.
289        """
290        return Boundary(self._native.update(point.latitude, point.longitude))
291
292    def contains(self, point: Coordinate) -> bool:
293        """Report whether a fix lies inside, without recording a crossing.
294
295        :param point: The fix to test.
296        :returns: ``True`` if the fix is inside the fence.
297        """
298        return self._native.contains(point.latitude, point.longitude)

Keeps a tracked point inside an area, and notices when it leaves.

A fence is a center and a radius; feeding it successive fixes reports whether each is inside or outside and, crucially, the single fix that crossed, so an alert fires once on the crossing rather than on every fix while away.

Example::

pen = Geofence(Coordinate(-1.2921, 36.8219), 50.0)
pen.update(Coordinate(-1.2921, 36.8219))  # Boundary.INSIDE
pen.update(Coordinate(-1.2930, 36.8219))  # Boundary.EXITED
Geofence(center: Coordinate, radius_m: float)
276    def __init__(self, center: Coordinate, radius_m: float) -> None:
277        """Create a circular fence around a center fix.
278
279        :param center: The center of the fence.
280        :param radius_m: The fence radius, in meters.
281        """
282        self._native = _NativeGeofence(center.latitude, center.longitude, radius_m)

Create a circular fence around a center fix.

Parameters
  • center: The center of the fence.
  • radius_m: The fence radius, in meters.
def update(self, point: Coordinate) -> Boundary:
284    def update(self, point: Coordinate) -> Boundary:
285        """Feed a fix in and report where it sits, including a single crossing.
286
287        :param point: The latest fix.
288        :returns: The boundary state for this fix.
289        """
290        return Boundary(self._native.update(point.latitude, point.longitude))

Feed a fix in and report where it sits, including a single crossing.

Parameters
  • point: The latest fix. :returns: The boundary state for this fix.
def contains(self, point: Coordinate) -> bool:
292    def contains(self, point: Coordinate) -> bool:
293        """Report whether a fix lies inside, without recording a crossing.
294
295        :param point: The fix to test.
296        :returns: ``True`` if the fix is inside the fence.
297        """
298        return self._native.contains(point.latitude, point.longitude)

Report whether a fix lies inside, without recording a crossing.

Parameters
  • point: The fix to test. :returns: True if the fix is inside the fence.
class Kalman:

Estimates a true value from noisy readings, trusting the model and the sensor in proportion to how noisy each is.

def update(self, /, reading):

Folds a reading in and returns the new estimate.

estimate

The current estimate.

class Pid:

Holds a value at a setpoint by trading off present, past, and predicted error.

def update(self, /, setpoint, measurement, dt):

Advances the controller by one step and returns the control output.

def reset(self, /):

Clears the accumulated integral, the last error, and the last output.

class Ramp:

Limits how fast a value may change, so a load is never slammed.

def update(self, /, target):

Moves one step toward target and returns the new value.

def set(self, /, value):

Forces the value without rate limiting.

value

The current value.

class Smoother:

Smooths a noisy reading by weighting each new sample against the running value.

def update(self, /, sample):

Folds a sample in and returns the smoothed value.

def reset(self, /):

Clears the smoother back to its initial state.

value

The current value, or None before the first sample.

class Surge:

Notices a step change between successive readings, such as a burst pipe.

def rising(limit):

Creates a detector for rises of more than limit between readings.

def falling(limit):

Creates a detector for falls of more than limit between readings.

def update(self, /, value):

Feeds a value in and returns the size of a step past the limit, or None.

class Thermostat:

Switches a load on and off around a setpoint, with hysteresis to stop chatter.

def cooling(setpoint, hysteresis):

Creates a cooling thermostat, which switches on when the reading rises.

def heating(setpoint, hysteresis):

Creates a heating thermostat, which switches on when the reading falls.

def update(self, /, reading):

Feeds a reading in and returns whether the load should be on.

is_on

Whether the load is on, as the last reading left it.

class Tilt:

Roll and pitch, in degrees.

pitch

Rotation about the right axis, in degrees, from -90 to 90.

roll

Rotation about the forward axis, in degrees, from -180 to 180.

class Trigger:
164class Trigger:
165    """Fires once when a reading crosses a line, and not again until it has come back.
166
167    A trigger is a threshold with hysteresis that reports its edges: ``update``
168    answers :attr:`Edge.SET` on the reading that crosses the line,
169    :attr:`Edge.CLEARED` on the one that comes back past the release band, and
170    ``None`` for every reading in between, so an alert is sent once rather than on
171    every reading while the condition holds.
172
173    Example::
174
175        dry = Trigger.below(20.0, 2.0)
176        dry.update(25.0)  # None
177        dry.update(19.0)  # Edge.SET
178        dry.update(21.0)  # None, still inside the release band
179        dry.update(22.5)  # Edge.CLEARED
180    """
181
182    __slots__ = ("_native",)
183
184    def __init__(self, native: _NativeTrigger) -> None:
185        """Wrap a generated trigger; use :meth:`above` or :meth:`below` instead."""
186        self._native = native
187
188    @classmethod
189    def above(cls, threshold: float, hysteresis: float) -> Trigger:
190        """Create a trigger that fires when a reading rises above the line.
191
192        :param threshold: The line to watch.
193        :param hysteresis: How far below the line the reading must fall to clear.
194        :returns: A trigger that has not fired.
195        """
196        return cls(_NativeTrigger.above(threshold, hysteresis))
197
198    @classmethod
199    def below(cls, threshold: float, hysteresis: float) -> Trigger:
200        """Create a trigger that fires when a reading falls below the line.
201
202        :param threshold: The line to watch.
203        :param hysteresis: How far above the line the reading must rise to clear.
204        :returns: A trigger that has not fired.
205        """
206        return cls(_NativeTrigger.below(threshold, hysteresis))
207
208    def update(self, reading: float) -> Edge | None:
209        """Feed a reading in and report the edge it caused.
210
211        :param reading: The latest reading. One that is not a finite number is ignored.
212        :returns: The edge, or ``None`` while nothing changed.
213        """
214        edge = self._native.update(reading)
215        return None if edge is None else Edge(edge)
216
217    @property
218    def is_set(self) -> bool:
219        """Whether the condition currently holds."""
220        return self._native.is_set
221
222    @property
223    def threshold(self) -> float:
224        """The line the trigger watches."""
225        return self._native.threshold
226
227    @property
228    def hysteresis(self) -> float:
229        """The release band on the far side of the line."""
230        return self._native.hysteresis
231
232    @property
233    def watches_above(self) -> bool:
234        """Whether the trigger watches a rising reading, as :meth:`above` makes it."""
235        return self._native.watches_above

Fires once when a reading crosses a line, and not again until it has come back.

A trigger is a threshold with hysteresis that reports its edges: update answers Edge.SET on the reading that crosses the line, Edge.CLEARED on the one that comes back past the release band, and None for every reading in between, so an alert is sent once rather than on every reading while the condition holds.

Example::

dry = Trigger.below(20.0, 2.0)
dry.update(25.0)  # None
dry.update(19.0)  # Edge.SET
dry.update(21.0)  # None, still inside the release band
dry.update(22.5)  # Edge.CLEARED
Trigger(native: Trigger)
184    def __init__(self, native: _NativeTrigger) -> None:
185        """Wrap a generated trigger; use :meth:`above` or :meth:`below` instead."""
186        self._native = native

Wrap a generated trigger; use above() or below() instead.

@classmethod
def above(cls, threshold: float, hysteresis: float) -> Trigger:
188    @classmethod
189    def above(cls, threshold: float, hysteresis: float) -> Trigger:
190        """Create a trigger that fires when a reading rises above the line.
191
192        :param threshold: The line to watch.
193        :param hysteresis: How far below the line the reading must fall to clear.
194        :returns: A trigger that has not fired.
195        """
196        return cls(_NativeTrigger.above(threshold, hysteresis))

Create a trigger that fires when a reading rises above the line.

Parameters
  • threshold: The line to watch.
  • hysteresis: How far below the line the reading must fall to clear. :returns: A trigger that has not fired.
@classmethod
def below(cls, threshold: float, hysteresis: float) -> Trigger:
198    @classmethod
199    def below(cls, threshold: float, hysteresis: float) -> Trigger:
200        """Create a trigger that fires when a reading falls below the line.
201
202        :param threshold: The line to watch.
203        :param hysteresis: How far above the line the reading must rise to clear.
204        :returns: A trigger that has not fired.
205        """
206        return cls(_NativeTrigger.below(threshold, hysteresis))

Create a trigger that fires when a reading falls below the line.

Parameters
  • threshold: The line to watch.
  • hysteresis: How far above the line the reading must rise to clear. :returns: A trigger that has not fired.
def update(self, reading: float) -> Edge | None:
208    def update(self, reading: float) -> Edge | None:
209        """Feed a reading in and report the edge it caused.
210
211        :param reading: The latest reading. One that is not a finite number is ignored.
212        :returns: The edge, or ``None`` while nothing changed.
213        """
214        edge = self._native.update(reading)
215        return None if edge is None else Edge(edge)

Feed a reading in and report the edge it caused.

Parameters
  • reading: The latest reading. One that is not a finite number is ignored. :returns: The edge, or None while nothing changed.
is_set: bool
217    @property
218    def is_set(self) -> bool:
219        """Whether the condition currently holds."""
220        return self._native.is_set

Whether the condition currently holds.

threshold: float
222    @property
223    def threshold(self) -> float:
224        """The line the trigger watches."""
225        return self._native.threshold

The line the trigger watches.

hysteresis: float
227    @property
228    def hysteresis(self) -> float:
229        """The release band on the far side of the line."""
230        return self._native.hysteresis

The release band on the far side of the line.

watches_above: bool
232    @property
233    def watches_above(self) -> bool:
234        """Whether the trigger watches a rising reading, as :meth:`above` makes it."""
235        return self._native.watches_above

Whether the trigger watches a rising reading, as above() makes it.

class Edge(builtins.str, enum.Enum):
155class Edge(str, enum.Enum):
156    """What a :class:`Trigger` reports when a reading changes its state."""
157
158    #: The reading just crossed the line: the condition became true.
159    SET = "set"
160    #: The reading just came back past the release band: the condition stopped holding.
161    CLEARED = "cleared"

What a Trigger reports when a reading changes its state.

SET = <Edge.SET: 'set'>
CLEARED = <Edge.CLEARED: 'cleared'>
def bearing_between( origin: Coordinate, destination: Coordinate) -> float:
313def bearing_between(origin: Coordinate, destination: Coordinate) -> float:
314    """Return the initial bearing from one coordinate to another, in degrees.
315
316    :param origin: The coordinate to measure from.
317    :param destination: The coordinate to measure to.
318    :returns: The bearing in degrees, clockwise from north.
319    """
320    return _bearing_between(
321        origin.latitude, origin.longitude, destination.latitude, destination.longitude
322    )

Return the initial bearing from one coordinate to another, in degrees.

Parameters
  • origin: The coordinate to measure from.
  • destination: The coordinate to measure to. :returns: The bearing in degrees, clockwise from north.
def deadband(value, center, width):

Holds value at center while it stays within width either side, and passes it through unchanged once it is further out.

def dew_point(celsius, humidity_percent):

Computes the dew point, in degrees Celsius, from the air temperature in degrees Celsius and the relative humidity in percent.

def distance_between( origin: Coordinate, destination: Coordinate) -> float:
301def distance_between(origin: Coordinate, destination: Coordinate) -> float:
302    """Return the great-circle distance between two coordinates, in meters.
303
304    :param origin: The coordinate to measure from.
305    :param destination: The coordinate to measure to.
306    :returns: The distance in meters.
307    """
308    return _distance_between(
309        origin.latitude, origin.longitude, destination.latitude, destination.longitude
310    )

Return the great-circle distance between two coordinates, in meters.

Parameters
  • origin: The coordinate to measure from.
  • destination: The coordinate to measure to. :returns: The distance in meters.
def tilt_from_accel(ax, ay, az):

Computes roll and pitch from a three-axis accelerometer at rest; the reading's unit does not matter, since only the ratios between axes set the angles.

def celsius_to_fahrenheit(celsius):

Converts degrees Celsius to degrees Fahrenheit.

def celsius_to_kelvin(celsius):

Converts degrees Celsius to kelvin.

def fahrenheit_to_celsius(fahrenheit):

Converts degrees Fahrenheit to degrees Celsius.

def hectopascals_to_pascals(hectopascals):

Converts hectopascals to pascals.

def kelvin_to_celsius(kelvin):

Converts kelvin to degrees Celsius.

def kilopascals_to_pascals(kilopascals):

Converts kilopascals to pascals.

def pascals_to_hectopascals(pascals):

Converts pascals to hectopascals.

def pascals_to_kilopascals(pascals):

Converts pascals to kilopascals.

def pascals_to_psi(pascals):

Converts pascals to pounds per square inch.

def percent_to_ratio(percent):

Converts a percentage to a ratio from 0 to 1.

def psi_to_pascals(psi):

Converts pounds per square inch to pascals.

def ratio_to_percent(ratio):

Converts a ratio from 0 to 1 to a percentage.

class Ackermann:

Car-like steering: one steered axle and a driven axle a wheelbase apart.

def steering_angle(self, /, linear, angular):

Returns the steering angle, in radians, for a forward speed and a yaw rate, or 0 while stopped.

def yaw_rate(self, /, linear, steering):

Returns the yaw rate a forward speed and a steering angle, in radians, produce.

def turn_radius(self, /, steering):

Returns the turn radius of a steering angle, in meters, or inf with the wheels straight.

def curvature(self, /, steering):

Returns the path curvature of a steering angle, the reciprocal of the turn radius.

class DhParameters:

One joint of a serial arm in the Denavit-Hartenberg convention.

def transform(self, /):

Returns the homogeneous transform this joint makes.

alpha

The link twist about the common normal, in radians.

theta

The joint angle about the previous z axis, in radians.

d

The link offset along the previous z axis, in meters.

a

The link length along the common normal, in meters.

class DiffDrive:

Wheel speeds for a desired body motion, and the body motion measured wheel speeds make, for a robot that steers by spinning two wheels at different speeds.

def wheel_speeds(self, /, linear, angular):

Returns the (left, right) wheel speeds for a forward speed and a turn rate.

def body_motion(self, /, left, right):

Returns the (linear, angular) body motion measured wheel speeds make.

class EStop:

An emergency stop that latches until a person resets it.

def engage(self, /):

Engages the stop; it holds until reset.

def reset(self, /):

Clears the stop.

def gate(self, /, desired):

Returns desired while clear, or a zero twist while engaged.

is_engaged

Whether the stop is engaged.

class Elbow(builtins.str, enum.Enum):
146class Elbow(str, enum.Enum):
147    """Which way a :class:`TwoLinkArm`'s elbow bends; both reach the same point."""
148
149    #: The elbow angle is positive, counter-clockwise.
150    UP = "up"
151    #: The elbow angle is negative, clockwise.
152    DOWN = "down"

Which way a TwoLinkArm's elbow bends; both reach the same point.

UP = <Elbow.UP: 'up'>
DOWN = <Elbow.DOWN: 'down'>
class Esc:

An electronic speed controller's pulse widths from full reverse to full forward.

def bidirectional():

The common reversible controller: 1000, 1500, and 2000 microseconds.

def pulse(self, /, throttle):

Returns the pulse width for a throttle from -1 to 1, held to that range, or the neutral one for a throttle that is not a number.

max_us

The pulse width at full forward, in microseconds.

neutral_us

The pulse width at rest, in microseconds.

min_us

The pulse width at full reverse, in microseconds.

class Guidance:

The command toward a waypoint, with the geometry behind it.

twist

The body motion to drive.

distance_m

The distance left to the target, in meters.

heading_error_deg

The heading error to the target, in degrees, in (-180, 180].

arrived

Whether the target is within the arrival radius.

class Limits:

Speed and acceleration limits, with the motion they last allowed.

def apply(self, /, desired, dt):

Returns desired held to the speed limits and eased toward within the acceleration limits over dt.

def reset(self, /):

Forgets the motion last allowed, so the next command eases up from rest.

class Mecanum:

A four-wheel mecanum base, which drives, strafes, and turns at once.

def wheel_speeds(self, /, twist):

Returns the four wheel speeds for a body twist.

def body_motion(self, /, wheels):

Returns the body twist measured wheel speeds make.

class Odometry:

Tracks a robot's pose by adding up its motion.

def reset(self, /, pose):

Sets the estimate to a known pose.

def integrate(self, /, linear, angular, dt):

Adds a forward speed and yaw rate held for dt, and returns the new pose.

def integrate_wheels(self, /, left, right, drive):

Adds the distances two wheels rolled, through a differential drive, and returns the new pose.

def fuse_heading(self, /, measured, weight):

Nudges the heading toward an absolute measurement by weight, from 0 to 1.

pose

The pose so far.

class Pose:

Where a robot is and which way it faces.

y

Position along the world y axis, in meters.

x

Position along the world x axis, in meters.

theta

Heading from the world x axis, in radians, in (-pi, pi], positive counter-clockwise.

class Quadrature:

Counts a quadrature encoder's steps from its A and B channels.

def update(self, /, a, b):

Feeds the channel levels read now, and returns 1 or -1 for a step in either direction, or 0 for no change or a jump past a step.

def reset(self, /):

Sets the count back to 0, keeping the channel state last read.

count

The signed count of steps so far.

class QuadratureScale:

Turns encoder steps into the distance and speed of the wheel they turn with.

def distance(self, /, count):

Returns the distance, in meters, a wheel rolled for a step count.

def velocity(self, /, delta_count, dt):

Returns the speed, in meters per second, from the steps counted over dt, or 0 when dt is 0.

class SafetyGate:

The gate every motion command passes through: an e-stop, a watchdog, and limits.

def feed(self, /):

Feeds the watchdog; call it whenever a fresh command arrives.

def engage_estop(self, /):

Engages the latching e-stop.

def reset_estop(self, /):

Clears the e-stop.

def command(self, /, desired, dt):

Returns the command that is safe to drive: a zero twist while stopped, otherwise desired bounded by the limits.

is_stopped

Whether the gate is forcing a stop: its e-stop is engaged or its watchdog expired.

class ServoMap:

A hobby servo's pulse widths across its travel.

def standard():

The standard hobby servo: 1000 to 2000 microseconds over 180 degrees.

def pulse(self, /, angle_deg):

Returns the pulse width for an angle, held to the travel, or 0, no pulse, for an angle that is not a number.

def angle(self, /, pulse_us):

Returns the angle a pulse width sets, held to the pulse range.

max_us

The pulse width at full travel, in microseconds.

range_deg

The full travel, in degrees.

min_us

The pulse width at zero degrees, in microseconds.

class SkidSteer:

A tracked or four-wheel drive that turns by skidding, with its track widened by a slip factor.

def wheel_speeds(self, /, linear, angular):

Returns the (left, right) speeds for a forward speed and a yaw rate.

def body_motion(self, /, left, right):

Returns the (linear, angular) body motion measured side speeds make.

class Transform:

A 4x4 homogeneous transform: a rotation and a translation.

def identity():

The identity: no rotation and no translation.

def multiply(self, /, other):

Returns self * other, the transform that applies other and then this one.

elements

The sixteen elements, row-major.

position

Where this transform places the origin, its translation, as (x, y, z).

class Twist:

How fast a robot is asked to move.

vx

Forward speed along the x axis.

vy

Leftward speed along the y axis; zero for drives that cannot strafe.

omega

Yaw rate about the z axis, positive counter-clockwise.

class TwoLinkArm:

A planar arm of two links, with a closed-form inverse.

def tip(self, /, shoulder, elbow):

Returns the hand's (x, y) for a shoulder and an elbow angle, in radians.

def joints_for(self, /, x, y, elbow='up'):

Returns the (shoulder, elbow) angles that put the hand at a point, or None for a point out of reach, an arm with a link of no length, or a coordinate that is not a finite number. elbow is "up" or "down".

reach

The (min, max) the hand reaches from the shoulder.

class Watchdog:

A deadman timer that expires unless fed often enough.

def feed(self, /):

Feeds the watchdog, restarting its silence timer.

def update(self, /, dt):

Advances the timer by dt and returns whether it has expired; a dt that is not a finite number expires it until it is fed.

is_expired

Whether the watchdog has expired.

class WaypointFollower:

Steers toward a waypoint: pivot toward it, then drive, slowing as the heading error grows.

def guide(self, /, here, heading_deg, target):

Returns the command from a (latitude, longitude) position and a compass heading, in degrees clockwise from north, toward a target.

class WheelSpeeds:

The four wheel speeds of a mecanum base.

rear_right

The rear-right wheel's speed.

rear_left

The rear-left wheel's speed.

front_left

The front-left wheel's speed.

front_right

The front-right wheel's speed.

def forward_kinematics(joints):

Returns the transform from a serial arm's base to its tool, or the identity for an arm with no joints.

def obstacle_stop(twist, range_m, stop_distance_m):

Cuts forward and sideways motion when an obstacle is within stop_distance_m, keeping the turn. A range that is not a number counts as an obstacle.