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, and notice when a tracked point leaves its area.

They are synchronous and allocation-free in the core, so this module re-exports 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 a Boundary enum for the crossing states.

  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, and notice when a tracked point leaves its area.
  6
  7They are synchronous and allocation-free in the core, so this module re-exports
  8the generated classes rather than wrapping them. What it adds is a
  9:class:`Coordinate` for the geo helpers, so a fix travels as one value instead of
 10a pair of loose floats, and a :class:`Boundary` enum for the crossing states.
 11"""
 12
 13from __future__ import annotations
 14
 15import enum
 16from typing import NamedTuple
 17
 18from pamoja._native import Anomaly, Median, Trend, Window
 19from pamoja._native import window_capacity as _window_capacity
 20from pamoja._native import Calibration, Debounce, Depletion, Kalman, Pid, Ramp, Smoother, Surge, Thermostat
 21from pamoja._native import Geofence as _NativeGeofence
 22from pamoja._native import bearing_between as _bearing_between
 23from pamoja._native import deadband
 24from pamoja._native import distance_between as _distance_between
 25
 26#: How many readings a windowed helper keeps.
 27WINDOW_CAPACITY = _window_capacity()
 28
 29__all__ = [
 30    "Window",
 31    "WINDOW_CAPACITY",
 32    "Trend",
 33    "Median",
 34    "Anomaly",
 35    "Boundary",
 36    "Calibration",
 37    "Coordinate",
 38    "Debounce",
 39    "Depletion",
 40    "Geofence",
 41    "Kalman",
 42    "Pid",
 43    "Ramp",
 44    "Smoother",
 45    "Surge",
 46    "Thermostat",
 47    "bearing_between",
 48    "deadband",
 49    "distance_between",
 50]
 51
 52
 53class Coordinate(NamedTuple):
 54    """A latitude and longitude in degrees."""
 55
 56    #: Degrees north of the equator, negative for south.
 57    latitude: float
 58    #: Degrees east of the prime meridian, negative for west.
 59    longitude: float
 60
 61
 62class Boundary(str, enum.Enum):
 63    """Where a fix sits relative to a :class:`Geofence`, including a crossing."""
 64
 65    #: The fix is inside the fence and was inside before, or is the first fix inside.
 66    INSIDE = "Inside"
 67    #: The fix is outside the fence and was outside before, or is the first fix outside.
 68    OUTSIDE = "Outside"
 69    #: The fix just crossed from inside to outside: the moment to raise a breach alert.
 70    EXITED = "Exited"
 71    #: The fix just crossed from outside back inside.
 72    ENTERED = "Entered"
 73
 74
 75class Geofence:
 76    """Keeps a tracked point inside an area, and notices when it leaves.
 77
 78    A fence is a centre and a radius; feeding it successive fixes reports whether
 79    each is inside or outside and, crucially, the single fix that crossed, so an
 80    alert fires once on the crossing rather than on every fix while away.
 81
 82    Example::
 83
 84        pen = Geofence(Coordinate(-1.2921, 36.8219), 50.0)
 85        pen.update(Coordinate(-1.2921, 36.8219))  # Boundary.INSIDE
 86        pen.update(Coordinate(-1.2930, 36.8219))  # Boundary.EXITED
 87    """
 88
 89    __slots__ = ("_native",)
 90
 91    def __init__(self, center: Coordinate, radius_m: float) -> None:
 92        """Create a circular fence around a centre fix.
 93
 94        :param center: The centre of the fence.
 95        :param radius_m: The fence radius, in metres.
 96        """
 97        self._native = _NativeGeofence(center.latitude, center.longitude, radius_m)
 98
 99    def update(self, point: Coordinate) -> Boundary:
100        """Feed a fix in and report where it sits, including a single crossing.
101
102        :param point: The latest fix.
103        :returns: The boundary state for this fix.
104        """
105        return Boundary(self._native.update(point.latitude, point.longitude))
106
107    def contains(self, point: Coordinate) -> bool:
108        """Report whether a fix lies inside, without recording a crossing.
109
110        :param point: The fix to test.
111        :returns: ``True`` if the fix is inside the fence.
112        """
113        return self._native.contains(point.latitude, point.longitude)
114
115
116def distance_between(origin: Coordinate, destination: Coordinate) -> float:
117    """Return the great-circle distance between two coordinates, in metres.
118
119    :param origin: The coordinate to measure from.
120    :param destination: The coordinate to measure to.
121    :returns: The distance in metres.
122    """
123    return _distance_between(
124        origin.latitude, origin.longitude, destination.latitude, destination.longitude
125    )
126
127
128def bearing_between(origin: Coordinate, destination: Coordinate) -> float:
129    """Return the initial bearing from one coordinate to another, in degrees.
130
131    :param origin: The coordinate to measure from.
132    :param destination: The coordinate to measure to.
133    :returns: The bearing in degrees, clockwise from north.
134    """
135    return _bearing_between(
136        origin.latitude, origin.longitude, destination.latitude, destination.longitude
137    )
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 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.

def variance(self, /):

The variance of the readings, or None without enough of them.

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 shows before it matters.

def push(self, /, reading):

Adds a reading.

slope

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

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.

class Anomaly:

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

def check(self, /, reading):

Folds a reading in and reports whether it stands out.

class Boundary(builtins.str, enum.Enum): 
63class Boundary(str, enum.Enum):
64    """Where a fix sits relative to a :class:`Geofence`, including a crossing."""
65
66    #: The fix is inside the fence and was inside before, or is the first fix inside.
67    INSIDE = "Inside"
68    #: The fix is outside the fence and was outside before, or is the first fix outside.
69    OUTSIDE = "Outside"
70    #: The fix just crossed from inside to outside: the moment to raise a breach alert.
71    EXITED = "Exited"
72    #: The fix just crossed from outside back inside.
73    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 Coordinate(typing.NamedTuple): 
54class Coordinate(NamedTuple):
55    """A latitude and longitude in degrees."""
56
57    #: Degrees north of the equator, negative for south.
58    latitude: float
59    #: Degrees east of the prime meridian, negative for west.
60    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 None while the level is steady or rising, and on the first reading, when no rate of fall is known yet.

class Geofence: 
 76class Geofence:
 77    """Keeps a tracked point inside an area, and notices when it leaves.
 78
 79    A fence is a centre and a radius; feeding it successive fixes reports whether
 80    each is inside or outside and, crucially, the single fix that crossed, so an
 81    alert fires once on the crossing rather than on every fix while away.
 82
 83    Example::
 84
 85        pen = Geofence(Coordinate(-1.2921, 36.8219), 50.0)
 86        pen.update(Coordinate(-1.2921, 36.8219))  # Boundary.INSIDE
 87        pen.update(Coordinate(-1.2930, 36.8219))  # Boundary.EXITED
 88    """
 89
 90    __slots__ = ("_native",)
 91
 92    def __init__(self, center: Coordinate, radius_m: float) -> None:
 93        """Create a circular fence around a centre fix.
 94
 95        :param center: The centre of the fence.
 96        :param radius_m: The fence radius, in metres.
 97        """
 98        self._native = _NativeGeofence(center.latitude, center.longitude, radius_m)
 99
100    def update(self, point: Coordinate) -> Boundary:
101        """Feed a fix in and report where it sits, including a single crossing.
102
103        :param point: The latest fix.
104        :returns: The boundary state for this fix.
105        """
106        return Boundary(self._native.update(point.latitude, point.longitude))
107
108    def contains(self, point: Coordinate) -> bool:
109        """Report whether a fix lies inside, without recording a crossing.
110
111        :param point: The fix to test.
112        :returns: ``True`` if the fix is inside the fence.
113        """
114        return self._native.contains(point.latitude, point.longitude)

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

A fence is a centre 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) 
92    def __init__(self, center: Coordinate, radius_m: float) -> None:
93        """Create a circular fence around a centre fix.
94
95        :param center: The centre of the fence.
96        :param radius_m: The fence radius, in metres.
97        """
98        self._native = _NativeGeofence(center.latitude, center.longitude, radius_m)

Create a circular fence around a centre fix.

Parameters
  • center: The centre of the fence.
  • radius_m: The fence radius, in metres.
def update(self, point: Coordinate) -> Boundary: 
100    def update(self, point: Coordinate) -> Boundary:
101        """Feed a fix in and report where it sits, including a single crossing.
102
103        :param point: The latest fix.
104        :returns: The boundary state for this fix.
105        """
106        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: 
108    def contains(self, point: Coordinate) -> bool:
109        """Report whether a fix lies inside, without recording a crossing.
110
111        :param point: The fix to test.
112        :returns: ``True`` if the fix is inside the fence.
113        """
114        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 and last error.

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 at least limit between readings.

def falling(limit):

Creates a detector for falls of at least limit between readings.

def update(self, /, value):

Feeds a value in and returns the size of a qualifying step, 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 should currently be on.

def bearing_between( origin: Coordinate, destination: Coordinate) -> float: 
129def bearing_between(origin: Coordinate, destination: Coordinate) -> float:
130    """Return the initial bearing from one coordinate to another, in degrees.
131
132    :param origin: The coordinate to measure from.
133    :param destination: The coordinate to measure to.
134    :returns: The bearing in degrees, clockwise from north.
135    """
136    return _bearing_between(
137        origin.latitude, origin.longitude, destination.latitude, destination.longitude
138    )

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):

Suppresses movement within width of center, so noise does not act.

def distance_between( origin: Coordinate, destination: Coordinate) -> float: 
117def distance_between(origin: Coordinate, destination: Coordinate) -> float:
118    """Return the great-circle distance between two coordinates, in metres.
119
120    :param origin: The coordinate to measure from.
121    :param destination: The coordinate to measure to.
122    :returns: The distance in metres.
123    """
124    return _distance_between(
125        origin.latitude, origin.longitude, destination.latitude, destination.longitude
126    )

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

Parameters
  • origin: The coordinate to measure from.
  • destination: The coordinate to measure to. :returns: The distance in metres.