Sensor Code and Filtering

Programming for Robotics

Course 1 · Chapter 4 · Sensor Code and Filtering

Programming for Robotics 3 ended with a simulator in which the robot of Programming for Robotics 1 ran unchanged, and with two things the simulator made visible: a range reading that jitters by a few millimetres, and a wheel speed that flickers by one tick from pass to pass. Programming for Robotics 2 showed where such readings come from. This chapter is about what the program does with them. A sensor never delivers “the distance”. It delivers a number that is late, noisy, sometimes wild, sometimes missing and sometimes frozen, and the controller must be given something better. The chapter builds that something in code: a Sensor class that says whether a reading can be believed, a set of small filters, code that measures a noise record and chooses a filter from it, fusion of two sensors, line and encoder code, calibration, and checks that notice a dead sensor. It uses the formulas of Robotics Fundamentals 5 section 9 (noise and filtering), section 4 (line position) and section 10 (calibration), and of Robotics Fundamentals 6 section 2 (speed from counts), each tied to the code that implements it. Every filter is tested on the simulator, whose noise is at first the easy kind (Gaussian, no outliers); the chapter then breaks the sensor on purpose to show what the easy case hides.

What was run, and what was not
Every Python block in this chapter that has an output below it was run with CPython 3, in one session, in the order of the page, and the output shown is what it printed. You need two files from earlier chapters in the same session (or importable): robot.py, the Config, clamp and Robot code of Programming for Robotics 1 section 6, and sim.py, the simulator of Programming for Robotics 3 (its section 14 listing). The blocks without an output, together, are the new module filters.py; section 12 lists it in one piece, and it needs neither of the other two (only the demonstrations do). Every random number comes from a random.Random(seed), so your outputs match the ones shown. Several numbers are assumptions chosen for the demonstration, not measurements: the noise of 5 mm of the simulated range sensor, a 10 % rate of wrong echoes, an infrared sensor of the form k / (d + d0) with k = 0.27 V m and d0 = 0.02 m, the levels of the line sensors. Measure your own robot before trusting them. Two short blocks are marked as written from memory and were not run on hardware.

1. From a reading to a trustworthy number

Between the hardware and the decision there are four jobs, and a robot that does them in the right order, each in its own small piece of code, is much easier to debug than one that mixes them:

hardware: range_m() a number, or None, or an exception | v +------------------------------+ | Sensor | in range? a number at all? when was it taken? | read() -> value or None | counts the bad readings +------------------------------+ | valid readings only v +------------------------------+ | Filter (or a Pipeline) | a median against wild values, then an average | update(x) -> y | or an exponential filter against noise +------------------------------+ | a smoothed value v +------------------------------+ | Health | stuck? too many bad readings? too old? | update() -> fault name | -> None (the range cannot be trusted) when it is +------------------------------+ | a value, or None v the Robot of Programming for Robotics 1 section 6: None for too long means stop
  • Validate. Is it a number at all, and inside what the part can really measure? None, NaN and out-of-range values are not measurements. Say so once, in one place (section 2).
  • Filter. Reduce the noise and remove the wild values, at a known price in delay (sections 3 to 5).
  • Combine and convert. Fuse two sensors, turn raw counts into metres, a row of line sensors into a position (sections 6 to 9).
  • Watch. Notice a sensor that has stopped telling the truth, and turn that into the safe default of Programming for Robotics 1 section 11 (section 10).

The order matters. A filter placed before validation averages a NaN into every later output. A health check placed after a heavy filter sees a calm signal and cannot tell it from a frozen one. And every filter costs delay, so the least filtering that does the job is the right amount.

2. The Sensor class: a value, a flag, a timestamp

The pattern is a thin wrapper around whatever function reads the hardware. It holds the newest valid value, whether the latest read was valid, the time of the newest valid value, and from that its age and whether it is stale. It never raises because of the sensor: the bus error of Programming for Robotics 2 section 11 becomes an invalid reading, not a crash in the middle of the controller. Two policies for a number outside the sensor’s range are offered, because they mean different things. By default it is invalid (the part is lying: a 4.35 m reading from a sensor that stops at 4 m is not 4.35 m). With clamp=True it is limited to the range and flagged: useful when the physical meaning is “at least this far” or “at least this near”, as for a reading that is just under the minimum range.

# filters.py -- the sensor and filter code of Programming for Robotics 4 (standard library only) # Needs nothing from the robot or the simulator: the rigs below only use methods a simulator or a robot has. import math import random from collections import deque def is_number(x): """True for a real, finite number; False for None, NaN and infinity.""" return x is not None and not (isinstance(x, float) and (math.isnan(x) or math.isinf(x))) class Sensor: """One measured quantity with a validity flag, a timestamp, an age and a stale check. read_fn returns a number or None (or raises OSError); clock_fn returns seconds. A reading outside [low, high] is INVALID, unless clamp=True: then it is limited to the range and flagged.""" def __init__(self, read_fn, clock_fn, low, high, timeout_s, clamp=False, name="sensor"): self.read_fn = read_fn self.clock_fn = clock_fn self.low = low self.high = high self.timeout_s = timeout_s # a value older than this is stale self.clamp = clamp self.name = name self.value = None # the newest VALID value; it stays while readings fail, so check age_s() self.valid = False # was the latest read() valid? self.clamped = False # was the latest value limited to the range? self.t_s = None # when the newest valid value was taken self.reads = 0 # counters for the health checks self.bad = 0 # invalid reads, in total self.bad_run = 0 # invalid reads in a row def read(self): """Take one reading. Returns the value if it is valid, else None. Never raises because of the sensor.""" now = self.clock_fn() self.reads += 1 try: raw = self.read_fn() except OSError: # a bus error or a timeout inside the driver raw = None self.clamped = False if is_number(raw) and self.clamp: limited = min(self.high, max(self.low, raw)) self.clamped = limited != raw raw = limited self.valid = is_number(raw) and self.low <= raw <= self.high if self.valid: self.value, self.t_s, self.bad_run = raw, now, 0 return raw self.bad += 1 self.bad_run += 1 return None def age_s(self): """Seconds since the newest valid value (infinity if there never was one).""" return float("inf") if self.t_s is None else self.clock_fn() - self.t_s def stale(self): return self.age_s() > self.timeout_s class ManualClock: """A clock you move by hand: for tests and demonstrations.""" def __init__(self): self.t = 0.0 def now_s(self): return self.t

The class has no filtering and no thresholds of its own; it only answers four questions: what is the value, is it valid, how old is it, is it too old. The first block below feeds it a scripted list of raw readings, a plain one and a clamping one side by side, with a clock that moves 0.02 s per reading and a timeout of 0.05 s:

def make_script(values): """A read function that returns the values one by one; the text "bus" raises OSError.""" it = iter(values) def read(): v = next(it) if v == "bus": raise OSError("i2c timeout") return v return read raws = [1.234, None, float("nan"), 0.012, 4.35, 1.301, "bus", 1.299] clock = ManualClock() strict = Sensor(make_script(raws), clock.now_s, 0.02, 4.0, 0.05, name="strict") limited = Sensor(make_script(raws), clock.now_s, 0.02, 4.0, 0.05, clamp=True, name="clamping") print(" t (s) raw strict: returns valid age (s) stale | clamping: returns clamped") for raw in raws: a = strict.read() b = limited.read() print(" %.2f %-9s %-15s %-5s %-7.2f %-5s | %-17s %s" % (clock.t, raw, "None" if a is None else "%.3f" % a, strict.valid, strict.age_s(), strict.stale(), "None" if b is None else "%.3f" % b, limited.clamped)) clock.t += 0.02 print("strict: %d reads, %d invalid in total, %d invalid in a row at the end" % (strict.reads, strict.bad, strict.bad_run))
t (s) raw strict: returns valid age (s) stale | clamping: returns clamped 0.00 1.234 1.234 True 0.00 False | 1.234 False 0.02 None None False 0.02 False | None False 0.04 nan None False 0.04 False | None False 0.06 0.012 None False 0.06 True | 0.020 True 0.08 4.35 None False 0.08 True | 4.000 True 0.10 1.301 1.301 True 0.00 False | 1.301 False 0.12 bus None False 0.02 False | None False 0.14 1.299 1.299 True 0.00 False | 1.299 False strict: 8 reads, 5 invalid in total, 0 invalid in a row at the end

Read the table row by row. A None, a NaN and a bus error (OSError) are all invalid and give None; the age grows by 0.02 s with each of them, and the value becomes stale once the age passes the 0.05 s timeout (the 4th and 5th rows). The reading 0.012 is below the 0.02 m minimum: the plain sensor rejects it, the clamping sensor returns 0.020 and sets clamped. A later valid reading resets the age to zero and the run of bad readings to zero. The counters (bad, bad_run) are the raw material of section 10.

The range sensor against a real simulated world

The simulator’s range sensor answers None in two different situations (Programming for Robotics 3 section 6): nothing within the maximum range, and a reading that noise pushed outside the limits. A real HC-SR04 style sensor times out in the first case and may do the second. The code cannot tell which it was, and the choice of what to do is the robot’s: treat the first as “far” if the program knows the arena is smaller than the range, treat it as a failure otherwise. The block below uses two small test tools (place, which teleports the robot along x, a step in the world that a real robot could not make, and collect, which takes a list of readings) and a 6 m long arena, with the robot facing the far wall, so that some readings are beyond the 4 m limit:

def place(sim, x_m): """Test tool: teleport the robot along x (a step the real world cannot make), then let one pass of time go by.""" sim.model.x = x_m sim.advance(0.02) def collect(hw, n, dt_s=0.02): """n range readings, one per pass, from any hardware object that also has advance(dt_s).""" out = [] for _ in range(n): hw.advance(dt_s) out.append(hw.range_m()) return out
world = World(6.0, 2.0) # a long arena: the far wall is 6 m from the west wall sim = Simulator(world, seed=3, pose=(0.3, 1.0, 0.0)) front = Sensor(sim.range_m, sim.now_s, 0.02, 4.0, 0.11, name="front") print(" robot x (m) true range (m) read() valid what it means") for x in (0.3, 1.0, 2.5, 5.80, 5.90, 5.91): place(sim, x) true_m = sim.range_sensors["front"].true_range(sim.pose()) got = front.read() if true_m is None: meaning = "nothing within 4 m: no echo" elif got is None: meaning = "so close that the noise put it under 0.02 m" else: meaning = "a measurement" print(" %9.2f %-13s %-7s %-5s %s" % (x, "None" if true_m is None else "%.4f" % true_m, "None" if got is None else "%.4f" % got, front.valid, meaning)) print("age of the newest valid value: %.2f s, stale (timeout 0.11 s): %s" % (front.age_s(), front.stale()))
robot x (m) true range (m) read() valid what it means 0.30 None None False nothing within 4 m: no echo 1.00 None None False nothing within 4 m: no echo 2.50 3.4300 3.4268 True a measurement 5.80 0.1300 0.1267 True a measurement 5.90 0.0300 None False so close that the noise put it under 0.02 m 5.91 0.0200 None False so close that the noise put it under 0.02 m age of the newest valid value: 0.04 s, stale (timeout 0.11 s): False

At x = 0.3 m and 1.0 m the wall is 5.63 m and 4.93 m away: nothing in range, None. At 2.5 m and 5.8 m the sensor reads close to the truth (3.4268 against 3.4300, and 0.1267 against 0.1300: the noise, a few millimetres). At 5.90 m the true range is 0.03 m and the reading is None: the noise (the standard deviation is 5 mm) pushed it below 0.02 m, which the sensor reports as no measurement. This is the near end of the range, and it is why a minimum range matters: a robot that stops at a range of 0.05 m may be using readings that are partly None. Notice also the age at the end: after a None the sensor still holds the last valid value, and it is 0.04 s old; the stale flag (a 0.11 s timeout here, as in Programming for Robotics 1 section 6) is what turns a run of missing readings into a stop.

Run it on the real robot
The same Sensor wraps real hardware: its read_fn is the driver’s function (the Ultrasonic.read_m of Programming for Robotics 2 section 8, or an ADC read) and its clock_fn returns seconds. The block below is written from memory of the MicroPython documentation and has not been run on a board. The one trap is the clock: time.ticks_ms() wraps around, so a plain subtraction of two readings can come out negative; the clock below adds up wrap-safe differences instead (Programming for Robotics 1 section 3 explains ticks_diff).
# MicroPython on a Raspberry Pi Pico: from memory, NOT run on hardware, NOT run here (syntax-checked only) from machine import ADC, Pin import time adc = ADC(Pin(26)) # GP26 = ADC0: the output of an infrared distance sensor def ir_counts(): return adc.read_u16() >> 4 # the Pico reports 16 bits; shift to the 12-bit counts used above _last_ms = time.ticks_ms() _t_s = 0.0 def now_s(): """Seconds since start as a growing float. ticks_ms() wraps around, so add up the (wrap-safe) differences.""" global _last_ms, _t_s now = time.ticks_ms() _t_s += time.ticks_diff(now, _last_ms) / 1000 _last_ms = now return _t_s # call it at least once per wrap period (days): the loop does, every pass ir = Sensor(ir_counts, now_s, low=100, high=3900, timeout_s=0.11, name="ir") # near the rails the ADC saturates: invalid

3. Filters as small stateful classes

A filter is a function with a memory. Written as a class, the memory is explicit, the filter can be reset, copied, tested and logged, and a pipeline of filters is just a list. All the filters below share the interface of the Filter base class: update(x) takes one reading and returns the filtered value, a bad input (None, NaN or infinity) is ignored and the previous output is held (so one bad value cannot poison the state), and reset() forgets everything. The two first filters are the ones of Robotics Fundamentals 5 section 9.

The moving average, in constant time

The mean of the last N readings is yk = (xk + xk−1 + … + xk−N+1) / N. Computing that sum afresh each time costs N additions per update. But the sum changes by only two terms when a new reading arrives, so the update is yk = yk−1 + (xk − xk−N) / N: add the new, subtract the one that falls out. The oldest reading is kept in a ring buffer: a list of N slots and an index that goes round and round, so the slot about to be overwritten always holds the oldest value. Each update touches one slot and the running sum, in time that does not depend on N:

N = 5, i = 2 (the next write goes to slot 2: its value, 6, is the oldest) slot: 0 1 2 3 4 [ 7 ][ 8 ][ 6 ][ 9 ][ 5 ] total = 7 + 8 + 6 + 9 + 5 = 35, mean 7.0 ^ i a new reading 10 arrives: total = total + new - buf[i] = 35 + 10 - 6 = 39 buf[i] = 10, i = (i + 1) mod 5 = 3 [ 7 ][ 8 ][10 ][ 9 ][ 5 ] mean = 39 / 5 = 7.8 ^ i (now the oldest is 9)

While the buffer is still filling, the unused slots hold 0.0 and the count of slots in use is divided by instead of N, so the first outputs are the mean of what has arrived. One honest warning: a running sum accumulates rounding errors, slowly. resync() recomputes the sum from the buffer; the block of section 3 shows how small the drift is.

class Filter: """Base class: update(x) filters one reading. A bad reading (None, NaN, infinity) is ignored: the output is held.""" def __init__(self): self.value = None # the latest output; None until the first good reading def update(self, x): if is_number(x): self.value = self._step(x) return self.value def _step(self, x): raise NotImplementedError def reset(self): self.value = None class MovingAverage(Filter): """The mean of the last n readings, in O(1) per update: a ring buffer and a running sum.""" def __init__(self, n): Filter.__init__(self) self.n = n self.reset() def reset(self): self.value = None self.buf = [0.0] * self.n # the ring buffer self.i = 0 # where the next reading goes (the oldest one, once the buffer is full) self.count = 0 # how many slots are in use self.total = 0.0 # the running sum of the slots def _step(self, x): self.total += x - self.buf[self.i] # add the new reading, take the oldest out (0.0 while filling) self.buf[self.i] = x self.i = (self.i + 1) % self.n self.count = min(self.count + 1, self.n) return self.total / self.count def resync(self): """Recompute the sum from the buffer: stops rounding errors from piling up over millions of updates.""" self.total = sum(self.buf)

The median: one wild value never gets through

The median of N readings (N odd) is the middle one when they are sorted. A single wild reading is the largest or the smallest of the window, never the middle, so it is ignored completely, and a sharp step in the true value is passed on without being smeared. The price is a sort of N values per update (cheap for N of 3 to 9) and less smoothing of ordinary noise than an average. With N = 2m + 1 the median survives up to m wild values in any window.

class MedianFilter(Filter): """The middle value of the last n readings (n odd): one wild reading never gets through.""" def __init__(self, n): Filter.__init__(self) self.n = n self.buf = deque(maxlen=n) # drops the oldest reading by itself def reset(self): self.value = None self.buf.clear() def _step(self, x): self.buf.append(x) s = sorted(self.buf) m = len(s) // 2 return s[m] if len(s) % 2 else 0.5 * (s[m - 1] + s[m]) # an even count only happens while filling

The exponential filter, and its time constant

yk = yk−1 + α (xk − yk−1), with 0 < α ≤ 1: each new reading moves the output a fraction α of the way towards it. One stored number, one multiplication. It is the digital version of the RC low-pass filter. After a step of size S in the input the output has covered the fraction 1 − (1 − α)n of it after n samples, which crosses 63.2 % at about n = 1 / (−ln(1 − α)) samples, roughly (1 − α) / α. With a sample interval dt that is the time constant τ = dt (1 − α) / α, and solving for α gives the form to use when the time constant is what you know: α = dt / (τ + dt). A worked example: dt = 0.02 s and a wanted τ of 0.1 s give α = 0.02 / 0.12 = 0.1667; check: 0.02 × 0.8333 / 0.1667 = 0.100 s (the exact time constant, −dt / ln(1 − α) = 0.02 / 0.1823 = 0.1097 s, is 10 % larger: the rule of thumb is close but not exact when α is not small). The class can be built from either. It starts at the first reading, not at zero: starting at zero would drag every later output up from nothing.

class ExpFilter(Filter): """y = y + alpha (x - y): one number of memory. alpha = 1 passes the readings unchanged.""" def __init__(self, alpha): Filter.__init__(self) assert 0.0 < alpha <= 1.0, "alpha must be in (0, 1]" self.alpha = alpha @classmethod def from_tau(cls, tau_s, dt_s): """The filter whose time constant is tau_s at a sample interval dt_s: alpha = dt / (tau + dt).""" return cls(dt_s / (tau_s + dt_s)) def tau_s(self, dt_s): return dt_s * (1.0 - self.alpha) / self.alpha def _step(self, x): if self.value is None: return x # start at the first reading, not at zero return self.value + self.alpha * (x - self.value)

Robotics Fundamentals 5’s outlier table, from the classes

Robotics Fundamentals 5 section 9 compared the three filters on twelve readings of a wall 100 cm away, with one wild echo (130.0). The classes reproduce its table digit for digit, which is the first check that the code implements the formulas it is supposed to:

raw = [100.4, 99.7, 100.2, 99.5, 130.0, 100.1, 99.8, 100.6, 99.9, 100.3, 99.6, 100.0] # Robotics Fundamentals 5 section 9 ma3, med3, ema = MovingAverage(3), MedianFilter(3), ExpFilter(0.25) print(" k raw MA3 Med3 EMA(0.25)") for k, x in enumerate(raw, start=1): a, m, e = ma3.update(x), med3.update(x), ema.update(x) if k < 3: print(" %2d %7.1f - - %8.2f" % (k, x, e)) else: print(" %2d %7.1f %7.2f %7.1f %8.2f%s" % (k, x, a, m, e, " <-- the outlier arrives" if x == 130.0 else "")) print("mean of all 12: %.2f median of all 12: %.2f" % (sum(raw) / 12, (sorted(raw)[5] + sorted(raw)[6]) / 2)) ma = MovingAverage(8) for i in range(200000): ma.update(1000.0 + (i * 7919 % 1000) / 1000.0) drift = ma.total - sum(ma.buf) print("running-sum drift after 200000 updates: %.2e; after resync(): %.2e" % (drift, (ma.resync(), ma.total - sum(ma.buf))[1]))
k raw MA3 Med3 EMA(0.25) 1 100.4 - - 100.40 2 99.7 - - 100.23 3 100.2 100.10 100.2 100.22 4 99.5 99.80 99.7 100.04 5 130.0 109.90 100.2 107.53 <-- the outlier arrives 6 100.1 109.87 100.1 105.67 7 99.8 109.97 100.1 104.20 8 100.6 100.17 100.1 103.30 9 99.9 100.10 99.9 102.45 10 100.3 100.27 100.3 101.91 11 99.6 99.93 99.9 101.34 12 100.0 99.97 100.0 101.00 mean of all 12: 102.51 median of all 12: 100.05 running-sum drift after 200000 updates: -2.76e-08; after resync(): 0.00e+00

Look at the outlier at k = 5. The moving average of 3 jumps to 109.90 and stays wrong for three outputs, because the 130.0 is inside its window of 3. The exponential filter jumps by 7.5 and decays, each step keeping 75 % of the error. The median of 3 is never more than 0.3 off: a single wild value never becomes the middle one. The last line shows the drift of the running sum: after 200000 updates on numbers near 1000 the running sum of the 8-point average differs from a fresh sum by 3 × 10−8 (a relative error of about 10−11), and resync() brings it to zero. For a robot that runs for hours at 50 Hz that is harmless, and a resync every few thousand updates is one line if you want certainty.

Three small non-linear pieces: slew limiter, Schmitt trigger, debouncer

Three more filters are not about noise in a measured quantity but about what the program does with it. They are short, and each fixes a failure that the averaging filters do not:

  • Slew-rate limiter. The output follows its target but moves at most max_rate units per second. On a command it is a ramp (a smooth start: the ramping of motor commands in Programming for Robotics 5, planned; not yet written). On a measurement it is a spike guard: a robot cannot move faster than its top speed, so a range that jumps by metres in one pass is not believable, and the limiter lets it creep towards the new value instead of taking the jump. Its weakness: a real jump is followed slowly too.
  • Hysteresis (a Schmitt trigger). A yes/no decision from a noisy number, such as “obstacle near”, flips back and forth when the number hovers around the threshold. Two thresholds fix it: turn on at 0.25 m, turn off again only above 0.28 m. The gap between them should be a few times the noise (here 30 mm against a noise of 5 mm, six sigma).
  • Debouncer. A switch or a bumper contact bounces: it flickers on and off for a few passes before it settles. The debouncer changes its output only after N readings in a row disagree with it. The cost is N passes of delay on every real change; unlike a median it is for yes/no signals.
class SlewLimiter: """The output follows the target but moves at most max_rate per second.""" def __init__(self, max_rate): self.max_rate = max_rate # units per second self.value = None def update(self, target, dt_s): if not is_number(target): return self.value # a bad reading: hold if self.value is None: self.value = target else: step = self.max_rate * dt_s self.value += max(-step, min(step, target - self.value)) return self.value
class Schmitt: """A comparator with two thresholds, so a noisy reading near the limit cannot make the decision chatter. on_at < off_at: a 'low' detector (e.g. obstacle near): on when x <= on_at, off again when x >= off_at. on_at > off_at: a 'high' detector: on when x >= on_at, off again when x <= off_at.""" def __init__(self, on_at, off_at, state=False): self.on_at = on_at self.off_at = off_at self.state = state def update(self, x): if is_number(x): if self.on_at <= self.off_at: if x <= self.on_at: self.state = True elif x >= self.off_at: self.state = False else: if x >= self.on_at: self.state = True elif x <= self.off_at: self.state = False return self.state
class Debouncer: """A yes/no input that changes only after n readings in a row disagree with the current state.""" def __init__(self, n, state=False): self.n = n self.state = state self.run = 0 # readings in a row that disagree with the state def update(self, raw): if raw is None: return self.state # no reading: no change if bool(raw) == self.state: self.run = 0 else: self.run += 1 if self.run >= self.n: self.state = bool(raw) self.run = 0 return self.state

Last, a Pipeline chains filters: the output of each is the input of the next. The usual one is a median first (it removes the wild values, which would otherwise be smeared by the next stage) and an average or exponential filter second (it smooths what is left).

class Pipeline(Filter): """Filters in a row: the output of each is the input of the next (for example a median, then an exponential filter).""" def __init__(self, *stages): Filter.__init__(self) self.stages = stages def reset(self): self.value = None for stage in self.stages: stage.reset() def _step(self, x): for stage in self.stages: x = stage.update(x) return x

All four on synthetic signals, where the right answer is known:

# 1. a slew limiter on a command: 0 to 0.3 m/s with at most 0.6 m/s per second of change slew = SlewLimiter(0.6) slew.update(0.0, 0.02) # start at rest cmd = [round(slew.update(0.3, 0.02), 3) for _ in range(30)] print("slew ramp at passes 1, 5, 10, 15, 25:", [cmd[i] for i in (0, 4, 9, 14, 24)]) # 2. a slew limiter as a spike guard on a range: a robot cannot move faster than 1 m/s, so 0.02 m per pass at most guard = SlewLimiter(1.0) signal = [1.00, 1.00, 1.00, 2.80, 1.00, 1.00, 1.00, 1.00] print("spike guard:", [round(guard.update(x, 0.02), 2) for x in signal]) # 3. chatter: a noisy range near a stop threshold of 0.25 m, as a bare comparison and through a Schmitt trigger rng = random.Random(4) rs = [0.30 - 0.0005 * k + rng.gauss(0.0, 0.004) for k in range(200)] # slowly approaching: 0.30 down to 0.20 m naive = [r < 0.25 for r in rs] schmitt = Schmitt(0.25, 0.28) trig = [schmitt.update(r) for r in rs] flips = lambda seq: sum(1 for a, b in zip(seq, seq[1:]) if a != b) print("changes of the 'near' decision: bare comparison %d, Schmitt trigger (0.25 on, 0.28 off) %d" % (flips(naive), flips(trig))) # 4. a bouncing switch: ones and zeros for a few passes, then a real press bounce = [0, 0, 1, 0, 1, 1, 0, 1, 1, 1, 1, 1, 1, 1, 1, 0, 1, 1, 1, 0, 0, 1, 0, 0, 0, 0, 0, 0] deb = Debouncer(4) out = [int(deb.update(b)) for b in bounce] print("raw :", "".join(str(b) for b in bounce)) print("debounced:", "".join(str(b) for b in out))
slew ramp at passes 1, 5, 10, 15, 25: [0.012, 0.06, 0.12, 0.18, 0.3] spike guard: [1.0, 1.0, 1.0, 1.02, 1.0, 1.0, 1.0, 1.0] changes of the 'near' decision: bare comparison 9, Schmitt trigger (0.25 on, 0.28 off) 1 raw : 0010110111111110111001000000 debounced: 0000000000111111111111111000

The slew limiter turns a jump of 0.3 into a ramp of 0.012 per pass (0.6 m/s per second times 0.02 s): 0.060 after 5 passes, 0.180 after 15, the full 0.300 after 25. As a spike guard it turns a 1.8 m jump into 0.02 m (1.00 to 1.02), and the next pass returns to 1.00. The bare comparison of the noisy approach with the 0.25 m threshold changes its answer 9 times as the robot creeps from 0.30 m to 0.20 m, and the Schmitt trigger exactly once. The debouncer swallows the two single-pass blips at the start (0010110111) and follows the real press, 4 passes late (the raw sequence turns to 1 for good at pass 8, the debounced one at pass 11, hence 0000000000111…), then ignores the brief dips and releases only when the zeros really stay.

4. What filtering costs: delay

Every smoothing filter answers with a past, not a present: its output is the average of what the sensor said over the last few samples, so a changing quantity shows up late. The cost is a number you can compute, and the simulator can measure it, with the noise turned off so that the delay is seen alone.

The delay of a moving average. The N-point average gives each of the last N readings the weight 1/N. Their ages are 0, 1, …, N − 1 samples, whose mean is (N − 1) / 2. That mean age is the delay: for a signal changing at a steady rate, the output equals what the input was (N − 1) / 2 samples ago. With N = 5 it is 2 samples, 0.04 s at 50 Hz.

The delay of an exponential filter. The weight on a reading that is n samples old is α (1 − α)n, and the mean age is the sum of n α (1 − α)n, which is (1 − α) / α: the same number as τ / dt, the time constant in samples. For α = 0.25 it is 3 samples; for α = 1/3, 2 samples.

A useful consequence. Setting (1 − α) / α = (N − 1) / 2 gives α = 2 / (N + 1): the exponential filter that has the same delay as an N-point average. And since the exponential filter reduces the variance of independent noise by the factor α / (2 − α), this α gives (2 / (N + 1)) / (2 N / (N + 1)) = 1 / N, exactly the average’s. To first order the two are equivalent, and the exponential filter needs one number of memory instead of N. (The average forgets a reading completely after N samples; the exponential filter forgets it gradually and never quite completely, which matters when an outlier gets in.)

The function below is the delay of a filter on a step: step_response returns the fraction of a step in the range that the output has covered at each sample after the step, and the delay is the sum of the missing fractions, 1 − sk (the area between the step and the response; for a moving average sk = (k + 1) / N for k < N, and the sum of 1 − (k + 1) / N is (N − 1) / 2). A small block of formulas for the delay, the noise ratios and the failure probability of a median comes first, so that the measurements can be compared with them:

def ma_delay_samples(n): """The delay of an n-point moving average: (n - 1) / 2 samples.""" return (n - 1) / 2 def ema_delay_samples(alpha): """The delay of an exponential filter: (1 - alpha) / alpha samples = tau / dt.""" return (1.0 - alpha) / alpha def ma_noise_ratio(n): """Output sigma / input sigma of an n-point moving average on independent noise: 1 / sqrt(n).""" return 1.0 / math.sqrt(n) def ema_noise_ratio(alpha): """Output sigma / input sigma of an exponential filter on independent noise: sqrt(alpha / (2 - alpha)).""" return math.sqrt(alpha / (2.0 - alpha)) def median_failure_probability(n, p): """The chance that a window of n readings holds a majority of outliers (each with probability p): the median fails.""" need = (n + 1) // 2 return sum(math.comb(n, k) * p ** k * (1 - p) ** (n - k) for k in range(need, n + 1))
def step_response(make_filter, n_before=60, n_after=40): """The fraction of a step in the range that a filter's output has covered, for each sample after the step.""" sim = Simulator(World(3.0, 2.0), seed=1, pose=(1.0, 1.0, 0.0), sigma_m=0.0) # no noise: the lag shows alone f = make_filter() for _ in range(n_before): sim.advance(0.02) f.update(sim.range_m()) before = sim.range_m() place(sim, 1.5) # the wall suddenly 0.5 m nearer after = sim.range_m() levels = [(f.update(after) - before) / (after - before)] for _ in range(n_after - 1): sim.advance(0.02) levels.append((f.update(sim.range_m()) - before) / (after - before)) return levels candidates = [("MA 5", lambda: MovingAverage(5), ma_delay_samples(5)), ("MA 9", lambda: MovingAverage(9), ma_delay_samples(9)), ("Median 5", lambda: MedianFilter(5), 2.0), ("EMA alpha 0.25", lambda: ExpFilter(0.25), ema_delay_samples(0.25)), ("EMA alpha 1/3", lambda: ExpFilter(1 / 3), ema_delay_samples(1 / 3)), ("Median 5 + EMA 0.5", lambda: Pipeline(MedianFilter(5), ExpFilter(0.5)), 3.0)] print("filter theory (samples) measured = seconds at 50 Hz samples to 90 %") for name, make, theory in candidates: levels = step_response(make) delay = sum(1.0 - s for s in levels) # the area between the step and the response, in samples to90 = next(k for k, s in enumerate(levels) if s >= 0.9) + 1 print("%-19s %10.2f %6.2f %8.3f s %d" % (name, theory, delay, delay * 0.02, to90))
filter theory (samples) measured = seconds at 50 Hz samples to 90 % MA 5 2.00 2.00 0.040 s 5 MA 9 4.00 4.00 0.080 s 9 Median 5 2.00 2.00 0.040 s 3 EMA alpha 0.25 3.00 3.00 0.060 s 9 EMA alpha 1/3 2.00 2.00 0.040 s 6 Median 5 + EMA 0.5 3.00 3.00 0.060 s 6

Every measured delay equals the formula to two decimals: 2.00 samples for the 5-point average, 4.00 for the 9-point one, 3.00 for the exponential filter with α = 0.25 and 2.00 for α = 1/3, and the median of 5 also has a delay of 2 samples on a step (the output switches after 3 of the 5 samples in the window are new). The pipeline of a median of 5 and an exponential filter of 0.5 has 2 + (1 − 0.5) / 0.5 = 3 samples: the delays of the stages add. The last column is a different measure, the number of samples until the output has covered 90 % of the step: 5 for the average of 5, 3 for the median, 9 for the exponential filter with α = 0.25. The average is quicker to finish than the exponential filter of the same delay (it reaches 100 % after N samples, the other only approaches it), which is the price of the exponential filter’s long tail.

The shape of the response, for the two filters with the same delay of 2 samples (the average of 5 and α = 1/3) and a unit step:

def ascii_plot(series, rows=10, lo=0.0, hi=1.0, cell=3, first=0): """Draw equal-length series on a character grid. series = [(char, values), ...]; later ones draw over earlier ones.""" n = len(series[0][1]) grid = [[" "] * (n * cell) for _ in range(rows + 1)] for ch, values in series: for k, v in enumerate(values): r = int(round((hi - min(hi, max(lo, v))) / (hi - lo) * rows)) grid[r][k * cell] = ch lines = [] for r in range(rows + 1): level = hi - (hi - lo) * r / rows lines.append("%4.1f |%s" % (level, "".join(grid[r]).rstrip())) lines.append(" +" + "-" * (n * cell)) axis = [" "] * (n * cell) for k in range(n): if (first + k) % 5 == 0: for j, c in enumerate(str(first + k)): axis[k * cell + j] = c lines.append(" " + "".join(axis).rstrip() + " (samples after the step)") return "\n".join(lines)
inp = [0.0] * 3 + [1.0] * 14 ma5, ema3 = MovingAverage(5), ExpFilter(1 / 3) for _ in range(10): ma5.update(0.0) ema3.update(0.0) ma_out = [ma5.update(0.0) for _ in range(3)] + [ma5.update(1.0) for _ in range(14)] ema_out = [ema3.update(0.0) for _ in range(3)] + [ema3.update(1.0) for _ in range(14)] print(ascii_plot([("E", ema_out), ("M", ma_out), (".", inp)], first=-3)) print(" . the step M moving average of 5 E exponential filter, alpha = 1/3 (both delay 2 samples)")
1.0 | . . . . . . . . . . . . . . 0.9 | E E E 0.8 | M 0.7 | E 0.6 | E M 0.5 | 0.4 | M 0.3 | E 0.2 | M 0.1 | 0.0 |. . . +--------------------------------------------------- 0 5 10 (samples after the step) . the step M moving average of 5 E exponential filter, alpha = 1/3 (both delay 2 samples)

The moving average climbs in a straight line over exactly 5 samples and stops; the exponential filter jumps at once to a third of the step and then approaches the final value ever more slowly. Both have the area of 2 samples between them and the step, so both lag a ramp by 2 samples.

Delay costs distance

A filter delay of D samples means that the number the controller sees is D samples old. For a robot approaching a wall at speed v, the range it reads is too large by D × v × dt. The block below drives the simulated robot at the wall at a steady 0.3 m/s (duty 0.75), lets the motors settle and the filters fill for 1.5 s, and averages the error of the filtered range against the true range of the same sensor over the next 2.5 s:

def drive_lag(make_filter, seconds=4.0): """Mean error (filtered - true) of a range filter on a robot driving at the wall at a steady 0.3 m/s.""" sim = Simulator(World(3.0, 2.0), seed=2, pose=(0.3, 1.0, 0.0)) f = make_filter() sim.set_duty(0.75, 0.75) errs = [] for k in range(int(seconds / 0.02)): sim.advance(0.02) y = f.update(sim.range_m()) if k >= 75: # skip the first 1.5 s: the motors settle, the filters fill errs.append(y - sim.range_sensors["front"].true_range(sim.pose())) return sum(errs) / len(errs) v = 0.3 print("speed %.1f m/s; a filter with delay D samples reads the range D x v x dt = D x %.4f m too far (we approach the wall)" % (v, v * 0.02)) for name, make, delay in candidates: if name in ("MA 5", "MA 9", "Median 5", "EMA alpha 1/3", "Median 5 + EMA 0.5"): print("%-19s predicted %5.1f mm measured %5.1f mm" % (name, delay * v * 0.02 * 1000, drive_lag(make) * 1000))
speed 0.3 m/s; a filter with delay D samples reads the range D x v x dt = D x 0.0060 m too far (we approach the wall) MA 5 predicted 12.0 mm measured 12.2 mm MA 9 predicted 24.0 mm measured 24.2 mm Median 5 predicted 12.0 mm measured 12.2 mm EMA alpha 1/3 predicted 12.0 mm measured 12.2 mm Median 5 + EMA 0.5 predicted 18.0 mm measured 18.2 mm

The prediction and the measurement agree within 0.2 mm for every filter: a delay of 2 samples is 12 mm at 0.3 m/s, 4 samples are 24 mm, and the pipeline of 3 samples costs 18 mm. The extra 0.2 mm is the mean of the noise over the 125 samples (5 mm / √125 = 0.45 mm of standard error). 12 mm does not sound much, but it adds to the stop line: a robot that stops at a filtered range of 0.25 m has really stopped 12 mm short of it while it is still moving, and the delay of the motors adds the 2.4 cm of Programming for Robotics 3 section 5. When a controller needs the speed of a decision, count the delays of the whole chain, as in the latency chain of Programming for Robotics 1 section 12.

5. Choosing a filter from a measured noise record

The right filter is not a matter of taste. Three numbers decide it, all of which can be read off a record of the sensor looking at something that does not move: the noise (standard deviation of the ordinary scatter), the outlier rate (how often a reading is wild), and the delay the controller can afford. The record is taken like a measurement of a real sensor: park the robot facing a wall, collect a few thousand readings, and compare each with the true distance (which a real experiment gets from a tape measure).

Four statistics do the work, all from the standard library. The mean and the (sample) variance, s2 = sum (x − mean)2 / (n − 1), are the ordinary ones, but a single wild reading ruins both. So there is also a robust sigma, from the median absolute deviation (MAD): 1.4826 times the median of |x − median|. For Gaussian noise it equals the standard deviation, and wild values barely move it, because the median ignores them. An outlier is then a reading further than 4 robust sigmas from the median. And an ascii_histogram draws the shape of the scatter in the terminal:

def mean(values): return sum(values) / len(values) def variance(values): """The sample variance (divides by n - 1).""" m = mean(values) return sum((v - m) ** 2 for v in values) / (len(values) - 1) def std(values): return math.sqrt(variance(values)) def median(values): s = sorted(values) m = len(s) // 2 return s[m] if len(s) % 2 else 0.5 * (s[m - 1] + s[m]) def robust_sigma(values): """Sigma from the median absolute deviation: outliers cannot move it (1.4826 makes it equal sigma for Gaussian noise).""" m = median(values) return 1.4826 * median([abs(v - m) for v in values]) def outlier_rate(values, k=4.0): """The fraction of readings further than k robust sigmas from the median.""" m, s = median(values), robust_sigma(values) return sum(1 for v in values if abs(v - m) > k * s) / len(values) def ascii_histogram(values, lo, hi, bins, width=40): """One row per bin: the bin centre, a bar, the count. Values outside lo..hi are not drawn.""" counts = [0] * bins step = (hi - lo) / bins for v in values: if lo <= v < hi: counts[int((v - lo) / step)] += 1 peak = max(counts) or 1 return "\n".join("%7.2f |%-*s %d" % (lo + (i + 0.5) * step, width, "#" * int(round(c / peak * width)), c) for i, c in enumerate(counts))
world = World(3.0, 2.0) sim = Simulator(world, seed=1, pose=(2.0, 1.0, 0.0)) # parked, facing the east wall true_m = sim.range_sensors["front"].true_range(sim.pose()) record = collect(sim, 4000) # 80 s of readings at 50 Hz good = [r for r in record if r is not None] dev_mm = [(r - true_m) * 1000 for r in good] # the error of each reading, in mm print("true range %.4f m; %d readings, %d missing (None)" % (true_m, len(record), len(record) - len(good))) print("mean error %+.3f mm sigma %.3f mm robust sigma %.3f mm outlier rate (beyond 4 sigma) %.4f" % (mean(dev_mm), std(dev_mm), robust_sigma(dev_mm), outlier_rate(dev_mm))) print("error (mm) count") print(ascii_histogram(dev_mm, -17.5, 17.5, 14, width=36))
true range 0.9300 m; 4000 readings, 0 missing (None) mean error +0.095 mm sigma 4.995 mm robust sigma 5.235 mm outlier rate (beyond 4 sigma) 0.0000 error (mm) count -16.25 | 2 -13.75 |# 13 -11.25 |### 57 -8.75 |######### 183 -6.25 |################### 398 -3.75 |########################### 557 -1.25 |#################################### 754 1.25 |#################################### 754 3.75 |############################ 591 6.25 |#################### 415 8.75 |######### 185 11.25 |### 68 13.75 |# 19 16.25 | 3

The robot is parked 0.9300 m from the wall (the sensor is 7 cm ahead of the centre). The 4000 readings have a mean error of +0.095 mm (no bias: the simulated sensor is perfect on average), a standard deviation of 4.995 mm (the 5 mm that was put in), and a robust sigma of 5.235 mm. The robust one is an estimate with a scatter of about 2 % from 4000 readings (about 0.1 mm), so this one is two standard errors high: close enough, but not exactly the same number as the plain one. The outlier rate is 0.0000: no reading is more than 4 robust sigmas away, as expected from a Gaussian. The histogram is the bell curve of Programming for Robotics 3 section 6: 754 readings in each of the two middle bins, about two thirds within ±5 mm, almost none beyond 15 mm.

The noise falls as one over the square root of N, and what that assumes

Robotics Fundamentals 5 section 9 gave the standard error of a mean: σ / √N for N independent readings. For an exponential filter the factor is √(α / (2 − α)). Both can be checked on the same record:

sigma = std(good) print("single-reading sigma %.2f mm. Filter outputs on the same 4000 readings (first 30 skipped):" % (sigma * 1000)) print("filter measured sigma predicted ratio measured / predicted") for n in (4, 9, 16, 25): f = MovingAverage(n) out = [f.update(r) for r in good][30:] pred = sigma * ma_noise_ratio(n) print("MA %-2d %8.3f mm sigma/sqrt(N) = %.3f mm %.2f" % (n, std(out) * 1000, pred * 1000, std(out) / pred)) for alpha in (0.2, 2 / 17): f = ExpFilter(alpha) out = [f.update(r) for r in good][30:] pred = sigma * ema_noise_ratio(alpha) print("EMA alpha %.3f %8.3f mm sigma sqrt(a/(2-a)) = %.3f mm %.2f" % (alpha, std(out) * 1000, pred * 1000, std(out) / pred))
single-reading sigma 5.00 mm. Filter outputs on the same 4000 readings (first 30 skipped): filter measured sigma predicted ratio measured / predicted MA 4 2.520 mm sigma/sqrt(N) = 2.498 mm 1.01 MA 9 1.705 mm sigma/sqrt(N) = 1.665 mm 1.02 MA 16 1.301 mm sigma/sqrt(N) = 1.249 mm 1.04 MA 25 1.055 mm sigma/sqrt(N) = 0.999 mm 1.06 EMA alpha 0.200 1.702 mm sigma sqrt(a/(2-a)) = 1.665 mm 1.02 EMA alpha 0.118 1.294 mm sigma sqrt(a/(2-a)) = 1.249 mm 1.04

The measured spread is within 1 to 6 % of the prediction for every filter, a little above it because the output of a filter is estimated from fewer independent values than the 4000 readings (a window of 25 has only 160 independent windows in 4000 readings, so its estimate is good to about 5 %). The exponential filter with α = 0.2 behaves like the 9-point average (α = 2 / (9 + 1) = 0.2), and α = 2/17 like the 16-point one, as section 4 said. The gain is slow: the 25-point average cuts the noise to a fifth (1.06 mm) at a delay of 12 samples, a quarter of a second at 50 Hz.

The word independent hides a trap. The formula holds only if each reading is a new, independent measurement. A real range sensor measures every 25 ms or so, and a control loop that runs faster than the sensor reads the same value several times. The simulator can do the same (period_s, Programming for Robotics 3 section 6): here the sensor makes a new measurement every 0.1 s, so every value is read five times in a row by the 50 Hz loop:

sim = Simulator(World(3.0, 2.0), seed=1, pose=(2.0, 1.0, 0.0), period_s=0.1) # a NEW measurement only every 0.1 s = every 5 passes held = collect(sim, 4000) # so each value is read 5 times in a row held = [r for r in held if r is not None] f = MovingAverage(5) every_pass = [f.update(r) for r in held][30:] f = MovingAverage(5) fresh_only = [] previous = None for r in held: if r != previous: # a changed value is a new measurement (a real sensor would say so with a flag or a timestamp) fresh_only.append(f.update(r)) previous = r fresh_only = fresh_only[6:] print("%d readings, %d of them new measurements; sigma of the readings %.2f mm" % (len(held), len(fresh_only) + 6, std(held) * 1000)) print("MA 5 fed on every pass: sigma %.2f mm, ratio %.2f (independent readings would give 1/sqrt(5) = %.2f)" % (std(every_pass) * 1000, std(every_pass) / std(held), ma_noise_ratio(5))) print("MA 5 fed only new readings: sigma %.2f mm, ratio %.2f" % (std(fresh_only) * 1000, std(fresh_only) / std(held)))
4000 readings, 801 of them new measurements; sigma of the readings 4.81 mm MA 5 fed on every pass: sigma 3.99 mm, ratio 0.83 (independent readings would give 1/sqrt(5) = 0.45) MA 5 fed only new readings: sigma 2.16 mm, ratio 0.45

An average of 5 fed on every pass reduces the noise by only 0.83 (instead of 0.45): its window mostly holds copies of one measurement. Fed only the new measurements, it reaches 0.45 = 1 / √5. The lesson for the code: update a filter when there is a new reading, not when the loop runs. A real driver should return a “new data” flag or a timestamp, and the Sensor class could keep one (here, as in the demonstration, a changed value stands in for it, which is not safe on a real sensor that can repeat a value by chance).

Outliers: breaking the sensor on purpose

The simulator’s noise is Gaussian and has no outliers, which is the easy case. Real ultrasonic sensors have a second kind of error: an echo off the wrong object, a missed pulse, a spike from the motors, a reading of the right shape and the wrong size. Everything that matters about a filter shows up there. So the next piece of code breaks the sensor on purpose: FaultyHardware wraps the simulator, which is a hardware object (Programming for Robotics 1 section 7), and replaces a chosen fraction of the range readings by a wrong echo, uniform between 0.05 m and 3.0 m. It passes everything else (ticks, set_duty, now_s, advance) through unchanged. This is fault injection, and it needs no new simulator code. The same wrapper can freeze the sensor at a chosen time, which section 10 uses.

class FaultyHardware: """Wraps a hardware object and breaks its range readings on purpose (fault injection). Everything else passes through. spike_rate: the chance that a reading is replaced by a wrong echo, uniform in spike_lo_m .. spike_hi_m. stuck_after_s: from then on range_m() repeats the last value it gave, forever.""" def __init__(self, hw, rng, spike_rate=0.0, spike_lo_m=0.05, spike_hi_m=3.0, stuck_after_s=None): self.hw = hw self.rng = rng self.spike_rate = spike_rate self.spike_lo_m = spike_lo_m self.spike_hi_m = spike_hi_m self.stuck_after_s = stuck_after_s self._last = None def range_m(self): r = self.hw.range_m() chance = self.rng.random() # always draw both numbers: the random stream does not depend on what happened spike = self.rng.uniform(self.spike_lo_m, self.spike_hi_m) if self.stuck_after_s is not None and self.hw.now_s() >= self.stuck_after_s: return self._last if r is not None and chance < self.spike_rate: r = spike self._last = r return r def __getattr__(self, name): # ticks, set_duty, now_s, advance ... go straight through return getattr(self.hw, name)

The assumed rate is 10 %, higher than a good sensor in a quiet room shows, and the right order of magnitude for a cheap ultrasonic sensor among furniture (an assumption: measure yours):

sim = Simulator(World(3.0, 2.0), seed=1, pose=(2.0, 1.0, 0.0)) bad = FaultyHardware(sim, random.Random(11), spike_rate=0.10) dirty = collect(bad, 4000) dirty = [r for r in dirty if r is not None] dev = [(r - true_m) * 1000 for r in dirty] print("with 10 %% wrong echoes: mean error %+.1f mm, plain sigma %.1f mm, robust sigma %.2f mm" % (mean(dev), std(dev), robust_sigma(dev))) print("outlier rate beyond 4 robust sigma: %.4f (the injected rate is 0.1000, less the %.1f %% of spikes that land near the truth)" % (outlier_rate(dev), 100 * 0.04 / 2.95)) print("%d readings within +-17.5 mm of the truth (the same bell as before); %d outside it, from %+.0f mm to %+.0f mm" % (sum(1 for d in dev if abs(d) < 17.5), sum(1 for d in dev if abs(d) >= 17.5), min(dev), max(dev)))
with 10 % wrong echoes: mean error +58.8 mm, plain sigma 317.0 mm, robust sigma 5.87 mm outlier rate beyond 4 robust sigma: 0.0980 (the injected rate is 0.1000, less the 1.4 % of spikes that land near the truth) 3605 readings within +-17.5 mm of the truth (the same bell as before); 395 outside it, from -877 mm to +2059 mm

The mean error jumps to +58.8 mm and the plain standard deviation to 317 mm: both are ruined by the 395 wild readings. The robust sigma moves from 5.2 to 5.87 mm, a little, because 10 % of far-away values shifts the median of the deviations a few percent up the bell curve. The outlier rate beyond 4 robust sigmas is 0.098: the injected 10 %, less the 1.4 % of the wrong echoes that happen to land within a few millimetres of the truth (the spike range is 2.95 m wide and the ±20 mm of the truth is 0.04 m of it). A plain standard deviation on this record would have said the sensor is 60 times noisier than it is. That is why the robust numbers belong in the measuring code: they describe the part that a filter can smooth, and the outlier rate describes the part it must reject.

Which filter?

The block below scores six candidates on the same record, as the root-mean-square error against the true range, the largest error, and the number of outputs more than 20 mm off:

def score(make_filter, readings, truth_m, skip=30): f = make_filter() errs = [(f.update(r) - truth_m) * 1000 for r in readings][skip:] big = sum(1 for e in errs if abs(e) > 20.0) return math.sqrt(sum(e * e for e in errs) / len(errs)), max(abs(e) for e in errs), big print("the 4000 readings with 10 % wrong echoes, error against the true range (first 30 skipped)") print("filter rms (mm) largest (mm) outputs off by more than 20 mm delay (samples)") for name, make, delay in [("none (raw)", lambda: Pipeline(), 0.0)] + candidates: rms, big_e, count = score(make, dirty, true_m) print("%-22s %9.1f %11.1f %15d %6.1f" % (name, rms, big_e, count, delay)) p_fail = median_failure_probability(5, 0.10 * (1 - 0.04 / 2.95)) print("median of 5, 10 %% outliers: chance of 3 or more outliers in a window = %.4f, so at most about %.0f bad outputs in 3970 (measured: 23)" % (p_fail, p_fail * 3970))
the 4000 readings with 10 % wrong echoes, error against the true range (first 30 skipped) filter rms (mm) largest (mm) outputs off by more than 20 mm delay (samples) none (raw) 322.2 2058.6 389 0.0 MA 5 153.5 792.6 1482 2.0 MA 9 120.3 624.8 2109 4.0 Median 5 33.4 967.9 23 2.0 EMA alpha 0.25 133.2 834.9 2070 3.0 EMA alpha 1/3 153.4 1045.9 1815 2.0 Median 5 + EMA 0.5 23.2 546.1 60 3.0 median of 5, 10 % outliers: chance of 3 or more outliers in a window = 0.0082, so at most about 33 bad outputs in 3970 (measured: 23)

The averages and the exponential filters are worse than useless here: the 5-point average has an rms error of 153 mm, half the raw 322 mm, because each wild echo is spread over five outputs (1482 outputs are off by more than 20 mm, against 389 raw readings: the filter has made the problem 4 times as wide). The median of 5 removes it: an rms of 33 mm and only 23 bad outputs. Chaining the exponential filter after it smooths the 5 mm of ordinary noise: an rms of 23 mm and a noise floor around the 3 mm that the formula predicts (the median has already done part of the smoothing), at a delay of 3 samples. The 60 bad outputs, against 23 for the median alone, are the median’s own failures (a majority of wild values in a window) spread over several outputs by the exponential filter.

The last line is a check with the binomial formula. A median of N fails when a majority of its window is wild: with a wild-value probability p, the chance that at least (N + 1) / 2 of N are wild is the sum, over k from (N + 1) / 2 to N, of C(N, k) pk (1 − p)N−k. For N = 5 and p = 0.10 that is 10 × 0.001 × 0.81 + 5 × 0.0001 × 0.9 + 0.00001 = 0.0086 (0.0082 when p is the 0.0986 left after the spikes that land near the truth). So there should be up to 33 bad outputs in 3970, and 23 were measured. The formula is an upper bound: when the wild values fall on both sides of the truth, the middle value of the window can still be a good one. Use the same formula the other way round to choose N for a given outlier rate: for p = 1 % a median of 3 fails with probability 3 × 10−4, a median of 5 with 1 × 10−5.

The rules that come out of this:

  • Outliers present? Median first. Pick N from the outlier rate with the binomial formula and from the delay you can afford (the delay of a median is (N − 1) / 2 samples). A moving average or an exponential filter before the median, or alone, smears every wild value over several outputs.
  • Ordinary noise too large? Then smooth, with the price in mind. Pick N or α from the noise ratio you need (1 / √N, or √(α / (2 − α))) and check the delay it costs: (N − 1) / 2 or (1 − α) / α samples, times v dt metres of lag at the speed you drive.
  • No outliers, little noise? Do not filter. Delay is the price; the controller of Programming for Robotics 6 (planned; not yet written) is the one that will pay it. A quick “obstacle ahead?” decision needs a median of 3 or 5 and a Schmitt trigger, not a 25-point average.
  • Measure again on the real sensor. The simulator’s numbers are assumptions (Programming for Robotics 3 section 13); the noise record of a real sensor, taken in the room it will work in, is the input to this section.

6. Fusing two range sensors

Two sensors that measure the same thing, with different noise, can give an estimate better than either. The best way to combine two independent estimates x1 and x2 with variances v1 = σ12 and v2 = σ22 is the inverse-variance weighted mean. Take the combination x = w1 x1 + w2 x2 with w1 + w2 = 1 (so that a true value of both gives a true value). Its variance is w12 v1 + w22 v2, which is smallest when each weight is proportional to the inverse of its variance:

w_i = (1 / v_i) / sum(1 / v_j) x = sum(x_i / v_i) / sum(1 / v_i) v = 1 / sum(1 / v_i) Worked example: A (ultrasonic) reads 0.9312 m, sigma 5 mm, v = 25 mm^2; B (infrared) reads 0.9291 m, sigma 2 mm, v = 4 mm^2 1 / v: A 0.040 B 0.250 (per mm^2) sum = 0.290 weights: A 0.040 / 0.290 = 0.138 B 0.250 / 0.290 = 0.862 x = 0.138 x 0.9312 + 0.862 x 0.9291 = 0.9294 m (much closer to B, the better sensor) v = 1 / 0.290 = 3.448 mm^2 sigma = 1.857 mm (better than B alone, 2 mm) Equal weights instead: sigma = 0.5 sqrt(25 + 4) = 2.693 mm (WORSE than B alone: a bad sensor given half the vote drags a good one down)

Two things follow from the formula. The fused variance 1 / (1/v1 + 1/v2) is always smaller than the smaller of the two, so adding even a poor sensor never hurts if the weights are right. And the weights need the variances, so the noise record of section 5 is not a luxury: it is where v1 and v2 come from. The function takes any number of (value, variance) pairs and skips a missing value, so it keeps working when one sensor drops out:

def fuse(estimates): """Inverse-variance weighted mean of (value, variance) pairs; None values are skipped. Returns (value, variance), or (None, None) when there is nothing to fuse.""" total_w = 0.0 total = 0.0 for value, var in estimates: if not is_number(value) or not is_number(var) or var <= 0.0: continue w = 1.0 / var total_w += w total += w * value if total_w == 0.0: return None, None return total / total_w, 1.0 / total_w

The block below uses the simulator with two range sensors at the same mount: the front sensor with a standard deviation of 5 mm and a second one (add_range_sensor) with 2 mm. The worked example is repeated first as code, then 4000 readings of both are fused:

x, var = fuse([(0.9312, 0.005 ** 2), (0.9291, 0.002 ** 2)]) print("by hand: weights 1/25 = %.3f and 1/4 = %.3f per mm^2; fused %.4f m, sigma %.3f mm (1 / sqrt(0.04 + 0.25))" % (1 / 25, 1 / 4, x, math.sqrt(var) * 1000)) sim = Simulator(World(3.0, 2.0), seed=5, pose=(2.0, 1.0, 0.0), sigma_m=0.005) # "front": sigma 5 mm sim.add_range_sensor("b", sigma_m=0.002) # a second sensor, same mount, sigma 2 mm truth = sim.range_sensors["front"].true_range(sim.pose()) a_err, b_err, fused_err, mean_err = [], [], [], [] for _ in range(4000): sim.advance(0.02) a, b = sim.range_m(), sim.range_of("b") f, _v = fuse([(a, 0.005 ** 2), (b, 0.002 ** 2)]) a_err.append((a - truth) * 1000) b_err.append((b - truth) * 1000) fused_err.append((f - truth) * 1000) mean_err.append(((a + b) / 2 - truth) * 1000) print("sigma of the error, 4000 readings (mm): A alone %.2f B alone %.2f fused with weights %.2f plain average %.2f" % (std(a_err), std(b_err), std(fused_err), std(mean_err))) print("predicted: A 5.00 B 2.00 fused %.2f plain average %.2f" % (math.sqrt(var) * 1000, 0.5 * math.sqrt(25 + 4)))
by hand: weights 1/25 = 0.040 and 1/4 = 0.250 per mm^2; fused 0.9294 m, sigma 1.857 mm (1 / sqrt(0.04 + 0.25)) sigma of the error, 4000 readings (mm): A alone 5.03 B alone 1.97 fused with weights 1.85 plain average 2.72 predicted: A 5.00 B 2.00 fused 1.86 plain average 2.69

The by-hand example gives 0.9294 m and a standard deviation of 1.857 mm, as in the worked example above. On the 4000 readings the fused error has a standard deviation of 1.85 mm, against 1.86 mm predicted, better than the 1.97 mm of the better sensor alone, while the plain average is worse than it (2.72 against 2.69 predicted). The method has three limits. The errors must be independent: two sensors that share a bias (the speed of sound for two ultrasonic sensors, a common power-supply ripple) do not average it away. A bias in one is carried into the result, scaled by its weight, so calibrate first (section 8). And the variances must be right: a sensor believed to be better than it is will dominate, and the fusion will follow it wherever it goes. The Kalman filter of Programming for Robotics 10 (planned; not yet written) is this same weighting, repeated over time, with a model of how the quantity moves.

7. Line sensor code

A reflectance array gives one number per sensor, and the program wants one number for the robot: where the line is. Robotics Fundamentals 5 section 4 gave the steps; here they are code, tested against its worked example and then against the simulator’s painted line. The simulator’s own line sensor (Programming for Robotics 3 section 6) answers 0 or 1 per sensor, which is the “on or off only” case. To have analogue values, with the awkward properties of real ones (each sensor with its own floor and line level, noise, partial coverage), the test rig AnalogLine samples the paint at seven points across each sensor’s 10 mm light spot and turns the covered fraction into ADC counts between that sensor’s white and black level, plus noise. It is a rig, not part of a robot: on the real robot its raw() is replaced by the ADC reads of Programming for Robotics 2 section 2.

class AnalogLine: """Test rig: turns the simulator's painted lines into raw numbers like an analogue reflectance array (ADC counts). Each sensor has its own white (floor) and black (line) level and its own noise. Its light spot is sampled at several points across the array, so a sensor half over the line reads about half way between its two levels.""" def __init__(self, sim, rng, white=(310, 280, 350, 300, 330), black=(3550, 3300, 3700, 3400, 3250), noise=8.0, spot_m=0.010, points=7): self.sim = sim self.rng = rng self.white = white self.black = black self.noise = noise # counts self.spot_m = spot_m self.points = points def coverage(self, left_m): """The fraction of the light spot (centred left_m to the left of the robot's axis) that is on paint.""" sim = self.sim x, y, theta = sim.pose() forward = sim.line_sensor.forward_m hits = 0 for j in range(self.points): left = left_m + (j / (self.points - 1) - 0.5) * self.spot_m px = x + forward * math.cos(theta) - left * math.sin(theta) py = y + forward * math.sin(theta) + left * math.cos(theta) hits += 1 if sim.world.on_line(px, py) else 0 return hits / self.points def raw(self): out = [] for i, left in enumerate(self.sim.line_sensor.offsets): level = self.white[i] + self.coverage(left) * (self.black[i] - self.white[i]) level += self.rng.gauss(0.0, self.noise) out.append(int(min(4095, max(0, round(level))))) return out

The code of the chapter proper has three parts. LineCalibration is the min/max calibration routine: while the robot sweeps over the floor and the line, keep each sensor’s lowest and highest raw value, and turn any raw value into a weight wi = (rawi − loi) / (hii − loi), limited to 0..1 (Robotics Fundamentals 5 section 4). line_position is the weighted average x = sum(wi xi) / sum(wi), with sensor positions xi = (i − (n − 1) / 2) × pitch, positive to the right, a noise floor under which a weight counts as zero, and the line declared lost when sum(w) is below 0.2. LineTracker adds a plan for the lost line: for a short time hold the last position, then report lost and remember which side the line was last seen on, which is the direction to search:

class LineCalibration: """Per-sensor min (floor) and max (line) from a sweep over both, and the 0..1 weights they give.""" def __init__(self, n): self.lo = [None] * n self.hi = [None] * n def observe(self, raw): for i, v in enumerate(raw): if self.lo[i] is None or v < self.lo[i]: self.lo[i] = v if self.hi[i] is None or v > self.hi[i]: self.hi[i] = v def ok(self, min_span): """Did every sensor see both the floor and the line? (A span that is too small means it never saw one of them.)""" return all(l is not None and h - l >= min_span for l, h in zip(self.lo, self.hi)) def weights(self, raw): """wi = (raw - lo) / (hi - lo), limited to 0..1.""" out = [] for v, l, h in zip(raw, self.lo, self.hi): w = (v - l) / (h - l) if h > l else 0.0 out.append(min(1.0, max(0.0, w))) return out def line_position(weights, pitch_m, floor=0.05, lost=0.2): """The weighted average of Robotics Fundamentals 5 section 4: sum(w x) / sum(w), x positive to the RIGHT of the centre. Weights below `floor` count as zero. Returns metres, or None when sum(w) < `lost` (no sensor sees the line).""" n = len(weights) total = 0.0 moment = 0.0 for i, w in enumerate(weights): if w < floor: w = 0.0 total += w moment += w * (i - (n - 1) / 2) * pitch_m if total < lost: return None return moment / total class LineTracker: """line_position with a plan for losing the line: hold the last position for hold_s, then report it lost and remember which side it was last seen on (that is the way to search).""" def __init__(self, pitch_m, floor=0.05, lost=0.2, hold_s=0.1): self.pitch_m = pitch_m self.floor = floor self.lost = lost self.hold_s = hold_s self.last_m = None # the last position the line was really seen at self.last_t = None self.status = "lost" # "on", "holding" or "lost" def update(self, weights, now_s): pos = line_position(weights, self.pitch_m, self.floor, self.lost) if pos is not None: self.last_m, self.last_t, self.status = pos, now_s, "on" return pos if self.last_t is not None and now_s - self.last_t <= self.hold_s: self.status = "holding" return self.last_m # short gap: carry on with the last position self.status = "lost" return None def side(self): """Which side the line was last seen on: 'left', 'right', or None if it was never seen.""" if self.last_m is None: return None return "right" if self.last_m > 0 else "left"

Calibration first. A 19 mm stripe is painted across the robot’s path, and the robot drives over it at 0.2 m/s (a sweep over both floor and line, as the real routine does by hand or by spinning in place):

world = World(3.0, 2.0) world.add_line([(1.5, 0.2), (1.5, 1.8)], 0.019) # 19 mm black tape across the robot's path (a calibration stripe) sim = Simulator(world, seed=1, pose=(1.2, 1.0, 0.0)) rig = AnalogLine(sim, random.Random(21)) cal = LineCalibration(5) sim.set_duty(0.5, 0.5) # drive over the stripe at 0.2 m/s for _ in range(100): sim.advance(0.02) cal.observe(rig.raw()) print("sensor true floor true line measured lo measured hi") for i in range(5): print(" %d %5d %5d %5d %5d" % (i, rig.white[i], rig.black[i], cal.lo[i], cal.hi[i])) print("calibration ok (every sensor saw a span of at least 1000 counts):", cal.ok(1000)) print("a raw reading of 1000 on sensor 0 becomes weight %.3f; on sensor 2: %.3f" % (cal.weights([1000, 0, 0, 0, 0])[0], cal.weights([0, 0, 1000, 0, 0])[2]))
sensor true floor true line measured lo measured hi 0 310 3550 293 3565 1 280 3300 255 3295 2 350 3700 329 3711 3 300 3400 275 3405 4 330 3250 317 3265 calibration ok (every sensor saw a span of at least 1000 counts): True a raw reading of 1000 on sensor 0 becomes weight 0.216; on sensor 2: 0.198

Each sensor found its own floor and line level, a few counts beyond the true ones: the lowest reading of 100 noisy readings is the true floor level minus a couple of standard deviations of noise (8 counts), and the highest the line level plus a couple. That is the weakness of min/max calibration: a single noise spike sets the extreme, and with a larger noise the weights would be a few percent compressed. The calibration check ok() matters more: a sensor whose span is small never saw both the floor and the line (the sweep missed the line, or the sensor is dead), and the robot should refuse to start and not follow garbage. The two weights at the end show what normalisation does: the same raw reading 1000 is 0.216 of the way to black on sensor 0 and 0.198 on sensor 2, because their levels differ.

Now the formula. The worked example of Robotics Fundamentals 5 section 4 (8 sensors, pitch 8 mm) as code:

pitch = 0.008 w = [0.0, 0.0, 0.0, 0.30, 1.00, 0.45, 0.0, 0.0] # the worked example of Robotics Fundamentals 5 section 4 print("RF5 example, 8 sensors, pitch 8 mm: position %.2f mm (the book: 4.69 mm)" % (line_position(w, pitch) * 1000)) w[7] = 0.05 # a noisy white-floor reading on the far sensor print("with 0.05 on sensor 7 (floor 0.05 keeps it): %.2f mm (the book: 5.33 mm)" % (line_position(w, pitch, floor=0.05) * 1000)) print("with the noise floor at 0.10: %.2f mm (the 0.05 is ignored)" % (line_position(w, pitch, floor=0.10) * 1000)) print("all weights 0: %s; weights 0.1, 0.05 (sum 0.15 < 0.2): %s" % (line_position([0.0] * 5, 0.015), line_position([0.0, 0.1, 0.05, 0.0, 0.0], 0.015, floor=0.0))) print("on/off only (threshold 0.5): %.1f mm, a step of 8 mm at a time" % (line_position([1.0 if v > 0.5 else 0.0 for v in [0, 0, 0, 0.3, 1.0, 0.45, 0, 0]], pitch) * 1000))
RF5 example, 8 sensors, pitch 8 mm: position 4.69 mm (the book: 4.69 mm) with 0.05 on sensor 7 (floor 0.05 keeps it): 5.33 mm (the book: 5.33 mm) with the noise floor at 0.10: 4.69 mm (the 0.05 is ignored) all weights 0: None; weights 0.1, 0.05 (sum 0.15 < 0.2): None on/off only (threshold 0.5): 4.0 mm, a step of 8 mm at a time

The code reproduces both numbers of the book, 4.69 mm and 5.33 mm. The first is the position with the weights 0.30, 1.00 and 0.45 at x = −4, +4 and +12 mm: (0.30 × (−4) + 1.00 × 4 + 0.45 × 12) / 1.75 = 8.20 / 1.75. The second shows what a single noisy white-floor reading of 0.05 on a sensor far from the line (x = +28 mm) does: it moves the answer by 0.64 mm, because it has a long lever arm. A noise floor of 0.10 ignores it, and the answer is 4.69 mm again. Reading on or off only gives 4.0 mm, a step of 8 mm at a time.

A second small worked example, for the 5-sensor array of the simulator (pitch 15 mm) and the 19 mm line, with ideal coverage: a line centred 7.5 mm to the right covers 70 % of the light spot of sensor 2 (the centre) and 70 % of sensor 3 (at +15 mm), and nothing else. The weighted average is (0.7 × 0 + 0.7 × 15) / 1.4 = 7.5 mm, exactly. Finally the whole thing against the simulator, where the true position is known. The robot starts 12 mm to the left of a line (so the line is 12 mm to its right), heading very slightly towards it, and drives along the line for 9 s; the line ends at x = 2.0 m:

world = World(3.0, 2.0) world.add_line([(0.3, 1.0), (2.0, 1.0)], 0.019) # a line along y = 1.0 that ENDS at x = 2.0 sim = Simulator(world, seed=1, pose=(0.4, 1.012, -0.012)) # 12 mm to the left of the line, heading very slightly towards it rig = AnalogLine(sim, random.Random(22)) # the calibration `cal` of the sweep above is reused tracker = LineTracker(0.015, hold_s=0.1) sim.set_duty(0.5, 0.5) # straight ahead at 0.2 m/s errors, errors_binary, trace = [], [], [] lost_at = None for step in range(1, 451): sim.advance(0.02) wts = cal.weights(rig.raw()) pos = tracker.update(wts, sim.now_s()) x, y, theta = sim.pose() sy = y + 0.07 * math.sin(theta) truth = (sy - 1.0) / math.cos(theta) # where the line really is across the array (right = +) if tracker.status == "on": errors.append((pos - truth) * 1000) on_off = line_position([1.0 if v > 0.5 else 0.0 for v in wts], 0.015) if on_off is not None: errors_binary.append((on_off - truth) * 1000) if tracker.status == "lost" and lost_at is None: lost_at = (sim.now_s(), x + 0.07) if step % 50 == 0: trace.append((sim.now_s(), truth * 1000, None if pos is None else pos * 1000, tracker.status)) print(" t (s) true line position (mm) measured (mm) status") for t, tr, ps, st in trace: print(" %5.2f %12.1f %8s %s" % (t, tr, "None" if ps is None else "%.1f" % ps, st)) print("while on the line (%d passes): weighted average error: mean %+.2f mm, rms %.2f mm, largest %.2f mm" % (len(errors), mean(errors), math.sqrt(mean([e * e for e in errors])), max(abs(e) for e in errors))) print("same passes with on/off decisions only: rms %.2f mm, largest %.2f mm" % (math.sqrt(mean([e * e for e in errors_binary])), max(abs(e) for e in errors_binary))) print("declared lost at t = %.2f s with the sensors at x = %.3f m (the line ends at x = 2.000 m); last seen on the %s" % (lost_at[0], lost_at[1], tracker.side()))
t (s) true line position (mm) measured (mm) status 1.00 9.0 9.0 on 2.00 6.6 6.7 on 3.00 4.2 4.5 on 4.00 1.8 1.9 on 5.00 -0.6 -2.0 on 6.00 -3.0 -3.4 on 7.00 -5.4 -5.0 on 8.00 -7.8 None lost 9.00 -10.2 None lost while on the line (388 passes): weighted average error: mean -0.03 mm, rms 0.50 mm, largest 1.45 mm same passes with on/off decisions only: rms 2.95 mm, largest 7.27 mm declared lost at t = 7.86 s with the sensors at x = 2.026 m (the line ends at x = 2.000 m); last seen on the left

The measured position follows the true one to within 0.4 mm in six of the seven rows shown (9.0 and 9.0, then 6.7 against 6.6, and so on); at t = 5.0 s the error is 1.4 mm, close to the largest of the whole run. Over the 388 passes on the line the error has a root-mean-square of 0.50 mm and a largest value of 1.45 mm. Using only the on/off decisions at the same threshold, the root-mean-square error is 2.95 mm and the largest 7.27 mm: the weighted average is six times as precise with the same hardware, as Robotics Fundamentals 5 promised. When the line ends the position is held for 0.1 s and then declared lost, at t = 7.86 s, when the sensors are 26 mm beyond the end of the line (the paint has a rounded end, 9.5 mm beyond its last point, plus the 0.1 s hold, 20 mm at this speed, less up to one pass of 4 mm). side() says “left” because the line was last at −7 mm: for a robot that has drifted off the line, that is the direction to turn to find it again. Here the line really ended, which is why a line follower also needs a rule for that (a stop, or a search): behaviours are the subject of Programming for Robotics 7 (planned; not yet written).

8. Calibration code

A calibration removes the systematic part of a sensor’s error by comparing it with known true values. Robotics Fundamentals 5 section 10 gave the formulas; the class below implements both. The model is a straight line, reading = gain × true + offset, and the inverse true = (reading − offset) / gain is what the program applies. Two points make the line pass through both exactly: gain = (r2 − r1) / (t2 − t1), offset = r1 − gain t1. Least squares chooses the line through all the points that minimises the sum of squared errors: with Sxx = sum (t − mean t)2 and Sxy = sum (t − mean t)(r − mean r), gain = Sxy / Sxx and offset = mean r − gain × mean t. The true values are on the x axis because they are the ones you know; the noise is in the readings. residuals gives, for each point, the calibrated value minus the true value: the evidence of whether the straight line was a good model.

class LinearCalibration: """reading = gain x true + offset, and the inverse true = (reading - offset) / gain.""" def __init__(self, gain, offset): self.gain = gain self.offset = offset def apply(self, reading): return (reading - self.offset) / self.gain @classmethod def two_point(cls, t1, r1, t2, r2): """Exact at both points: the readings r1 and r2 at the true values t1 and t2.""" gain = (r2 - r1) / (t2 - t1) return cls(gain, r1 - gain * t1) @classmethod def fit(cls, true_values, readings): """Least squares: the line through all the points that makes the sum of squared errors smallest. The true values are the known ones (the x axis); the readings carry the noise.""" mt, mr = mean(true_values), mean(readings) sxx = sum((t - mt) ** 2 for t in true_values) sxy = sum((t - mt) * (r - mr) for t, r in zip(true_values, readings)) gain = sxy / sxx return cls(gain, mr - gain * mt) def residuals(self, true_values, readings): """Calibrated value minus true value, for each point.""" return [self.apply(r) - t for t, r in zip(true_values, readings)]

The numbers of Robotics Fundamentals 5 section 10 (a distance sensor against true distances of 10 to 100 cm), with the worked least-squares arithmetic: the means are 53 and 55.46, Sxx = 432 + 232 + 32 + 222 + 472 = 5080, Sxy = 5107.6, so gain = 5107.6 / 5080 = 1.00543 and offset = 55.46 − 1.00543 × 53 = 2.172:

t = [10.0, 30.0, 50.0, 75.0, 100.0] # Robotics Fundamentals 5 section 10 r = [12.4, 32.1, 52.6, 77.3, 102.9] two = LinearCalibration.two_point(t[0], r[0], t[-1], r[-1]) print("two-point: gain %.5f, offset %.3f; true = %.5f r %+.3f; reading 77.3 -> %.2f cm" % (two.gain, two.offset, 1 / two.gain, -two.offset / two.gain, two.apply(77.3))) print(" errors at the check points 30, 50, 75:", ["%+.2f" % e for e in two.residuals(t[1:4], r[1:4])]) ls = LinearCalibration.fit(t, r) print("least squares: reading = %.5f true + %.3f" % (ls.gain, ls.offset)) print(" errors at the five points:", ["%+.2f" % e for e in ls.residuals(t, r)], " largest %.2f cm" % max(abs(e) for e in ls.residuals(t, r)))
two-point: gain 1.00556, offset 2.344; true = 0.99448 r -2.331; reading 77.3 -> 74.54 cm errors at the check points 30, 50, 75: ['-0.41', '-0.02', '-0.46'] least squares: reading = 1.00543 true + 2.172 errors at the five points: ['+0.17', '-0.23', '+0.16', '-0.28', '+0.18'] largest 0.28 cm

The class reproduces the book’s numbers: a two-point calibration with true = 0.99448 r − 2.331 (gain 1.00556 in the other direction), a reading of 77.3 cm is really 74.54 cm, its errors at the three check points are −0.41, −0.02 and −0.46 cm; the least-squares line reading = 1.00543 true + 2.172 has errors +0.17, −0.23, +0.16, −0.28 and +0.18 cm, the largest of them 0.28 cm, smaller than the 0.46 cm of the two-point line, as the book said. Two practical points from the book matter in the code: average many readings at each calibration point (a two-point calibration puts the noise of two readings straight into the gain and the offset), and keep a check point that is not used in the fit.

A nonlinear sensor: calibrate in a linear variable

A straight line is the wrong model for an infrared triangulation sensor, whose output is strongly nonlinear in distance (Robotics Fundamentals 5 section 3: V is about k / d, so the spot moves as 1 / d). The sensor of the rig below, IRSensor, behaves like that: V = k / (d + d0) with k = 0.27 V m and d0 = 0.02 m, a noise of 10 mV, read by a 12-bit ADC with a 3.3 V reference (an assumed sensor, of the form of the common ones; I have not measured one). The remedy is to change variable: 1 / V = (d + d0) / k is a straight line in d, with gain 1/k and offset d0/k. So the same LinearCalibration fits it, with the readings replaced by 1 / counts, and the inverse, d = (1/counts − offset) / gain, turns a reading into a distance: the “inverse-distance linearisation fitted in code”.

class IRSensor: """Test rig: an infrared triangulation sensor. ADC counts of V = k / (d + d0) plus noise: strongly nonlinear in distance. It reads the TRUE range from the simulator and turns it into counts.""" def __init__(self, sim, rng, k_vm=0.27, d0_m=0.02, noise_v=0.010, vref_v=3.3, bits=12): self.sim = sim self.rng = rng self.k_vm = k_vm self.d0_m = d0_m self.noise_v = noise_v self.vref_v = vref_v self.full = (1 << bits) - 1 def counts(self): d = self.sim.range_sensors["front"].true_range(self.sim.pose()) volts = 0.0 if d is None else self.k_vm / (d + self.d0_m) volts += self.rng.gauss(0.0, self.noise_v) return int(min(self.full, max(0, round(volts / self.vref_v * self.full))))

The block places the robot at eight known distances from the wall, takes 50 readings at each, fits a straight line of counts against distance (a) and a straight line of 1 / counts against distance (b), and then checks both at five distances that were not used in the fit:

sim = Simulator(World(3.0, 2.0), seed=1, pose=(1.0, 1.0, 0.0)) ir = IRSensor(sim, random.Random(31)) def mean_counts_at(d_m, n=50): place(sim, 3.0 - 0.07 - d_m) # the sensor is 0.07 m ahead of the centre return mean([ir.counts() for _ in range(n)]) train_d = [0.10, 0.15, 0.20, 0.30, 0.40, 0.50, 0.60, 0.80] train_c = [mean_counts_at(d) for d in train_d] print("distance (m):", " ".join("%6.2f" % d for d in train_d)) print("mean counts :", " ".join("%6.0f" % c for c in train_c), " (50 readings each; 4095 counts = 3.3 V)") cal_c = LinearCalibration.fit(train_d, train_c) # (a) a straight line: counts against distance cal_u = LinearCalibration.fit(train_d, [1.0 / c for c in train_c]) # (b) a straight line: 1/counts against distance print("(b) 1/counts = %.6f d %+.6f so d = %.1f / counts - %.4f m (the sensor itself: k / (d + d0) with k = 0.27 V m = %.1f counts m, d0 = 0.0200 m)" % (cal_u.gain, cal_u.offset, 1 / cal_u.gain, cal_u.offset / cal_u.gain, 0.27 / 3.3 * 4095)) check_d = [0.12, 0.25, 0.35, 0.45, 0.70] print("check distance (m) " + "".join("%8.2f" % d for d in check_d)) rows = {"(a) line through counts ": [], "(b) line through 1/counts": []} for d in check_d: c = mean_counts_at(d) rows["(a) line through counts "].append((cal_c.apply(c) - d) * 1000) rows["(b) line through 1/counts"].append((cal_u.apply(1.0 / c) - d) * 1000) for name, errs in rows.items(): print("error of %s (mm) " % name + "".join("%+8.1f" % e for e in errs)) sd = {} for d in (0.15, 0.70): place(sim, 3.0 - 0.07 - d) sd[d] = std([cal_u.apply(1.0 / ir.counts()) * 1000 for _ in range(400)]) print("noise of ONE calibrated reading: sigma %.1f mm at 0.15 m, %.1f mm at 0.70 m (the 1/d curve is flat far away)" % (sd[0.15], sd[0.70]))
distance (m): 0.10 0.15 0.20 0.30 0.40 0.50 0.60 0.80 mean counts : 2793 1971 1524 1046 797 645 537 411 (50 readings each; 4095 counts = 3.3 V) (b) 1/counts = 0.002977 d +0.000062 so d = 335.9 / counts - 0.0210 m (the sensor itself: k / (d + d0) with k = 0.27 V m = 335.0 counts m, d0 = 0.0200 m) check distance (m) 0.12 0.25 0.35 0.45 0.70 error of (a) line through counts (mm) -133.0 +123.0 +135.6 +100.1 -66.7 error of (b) line through 1/counts (mm) -0.5 -0.1 +0.5 +0.7 +3.1 noise of ONE calibrated reading: sigma 1.2 mm at 0.15 m, 18.6 mm at 0.70 m (the 1/d curve is flat far away)

The counts fall from 2793 at 0.10 m to 411 at 0.80 m, a factor of seven, and not linearly. The fit in the inverse variable recovers the sensor’s own constants: d = 335.9 / counts − 0.0210 m, against the true k = 0.27 V m = 335.0 counts m and d0 = 0.0200 m, which is 0.3 % and 1 mm off. Its errors at the five check distances are −0.5, −0.1, +0.5, +0.7 and +3.1 mm. The straight line through the counts, fitted by least squares through the same points, is wrong by 67 to 136 mm, which is as bad as having no calibration: the model was wrong, and no amount of data would have helped (the residuals of a good fit have no pattern; here they would have been large and systematic, which is what residuals is for). The last line shows the price: the noise of one calibrated reading is 1.2 mm at 0.15 m and 18.6 mm at 0.70 m, since 1/d is steep near the sensor and flat far away, so a fixed ADC noise means a large distance noise far out. That is the reason to filter an infrared sensor harder at long range, and to trust it less. Two caveats from memory of the sensor type rather than the simulation: many real infrared sensors give a lower voltage at very short range (below a few centimetres), which makes the reading ambiguous, and their output depends on the colour and the surface of the target.

9. Encoders: ticks, speed and wrap-around

Programming for Robotics 2 section 9 decoded the quadrature signals into one signed tick count per wheel, and Programming for Robotics 1 section 6 turned ticks into metres (0.340 mm per tick for the build robot’s 600 ticks per revolution and 65 mm wheels). This section does what is left: a speed that is steadier than one tick per pass, and a count that survives wrapping around.

Counters wrap around

A hardware counter register has a fixed number of bits and wraps from its maximum back to zero (or the other way). An 8-bit counter wraps every 256 ticks, 8.7 cm of travel for this robot; a 16-bit counter every 65536 ticks, 22.3 m. A program that subtracts the previous count from the new one gets a huge jump at each wrap. The fix is a modular difference: the change of an N-bit counter is the difference reduced modulo 2N into the range −2N−1 .. +2N−1 − 1. In code, ((new − old + 2N−1) mod 2N) − 2N−1. Worked example, 8 bits: the counter goes from 250 to 5; the plain difference is −245; the modular one is ((5 − 250 + 128) mod 256) − 128 = (−117 mod 256) − 128 = 139 − 128 = +11: it went forward 6 ticks to 255, wrapped, and another 5. The method assumes the true change between two readings is less than 2N−1: it cannot tell a change of +200 from one of −56. That is a limit on the loop period: the counter must be read at least every 2N−1 × (metres per tick) / vmax seconds. For 8 bits at 0.4 m/s that is 128 × 0.340 mm / 0.4 = 0.11 s.

SpeedEstimator unwraps the counts into a growing total, and then does two things to make the speed steadier. It divides the change over the last window passes by the real time those passes took (the timestamps, not the nominal dt, so jitter in the loop does not become noise in the speed), and it can follow that with an exponential filter. WrappingTicks is a rig: it wraps the simulator’s tick count at a chosen number of bits, as a real register would.

def delta_ticks(new, old, bits): """The signed change of a counter that wraps around at 2**bits. Right while the true change is under 2**(bits-1).""" full = 1 << bits return ((new - old + full // 2) % full) - full // 2 class WrappingTicks: """Wraps a hardware object so that ticks() wraps around at 2**bits, like a real counter register.""" def __init__(self, hw, bits): self.hw = hw self.mask = (1 << bits) - 1 def ticks(self): left, right = self.hw.ticks() return (left & self.mask, right & self.mask) def __getattr__(self, name): return getattr(self.hw, name) class SpeedEstimator: """Speed from encoder counts: the change over the last `window` passes divided by the REAL time they took, then an exponential filter (alpha = 1: none). Counts that wrap around (bits) are unwrapped first.""" def __init__(self, metres_per_tick, window=1, alpha=1.0, bits=None): self.mpt = metres_per_tick self.window = window self.alpha = alpha self.bits = bits self.last = None # the previous raw count self.total = 0 # unwrapped total of ticks self.hist = deque(maxlen=window + 1) # (total, time) of the last window + 1 passes self.speed_mps = 0.0 def update(self, ticks, now_s): if self.last is None: self.last = ticks step = delta_ticks(ticks, self.last, self.bits) if self.bits else ticks - self.last self.last = ticks self.total += step self.hist.append((self.total, now_s)) old_total, old_t = self.hist[0] if now_s > old_t: raw = (self.total - old_total) * self.mpt / (now_s - old_t) self.speed_mps += self.alpha * (raw - self.speed_mps) return self.speed_mps @property def distance_m(self): return self.total * self.mpt
mpt = Config.METRES_PER_TICK # 0.340 mm per tick (Programming for Robotics 1 section 6) sim = Simulator(World(3.0, 2.0), seed=1, pose=(0.3, 1.0, 0.0)) hw = WrappingTicks(sim, 8) # an 8-bit counter wraps every 256 ticks = 8.7 cm sim.set_duty(0.5, 0.5) naive_total, last = 0, hw.ticks()[0] est = SpeedEstimator(mpt, bits=8) est.update(hw.ticks()[0], sim.now_s()) # start counting from the first reading for _ in range(100): # 2 s at 0.2 m/s = 0.4 m sim.advance(0.02) now = hw.ticks()[0] naive_total += now - last # the wrong way: plain subtraction last = now est.update(now, sim.now_s()) print("true distance %.4f m (from the position); true ticks %d, counter now reads %d" % (sim.pose()[0] - 0.3, sim.ticks()[0], hw.ticks()[0])) print("plain subtraction: %d ticks = %.3f m; modular difference: %d ticks = %.4f m" % (naive_total, naive_total * mpt, est.total, est.distance_m)) print("delta_ticks(5, 250, 8) = %d, delta_ticks(250, 5, 8) = %d, delta_ticks(100, 90, 8) = %d" % (delta_ticks(5, 250, 8), delta_ticks(250, 5, 8), delta_ticks(100, 90, 8))) print("limit: at 0.4 m/s one 8-bit counter can only be trusted if it is read every %.2f s (half the count range 128 ticks = %.1f cm)" % (128 * mpt / 0.4, 128 * mpt * 100))
true distance 0.3840 m (from the position); true ticks 1128, counter now reads 104 plain subtraction: 104 ticks = 0.035 m; modular difference: 1128 ticks = 0.3839 m delta_ticks(5, 250, 8) = 11, delta_ticks(250, 5, 8) = -11, delta_ticks(100, 90, 8) = 10 limit: at 0.4 m/s one 8-bit counter can only be trusted if it is read every 0.11 s (half the count range 128 ticks = 4.4 cm)

The robot drove 0.384 m (1128 ticks). The 8-bit counter now reads 104. Plain subtraction accumulates 104 ticks, 0.035 m: it only knows that the counter went from 0 to 104, and has lost the four wraps. (The sum of plain differences is the last count minus the first, whatever happened in between.) The modular difference counts 1128 ticks, 0.3839 m, which is the truth to within the 0.00034 m of a single tick (the position is the exact one, the count is rounded down to whole ticks). The three delta_ticks examples are the worked example in the text. And the last line is the loop-period limit: at 0.4 m/s an 8-bit counter has to be read every 0.11 s. On a real robot the wrap is rarely an 8-bit counter, but a 16-bit timer counter in encoder mode at a loop that stalls for a second is exactly this problem. The same arithmetic in C++ happens for free with integer types of the right width, which is worth knowing because it is the usual way the problem is solved on an Arduino (from memory; not compiled here):

// Arduino C++: from memory, NOT compiled, NOT run here uint16_t last_count = 0; // a 16-bit hardware counter register, read each pass int32_t total_ticks = 0; // the unwrapped total: a 32-bit signed integer void update(uint16_t count) { int16_t delta = (int16_t)(count - last_count); // the same modular difference as delta_ticks(count, last_count, 16) last_count = count; total_ticks += delta; }

Speed from ticks: quantisation and smoothing

The speed from the change in ticks over an interval dt is v = Δticks × (metres per tick) / dt (Robotics Fundamentals 6 section 2). The ticks are whole numbers, so the speed moves in steps of one tick: the resolution is metres per tick / dt, which is 0.00034034 / 0.02 = 0.0170 m/s here, independent of how fast the robot goes. In revolutions per second it is 1 / (CPR × dt) = 1 / (600 × 0.02) = 0.0833 rev/s. At 0.3 m/s the robot moves 17.63 ticks per pass, so the count is 17 or 18, and the speed is 0.289 or 0.306 m/s, as Programming for Robotics 3 section 8 showed. The spread of that flicker can be predicted. If f is the fractional part of the ticks per pass (here 0.63), the count is the larger value a fraction f of the time, so the standard deviation of the speed is the resolution × √(f (1 − f)), at most half a resolution step (f = 1/2). Over a window of N passes the resolution improves to (1 / N) of that, with the f of N passes of travel:

dt = 0.02 res = mpt / dt print("one tick per pass = %.4f m/s (= 1 / (CPR x dt) = 1 / (600 x 0.02) = %.4f rev/s); at 0.3 m/s the ticks per pass are %.2f" % (res, 1 / (600 * dt), 0.3 * dt / mpt)) ticks_per_pass = 0.3 * dt / mpt f = ticks_per_pass - int(ticks_per_pass) print("predicted spread of the raw speed: resolution x sqrt(f (1 - f)) = %.4f x %.3f = %.4f m/s (largest possible, f = 1/2: %.4f)" % (res, math.sqrt(f * (1 - f)), res * math.sqrt(f * (1 - f)), res / 2)) f5 = 5 * ticks_per_pass - int(5 * ticks_per_pass) print("over a window of 5 passes: (resolution / 5) x sqrt(f5 (1 - f5)) = %.5f x %.3f = %.4f m/s (f5 = %.2f ticks left over)" % (res / 5, math.sqrt(f5 * (1 - f5)), res / 5 * math.sqrt(f5 * (1 - f5)), f5)) def speed_run(window, alpha, seconds=4.0): sim = Simulator(World(3.0, 2.0), seed=1, pose=(0.3, 1.0, 0.0)) est = SpeedEstimator(mpt, window, alpha) sim.set_duty(0.75, 0.75) t_true = [] speeds = [] for k in range(int(seconds / dt)): sim.advance(dt) speeds.append(est.update(sim.ticks()[0], sim.now_s())) t_true.append(sim.model.true_speeds_mps()[0]) rise = lambda seq: next(i for i, s in enumerate(seq) if s >= 0.27) * dt # first time at 90 % of 0.3 m/s steady = speeds[100:] return mean(steady), std(steady), rise(speeds), rise(t_true) print("estimator steady mean (m/s) sigma (m/s) time to 90 %% (s) (true speed: 90 %% at %.2f s)" % speed_run(1, 1.0)[3]) for name, window, alpha in [("raw: 1 pass", 1, 1.0), ("window of 5 passes", 5, 1.0), ("raw + EMA alpha 0.2", 1, 0.2), ("window 5 + EMA alpha 0.5", 5, 0.5)]: m, s, rise, _ = speed_run(window, alpha) print("%-30s %12.4f %9.4f %9.2f" % (name, m, s, rise))
one tick per pass = 0.0170 m/s (= 1 / (CPR x dt) = 1 / (600 x 0.02) = 0.0833 rev/s); at 0.3 m/s the ticks per pass are 17.63 predicted spread of the raw speed: resolution x sqrt(f (1 - f)) = 0.0170 x 0.483 = 0.0082 m/s (largest possible, f = 1/2: 0.0085) over a window of 5 passes: (resolution / 5) x sqrt(f5 (1 - f5)) = 0.00340 x 0.354 = 0.0012 m/s (f5 = 0.15 ticks left over) estimator steady mean (m/s) sigma (m/s) time to 90 % (s) (true speed: 90 % at 0.18 s) raw: 1 pass 0.3000 0.0083 0.18 window of 5 passes 0.3000 0.0012 0.24 raw + EMA alpha 0.2 0.3000 0.0011 0.32 window 5 + EMA alpha 0.5 0.3000 0.0006 0.26

The prediction for one pass, 0.0082 m/s, matches the measured spread of the raw speed, 0.0083 m/s, and the 5-pass window predicts 0.0012 m/s and measures 0.0012. The mean is exactly right in all four rows (0.3000 m/s): quantisation does not bias a speed over time, it only flickers. So the choice is the usual trade of noise for delay: the 5-pass window cuts the spread by 7 times, at the cost of a slower response, 0.24 s to reach 90 % of the speed, against 0.18 s for the true speed (the motor’s own lag, the 0.08 s time constant of Programming for Robotics 3 section 5: ln 10 × 0.08 = 0.184 s) and 0.18 s for the raw estimate. The exponential filter with α = 0.2 is almost as smooth (0.0011) and slower (0.32 s), and the combination of a 5-pass window and α = 0.5 has the smallest spread, 0.0006 m/s, at 0.26 s. Which is right depends on the loop that uses the speed: a speed controller (Programming for Robotics 6, planned; not yet written) is the customer that pays for delay, so it takes the shortest window that its resolution allows. For very slow speeds, timing the gap between ticks is better than counting them (Robotics Fundamentals 6 section 2), but it needs a timestamp for each tick from the interrupt handler of Programming for Robotics 2 section 9, which a polling loop does not have.

10. Sensor health: stuck, bad and stale, and the safe default

The Sensor class gives a value that is valid or not. It does not notice a sensor that is valid and wrong: one that has frozen on a reading that is perfectly in range. A frozen range is the most dangerous failure of this chapter, because everything downstream looks healthy: the filter is calm, the controller is happy, and the robot drives at the wall at full speed while its sensor insists the wall is 1.7 m away. The health monitor looks for the three signs, each with a number you choose:

  • Stuck. The same value, exactly, stuck_n valid readings in a row. A real sensor with noise almost never repeats a value: if the chance of a repeat is p per reading, a run of n has chance pn, which for even p = 0.5 and n = 20 is one in a million. A sensor with a coarse ADC and a very quiet signal repeats often, so stuck_n must be large for those; and clamped readings (the range limit) are excluded because they repeat legitimately. The cost of the check is detection time: stuck_n passes (0.4 s at 50 Hz and 20), during which the robot moves 12 cm at 0.3 m/s.
  • Bad. bad_max invalid readings in a row (None, NaN, out of range). One is a dropped echo; ten in a row is a sensor that has stopped working.
  • Stale. No valid reading for the sensor’s timeout (the 0.11 s of Programming for Robotics 1 section 6).

Health watches a Sensor and names the fault. FilteredHardware then puts the whole pipeline of this chapter behind the same three methods as the hardware: range_m() returns the validated, filtered value, and returns None when the health check has a fault. The Robot of Programming for Robotics 1 needs no change at all: its own timeout logic (Programming for Robotics 1 section 6 and 11) counts the age of a missing range and stops with the fault “range stale”, which is exactly the safe default of that chapter, fed by a better detector. A stuck sensor’s readings are kept out of the filter (so the filter does not keep averaging a frozen number) and a good sensor’s readings resume the output when the fault clears, since a new valid value resets the run of bad ones.

class Health: """Watches one Sensor and names what is wrong with it: 'stuck', 'bad' or 'stale' (None when it looks fine). stuck: the same value, exactly, stuck_n valid readings in a row. A real sensor with noise never does that. bad: bad_max invalid readings in a row (None, NaN, out of range). stale: no valid reading for the sensor's timeout.""" def __init__(self, sensor, stuck_n=20, bad_max=10): self.sensor = sensor self.stuck_n = stuck_n self.bad_max = bad_max self.last = None self.same_run = 0 self.fault = None def update(self): """Call once after every sensor.read().""" s = self.sensor if s.valid and not s.clamped: self.same_run = self.same_run + 1 if s.value == self.last else 0 self.last = s.value if self.same_run >= self.stuck_n: self.fault = "stuck" elif s.bad_run >= self.bad_max: self.fault = "bad" elif s.stale(): self.fault = "stale" else: self.fault = None return self.fault class FilteredHardware: """Wraps a hardware object. range_m() returns the cleaned range: validated, filtered and health-checked. It returns None when the range cannot be trusted, so the robot's own timeout logic takes over. The rest passes through.""" def __init__(self, hw, pipeline, low=0.02, high=4.0, timeout_s=0.11, stuck_n=20): self.hw = hw self.pipeline = pipeline self.sensor = Sensor(hw.range_m, hw.now_s, low, high, timeout_s, name="range") self.health = Health(self.sensor, stuck_n) def range_m(self): raw = self.sensor.read() fault = self.health.update() if raw is not None and fault != "stuck": self.pipeline.update(raw) # only good, moving readings go in return None if fault else self.pipeline.value def __getattr__(self, name): return getattr(self.hw, name)

Two experiments, in both the Robot of Programming for Robotics 1 is used unchanged, with the simulator for a world and FaultyHardware for the fault. In the first the sensor has 10 % of wrong echoes, and the robot drives at the wall to the stop line of 0.25 m. Because FilteredHardware already filters, the robot’s own exponential filter is switched off for the third case with a configuration that has RANGE_FILTER_ALPHA = 1.0. The score is the root-mean-square difference between the speed the robot commands and the speed it should command for the true range:

class NoFilterConfig(Config): RANGE_FILTER_ALPHA = 1.0 # the Robot of Programming for Robotics 1 section 6 does no filtering of its own def approach_run(make_hw, cfg, seconds=14.0, seed=1): """The Robot of Programming for Robotics 1 drives at the wall; the error of its command against the ideal command.""" sim = Simulator(World(3.0, 2.0), seed=seed, pose=(0.3, 1.0, 0.0)) robot = Robot(make_hw(sim), cfg) errs = [] for _ in range(int(seconds / 0.02)): sim.advance(0.02) robot.update(0.02) true_m = sim.range_sensors["front"].true_range(sim.pose()) ideal = cfg.CRUISE_SPEED_MPS * clamp((true_m - cfg.STOP_RANGE_M) / (cfg.SLOW_RANGE_M - cfg.STOP_RANGE_M), 0.0, 1.0) errs.append(robot.state.cmd_left_mps - ideal) return sim, robot, errs cases = [("clean sensor, robot's own EMA 0.5", lambda sim: sim, Config), ("10 % wrong echoes, robot's own EMA 0.5", lambda sim: FaultyHardware(sim, random.Random(11), spike_rate=0.10), Config), ("10 % wrong echoes, median 5 + EMA 0.5", lambda sim: FilteredHardware(FaultyHardware(sim, random.Random(11), spike_rate=0.10), Pipeline(MedianFilter(5), ExpFilter(0.5))), NoFilterConfig)] print("case rms cmd error (m/s) largest (m/s) final true range (m) bumps fault") for name, make_hw, cfg in cases: sim, robot, errs = approach_run(make_hw, cfg) rng_final = sim.range_sensors["front"].true_range(sim.pose()) print("%-40s %10.4f %6.3f %6.3f %d %s" % (name, math.sqrt(mean([e * e for e in errs])), max(abs(e) for e in errs), rng_final, sim.bumps, robot.state.fault))
case rms cmd error (m/s) largest (m/s) final true range (m) bumps fault clean sensor, robot's own EMA 0.5 0.0021 0.010 0.250 0 None 10 % wrong echoes, robot's own EMA 0.5 0.0794 0.300 0.020 12 None 10 % wrong echoes, median 5 + EMA 0.5 0.0144 0.300 0.236 0 None

With a clean sensor the robot’s command is within 2 mm/s (an rms of 0.0021 m/s) of the ideal, it stops 0.250 m from the wall, and nothing is bumped. With 10 % of wrong echoes and only its own exponential filter, the rms command error is 0.079 m/s, the command is off by the full 0.300 m/s at the worst moment (for example a wild reading of 0.1 m makes the robot brake, and one of 2 m makes it speed up near the wall), and 12 passes end in a collision with the wall and the robot ends 0.020 m from it, touching. With a median of 5 and the exponential filter behind FilteredHardware, the same sensor gives an rms error of 0.014 m/s, no bumps, and a stop 0.236 m from the wall, 14 mm inside the stop line: an occasional wild reading that survives the median makes the robot surge forward (the same pipeline on a clean sensor stops at 0.2497 m). The largest error is still 0.300 m/s: a median of 5 fails when three of five readings are wild, about once in 120 windows at this rate, and then the robot is wrong for a pass or two. A filter reduces the damage of a bad sensor; it does not make the sensor good.

In the second experiment the sensor freezes at t = 3.0 s (stuck_after_s), repeating its last range of about 1.7 m for ever, while the robot keeps driving:

def stuck_run(monitored, seconds=12.0): sim = Simulator(World(3.0, 2.0), seed=1, pose=(0.3, 1.0, 0.0)) hw = FaultyHardware(sim, random.Random(2), stuck_after_s=3.0) # the sensor freezes at t = 3 s cfg = Config if monitored: hw = FilteredHardware(hw, Pipeline(MedianFilter(5), ExpFilter(0.5))) cfg = NoFilterConfig robot = Robot(hw, cfg) first_fault = None for _ in range(int(seconds / 0.02)): sim.advance(0.02) robot.update(0.02) if first_fault is None and robot.state.fault: first_fault = (sim.now_s(), sim.pose()[0], robot.state.fault, hw.health.fault if monitored else None) return sim, robot, first_fault for monitored in (False, True): sim, robot, ff = stuck_run(monitored) print("%s: final x = %.3f m, bumps %d of 600 passes, wheel command %.2f m/s" % ("with health checks " if monitored else "no health checks ", sim.pose()[0], sim.bumps, robot.state.cmd_left_mps)) if ff: print(" first fault at t = %.2f s (x = %.3f m): robot says %r, the health check says %r" % ff) else: print(" the robot never raised a fault")
no health checks : final x = 2.910 m, bumps 160 of 600 passes, wheel command 0.30 m/s the robot never raised a fault with health checks : final x = 1.338 m, bumps 0 of 600 passes, wheel command 0.00 m/s first fault at t = 3.48 s (x = 1.314 m): robot says 'range stale', the health check says 'stuck'

Without health checks the robot sees a calm, believable 1.7 m, never raises a fault, drives at 0.30 m/s into the wall and pushes against it: 160 of the 600 passes end in a collision, and it stops at x = 2.910 m only because the wall stops it. With the checks, the health monitor notices at about 3.4 s that the value has not changed for 20 readings (the stuck run), the range goes missing, the robot’s own timeout turns it into the fault “range stale” at t = 3.48 s (x = 1.314 m, still 1.5 m from the wall), and the wheel command is 0.00 m/s from then on. The robot stopped for the right reason (the health check says “stuck”, the robot says “range stale”) with no bump. Log both: “range stale” alone would hide whether the sensor was silent or frozen, and the two have different causes.

The detector is a design decision, not a default
A stop is the safe default for this small wheeled robot, but not for every robot (Programming for Robotics 1 section 11, and Robotics Fundamentals 10 section 7): an arm holding a load should hold, not stop. The thresholds are also a trade: a short stuck_n and bad_max catch a failure early and stop the robot on a fluke; long ones are patient and let the robot drive blind for longer. Choose them from the distance covered while waiting (speed × passes × dt), the way the lag was chosen in section 4, and test them by fault injection, as above.

11. Testing filters, with synthetic signals and with the simulator

A filter looks right on a plot and fails on the first strange input, so each filter gets tests, and the best tests need no robot: synthetic signals whose correct output you can work out by hand. Four signals cover most of the ground:

  • A constant. The output must equal the input (to rounding), whatever the filter. It catches a wrong normalisation (a moving average divided by the wrong number) and a wrong start-up.
  • A step. The output must follow a response you can write down: a 5-point average goes 0.2, 0.4, 0.6, 0.8, 1.0; a median goes from 0 to 1 at the third new sample; an exponential filter with α = 0.25 goes 0.25, 0.4375, 0.578125.
  • A spike. One wild value: a median must ignore it completely, an average must spread exactly spike / N over exactly N outputs and then forget it.
  • Garbage. None, NaN, infinity: the output must not change, and the next good reading must give the same answer as if the garbage had never arrived. This is the test that finds the NaN that poisons a running sum for ever.

Add one property test with random input (a fixed seed): the output of a median or a moving average always lies between the smallest and the largest of the last N inputs. It catches off-by-one errors in the ring buffer that no hand-made case thought of. The tests below are plain functions that use assert, and a ten-line runner finds every function called test_… in a namespace and runs it (the same job pytest does, with no installation):

def approx(a, b, tol=1e-9): return abs(a - b) <= tol def test_constant_input_passes_through(): for f in (MovingAverage(5), MedianFilter(5), ExpFilter(0.25), Pipeline(MedianFilter(5), ExpFilter(0.5))): for _ in range(30): y = f.update(2.5) assert approx(y, 2.5, 1e-12), type(f).__name__ def test_moving_average_step(): f = MovingAverage(5) for _ in range(10): f.update(0.0) out = [f.update(1.0) for _ in range(6)] assert all(approx(a, b) for a, b in zip(out, [0.2, 0.4, 0.6, 0.8, 1.0, 1.0])), out def test_median_step(): f = MedianFilter(5) for _ in range(10): f.update(0.0) out = [f.update(1.0) for _ in range(6)] assert out == [0.0, 0.0, 1.0, 1.0, 1.0, 1.0], out def test_exponential_step(): f = ExpFilter(0.25) f.update(0.0) out = [f.update(1.0) for _ in range(3)] assert all(approx(a, b) for a, b in zip(out, [0.25, 0.4375, 0.578125])), out def test_spike_median_ignores_it_average_smears_it(): med, ma = MedianFilter(5), MovingAverage(5) for f in (med, ma): for _ in range(10): f.update(1.0) stream = [50.0, 1.0, 1.0, 1.0, 1.0, 1.0] med_out = [med.update(x) for x in stream] ma_out = [ma.update(x) for x in stream] assert all(approx(y, 1.0) for y in med_out), med_out assert all(approx(y, 1.0 + 49.0 / 5) for y in ma_out[:5]) and approx(ma_out[5], 1.0), ma_out def test_bad_readings_are_ignored(): makers = (lambda: MovingAverage(5), lambda: MedianFilter(5), lambda: ExpFilter(0.3), lambda: Pipeline(MedianFilter(3), ExpFilter(0.5))) for make in makers: a, b = make(), make() for x in (1.0, 2.0, 3.0): a.update(x) b.update(x) before = a.value for bad in (None, float("nan"), float("inf"), -float("inf")): assert a.update(bad) == before, (type(a).__name__, bad) assert approx(a.update(4.0), b.update(4.0)), type(a).__name__ def test_first_bad_reading_gives_none(): assert MovingAverage(3).update(float("nan")) is None assert ExpFilter(0.5).update(None) is None def test_exponential_time_constant(): f = ExpFilter.from_tau(0.2, 0.02) assert approx(f.alpha, 0.02 / 0.22) and approx(f.tau_s(0.02), 0.2) def test_slew_limiter(): s = SlewLimiter(1.0) s.update(0.0, 0.1) up = [s.update(10.0, 0.1) for _ in range(3)] assert all(approx(a, b) for a, b in zip(up, [0.1, 0.2, 0.3])), up assert approx(s.update(0.25, 0.1), 0.25) def test_schmitt_does_not_chatter(): s = Schmitt(0.25, 0.30) out = [s.update(x) for x in (0.5, 0.28, 0.26, 0.249, 0.26, 0.29, 0.31, 0.29)] assert out == [False, False, False, True, True, True, False, False], out def test_debouncer_needs_n_in_a_row(): d = Debouncer(3) out = [int(d.update(x)) for x in (1, 1, 0, 1, 1, 1, 0, 0, 1, 0, 0, 0)] assert out == [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 0], out def test_counter_wrap(): assert delta_ticks(5, 250, 8) == 11 and delta_ticks(250, 5, 8) == -11 and delta_ticks(100, 90, 8) == 10 def test_fuse(): x, var = fuse([(1.0, 4.0), (3.0, 4.0)]) assert approx(x, 2.0) and approx(var, 2.0), (x, var) # equal variances: the plain mean, half the variance x, var = fuse([(None, 1.0), (5.0, 9.0)]) assert approx(x, 5.0) and approx(var, 9.0), (x, var) # a missing reading is skipped assert fuse([]) == (None, None) and fuse([(float("nan"), 1.0)]) == (None, None) def test_calibration_hits_its_points(): c = LinearCalibration.two_point(10.0, 12.4, 100.0, 102.9) assert approx(c.apply(12.4), 10.0) and approx(c.apply(102.9), 100.0) exact = LinearCalibration.fit([1.0, 2.0, 3.0], [5.0, 7.0, 9.0]) # reading = 2 true + 3 assert approx(exact.gain, 2.0) and approx(exact.offset, 3.0) assert all(approx(e, 0.0) for e in exact.residuals([1.0, 2.0, 3.0], [5.0, 7.0, 9.0])) def test_line_position_and_lost_line(): assert line_position([0, 0, 0, 0, 0], 0.015) is None assert approx(line_position([0, 0, 1, 0, 0], 0.015), 0.0) assert approx(line_position([0, 0, 0, 1, 0], 0.015), 0.015) assert approx(line_position([0, 1, 1, 0, 0], 0.015), -0.0075) t = LineTracker(0.015, hold_s=0.1) assert t.update([0, 0, 0, 1, 0], 0.0) == 0.015 and t.status == "on" assert t.update([0, 0, 0, 0, 0], 0.05) == 0.015 and t.status == "holding" assert t.update([0, 0, 0, 0, 0], 0.20) is None and t.status == "lost" and t.side() == "right" def test_sensor_rejects_garbage_and_goes_stale(): clock = ManualClock() s = Sensor(lambda: float("nan"), clock.now_s, 0.02, 4.0, 0.05) assert s.read() is None and not s.valid and s.stale() values = iter([1.0, float("nan"), float("nan"), float("nan")]) s = Sensor(lambda: next(values), clock.now_s, 0.02, 4.0, 0.05) s.read() for _ in range(3): clock.t += 0.02 s.read() assert not s.valid and s.value == 1.0 and s.stale() and s.bad_run == 3 def test_health_names_a_stuck_sensor(): clock = ManualClock() s = Sensor(lambda: 1.0, clock.now_s, 0.02, 4.0, 0.05) h = Health(s, stuck_n=5) faults = [] for _ in range(8): s.read() faults.append(h.update()) clock.t += 0.02 assert faults == [None] * 5 + ["stuck"] * 3, faults def test_filters_stay_inside_the_data(): rng = random.Random(7) ma, med = MovingAverage(7), MedianFilter(7) window = deque(maxlen=7) for _ in range(500): x = rng.uniform(-3.0, 9.0) window.append(x) for f in (ma, med): y = f.update(x) assert min(window) - 1e-9 <= y <= max(window) + 1e-9 def run_tests(namespace): """Run every function called test_* in the namespace; print one line each.""" failures = 0 names = sorted(n for n, f in namespace.items() if n.startswith("test_") and callable(f)) for name in names: try: namespace[name]() except AssertionError as err: print("FAIL", name, err) failures += 1 else: print("ok ", name) print("%d tests, %d failed" % (len(names), failures)) return failures

Then the tests that need the simulator, written the same way: the noise of a 9-point average must be a third of the sensor’s, the delays measured on a step must match the formulas, and a run at the wall with a dirty sensor behind FilteredHardware must never bump. They are defined in the block below, which then runs everything in the session:

def test_sim_noise_falls_as_one_over_root_n(): sim = Simulator(World(3.0, 2.0), seed=8, pose=(2.0, 1.0, 0.0)) readings = collect(sim, 3000) f = MovingAverage(9) out = [f.update(r) for r in readings][30:] ratio = std(out) / std(readings) assert abs(ratio - 1 / 3) < 0.05, ratio def test_sim_step_delay_matches_theory(): for make, theory in ((lambda: MovingAverage(5), 2.0), (lambda: ExpFilter(0.25), 3.0)): delay = sum(1.0 - s for s in step_response(make)) assert abs(delay - theory) < 0.01, delay def test_sim_wall_run_with_dirty_sensor_never_bumps(): sim, robot, errs = approach_run(lambda s: FilteredHardware(FaultyHardware(s, random.Random(5), spike_rate=0.10), Pipeline(MedianFilter(5), ExpFilter(0.5))), NoFilterConfig) assert sim.bumps == 0 and robot.state.fault is None run_tests(globals())
ok test_bad_readings_are_ignored ok test_calibration_hits_its_points ok test_constant_input_passes_through ok test_counter_wrap ok test_debouncer_needs_n_in_a_row ok test_exponential_step ok test_exponential_time_constant ok test_filters_stay_inside_the_data ok test_first_bad_reading_gives_none ok test_fuse ok test_health_names_a_stuck_sensor ok test_line_position_and_lost_line ok test_median_step ok test_moving_average_step ok test_schmitt_does_not_chatter ok test_sensor_rejects_garbage_and_goes_stale ok test_sim_noise_falls_as_one_over_root_n ok test_sim_step_delay_matches_theory ok test_sim_wall_run_with_dirty_sensor_never_bumps ok test_slew_limiter ok test_spike_median_ignores_it_average_smears_it ok test_stops_before_the_wall ok test_stops_when_sensor_goes_silent 23 tests, 0 failed

Twenty-three tests pass. Two of them (test_stops_before_the_wall and test_stops_when_sensor_goes_silent) are not from this chapter: they are the tests of Programming for Robotics 1 section 7, which are in the same session, and the runner found them because it looks for the name, not the place. If the module is run as a file, the last two lines (below) run the tests that do not need the simulator. A failing test names itself. Two bugs were put into the module on purpose, to see: dividing by self.n instead of self.count in the moving average is caught by test_filters_stay_inside_the_data and by no other test (the others all start with a full buffer; the start-up averaged in empty slots), and forgetting to subtract the oldest reading fails four tests, the constant, the step, the spike and the property test. A test that nothing else duplicates, like the first, is the one that earns the random-input test its place. Keep these tests: the next change to a filter will be made in a hurry.

if __name__ == "__main__": run_tests(globals())

12. The reference listing: filters.py

The blocks of this chapter that have no output and are not marked as written from memory, in the order of the page, are the whole module. Save it as filters.py. It imports only math, random and collections, it needs neither robot.py nor sim.py (the rigs AnalogLine, IRSensor, WrappingTicks and FaultyHardware only use methods that a simulator or a robot has), and running it as a file runs its own tests. The demonstration blocks stay in the chapter; later chapters import what they need from here: from filters import Sensor, MedianFilter, ExpFilter, Pipeline, FilteredHardware.

# filters.py -- the sensor and filter code of Programming for Robotics 4 (standard library only) # Needs nothing from the robot or the simulator: the rigs below only use methods a simulator or a robot has. import math import random from collections import deque def is_number(x): """True for a real, finite number; False for None, NaN and infinity.""" return x is not None and not (isinstance(x, float) and (math.isnan(x) or math.isinf(x))) class Sensor: """One measured quantity with a validity flag, a timestamp, an age and a stale check. read_fn returns a number or None (or raises OSError); clock_fn returns seconds. A reading outside [low, high] is INVALID, unless clamp=True: then it is limited to the range and flagged.""" def __init__(self, read_fn, clock_fn, low, high, timeout_s, clamp=False, name="sensor"): self.read_fn = read_fn self.clock_fn = clock_fn self.low = low self.high = high self.timeout_s = timeout_s # a value older than this is stale self.clamp = clamp self.name = name self.value = None # the newest VALID value; it stays while readings fail, so check age_s() self.valid = False # was the latest read() valid? self.clamped = False # was the latest value limited to the range? self.t_s = None # when the newest valid value was taken self.reads = 0 # counters for the health checks self.bad = 0 # invalid reads, in total self.bad_run = 0 # invalid reads in a row def read(self): """Take one reading. Returns the value if it is valid, else None. Never raises because of the sensor.""" now = self.clock_fn() self.reads += 1 try: raw = self.read_fn() except OSError: # a bus error or a timeout inside the driver raw = None self.clamped = False if is_number(raw) and self.clamp: limited = min(self.high, max(self.low, raw)) self.clamped = limited != raw raw = limited self.valid = is_number(raw) and self.low <= raw <= self.high if self.valid: self.value, self.t_s, self.bad_run = raw, now, 0 return raw self.bad += 1 self.bad_run += 1 return None def age_s(self): """Seconds since the newest valid value (infinity if there never was one).""" return float("inf") if self.t_s is None else self.clock_fn() - self.t_s def stale(self): return self.age_s() > self.timeout_s class ManualClock: """A clock you move by hand: for tests and demonstrations.""" def __init__(self): self.t = 0.0 def now_s(self): return self.t def place(sim, x_m): """Test tool: teleport the robot along x (a step the real world cannot make), then let one pass of time go by.""" sim.model.x = x_m sim.advance(0.02) def collect(hw, n, dt_s=0.02): """n range readings, one per pass, from any hardware object that also has advance(dt_s).""" out = [] for _ in range(n): hw.advance(dt_s) out.append(hw.range_m()) return out class Filter: """Base class: update(x) filters one reading. A bad reading (None, NaN, infinity) is ignored: the output is held.""" def __init__(self): self.value = None # the latest output; None until the first good reading def update(self, x): if is_number(x): self.value = self._step(x) return self.value def _step(self, x): raise NotImplementedError def reset(self): self.value = None class MovingAverage(Filter): """The mean of the last n readings, in O(1) per update: a ring buffer and a running sum.""" def __init__(self, n): Filter.__init__(self) self.n = n self.reset() def reset(self): self.value = None self.buf = [0.0] * self.n # the ring buffer self.i = 0 # where the next reading goes (the oldest one, once the buffer is full) self.count = 0 # how many slots are in use self.total = 0.0 # the running sum of the slots def _step(self, x): self.total += x - self.buf[self.i] # add the new reading, take the oldest out (0.0 while filling) self.buf[self.i] = x self.i = (self.i + 1) % self.n self.count = min(self.count + 1, self.n) return self.total / self.count def resync(self): """Recompute the sum from the buffer: stops rounding errors from piling up over millions of updates.""" self.total = sum(self.buf) class MedianFilter(Filter): """The middle value of the last n readings (n odd): one wild reading never gets through.""" def __init__(self, n): Filter.__init__(self) self.n = n self.buf = deque(maxlen=n) # drops the oldest reading by itself def reset(self): self.value = None self.buf.clear() def _step(self, x): self.buf.append(x) s = sorted(self.buf) m = len(s) // 2 return s[m] if len(s) % 2 else 0.5 * (s[m - 1] + s[m]) # an even count only happens while filling class ExpFilter(Filter): """y = y + alpha (x - y): one number of memory. alpha = 1 passes the readings unchanged.""" def __init__(self, alpha): Filter.__init__(self) assert 0.0 < alpha <= 1.0, "alpha must be in (0, 1]" self.alpha = alpha @classmethod def from_tau(cls, tau_s, dt_s): """The filter whose time constant is tau_s at a sample interval dt_s: alpha = dt / (tau + dt).""" return cls(dt_s / (tau_s + dt_s)) def tau_s(self, dt_s): return dt_s * (1.0 - self.alpha) / self.alpha def _step(self, x): if self.value is None: return x # start at the first reading, not at zero return self.value + self.alpha * (x - self.value) class SlewLimiter: """The output follows the target but moves at most max_rate per second.""" def __init__(self, max_rate): self.max_rate = max_rate # units per second self.value = None def update(self, target, dt_s): if not is_number(target): return self.value # a bad reading: hold if self.value is None: self.value = target else: step = self.max_rate * dt_s self.value += max(-step, min(step, target - self.value)) return self.value class Schmitt: """A comparator with two thresholds, so a noisy reading near the limit cannot make the decision chatter. on_at < off_at: a 'low' detector (e.g. obstacle near): on when x <= on_at, off again when x >= off_at. on_at > off_at: a 'high' detector: on when x >= on_at, off again when x <= off_at.""" def __init__(self, on_at, off_at, state=False): self.on_at = on_at self.off_at = off_at self.state = state def update(self, x): if is_number(x): if self.on_at <= self.off_at: if x <= self.on_at: self.state = True elif x >= self.off_at: self.state = False else: if x >= self.on_at: self.state = True elif x <= self.off_at: self.state = False return self.state class Debouncer: """A yes/no input that changes only after n readings in a row disagree with the current state.""" def __init__(self, n, state=False): self.n = n self.state = state self.run = 0 # readings in a row that disagree with the state def update(self, raw): if raw is None: return self.state # no reading: no change if bool(raw) == self.state: self.run = 0 else: self.run += 1 if self.run >= self.n: self.state = bool(raw) self.run = 0 return self.state class Pipeline(Filter): """Filters in a row: the output of each is the input of the next (for example a median, then an exponential filter).""" def __init__(self, *stages): Filter.__init__(self) self.stages = stages def reset(self): self.value = None for stage in self.stages: stage.reset() def _step(self, x): for stage in self.stages: x = stage.update(x) return x def ma_delay_samples(n): """The delay of an n-point moving average: (n - 1) / 2 samples.""" return (n - 1) / 2 def ema_delay_samples(alpha): """The delay of an exponential filter: (1 - alpha) / alpha samples = tau / dt.""" return (1.0 - alpha) / alpha def ma_noise_ratio(n): """Output sigma / input sigma of an n-point moving average on independent noise: 1 / sqrt(n).""" return 1.0 / math.sqrt(n) def ema_noise_ratio(alpha): """Output sigma / input sigma of an exponential filter on independent noise: sqrt(alpha / (2 - alpha)).""" return math.sqrt(alpha / (2.0 - alpha)) def median_failure_probability(n, p): """The chance that a window of n readings holds a majority of outliers (each with probability p): the median fails.""" need = (n + 1) // 2 return sum(math.comb(n, k) * p ** k * (1 - p) ** (n - k) for k in range(need, n + 1)) def ascii_plot(series, rows=10, lo=0.0, hi=1.0, cell=3, first=0): """Draw equal-length series on a character grid. series = [(char, values), ...]; later ones draw over earlier ones.""" n = len(series[0][1]) grid = [[" "] * (n * cell) for _ in range(rows + 1)] for ch, values in series: for k, v in enumerate(values): r = int(round((hi - min(hi, max(lo, v))) / (hi - lo) * rows)) grid[r][k * cell] = ch lines = [] for r in range(rows + 1): level = hi - (hi - lo) * r / rows lines.append("%4.1f |%s" % (level, "".join(grid[r]).rstrip())) lines.append(" +" + "-" * (n * cell)) axis = [" "] * (n * cell) for k in range(n): if (first + k) % 5 == 0: for j, c in enumerate(str(first + k)): axis[k * cell + j] = c lines.append(" " + "".join(axis).rstrip() + " (samples after the step)") return "\n".join(lines) def mean(values): return sum(values) / len(values) def variance(values): """The sample variance (divides by n - 1).""" m = mean(values) return sum((v - m) ** 2 for v in values) / (len(values) - 1) def std(values): return math.sqrt(variance(values)) def median(values): s = sorted(values) m = len(s) // 2 return s[m] if len(s) % 2 else 0.5 * (s[m - 1] + s[m]) def robust_sigma(values): """Sigma from the median absolute deviation: outliers cannot move it (1.4826 makes it equal sigma for Gaussian noise).""" m = median(values) return 1.4826 * median([abs(v - m) for v in values]) def outlier_rate(values, k=4.0): """The fraction of readings further than k robust sigmas from the median.""" m, s = median(values), robust_sigma(values) return sum(1 for v in values if abs(v - m) > k * s) / len(values) def ascii_histogram(values, lo, hi, bins, width=40): """One row per bin: the bin centre, a bar, the count. Values outside lo..hi are not drawn.""" counts = [0] * bins step = (hi - lo) / bins for v in values: if lo <= v < hi: counts[int((v - lo) / step)] += 1 peak = max(counts) or 1 return "\n".join("%7.2f |%-*s %d" % (lo + (i + 0.5) * step, width, "#" * int(round(c / peak * width)), c) for i, c in enumerate(counts)) class FaultyHardware: """Wraps a hardware object and breaks its range readings on purpose (fault injection). Everything else passes through. spike_rate: the chance that a reading is replaced by a wrong echo, uniform in spike_lo_m .. spike_hi_m. stuck_after_s: from then on range_m() repeats the last value it gave, forever.""" def __init__(self, hw, rng, spike_rate=0.0, spike_lo_m=0.05, spike_hi_m=3.0, stuck_after_s=None): self.hw = hw self.rng = rng self.spike_rate = spike_rate self.spike_lo_m = spike_lo_m self.spike_hi_m = spike_hi_m self.stuck_after_s = stuck_after_s self._last = None def range_m(self): r = self.hw.range_m() chance = self.rng.random() # always draw both numbers: the random stream does not depend on what happened spike = self.rng.uniform(self.spike_lo_m, self.spike_hi_m) if self.stuck_after_s is not None and self.hw.now_s() >= self.stuck_after_s: return self._last if r is not None and chance < self.spike_rate: r = spike self._last = r return r def __getattr__(self, name): # ticks, set_duty, now_s, advance ... go straight through return getattr(self.hw, name) def fuse(estimates): """Inverse-variance weighted mean of (value, variance) pairs; None values are skipped. Returns (value, variance), or (None, None) when there is nothing to fuse.""" total_w = 0.0 total = 0.0 for value, var in estimates: if not is_number(value) or not is_number(var) or var <= 0.0: continue w = 1.0 / var total_w += w total += w * value if total_w == 0.0: return None, None return total / total_w, 1.0 / total_w class AnalogLine: """Test rig: turns the simulator's painted lines into raw numbers like an analogue reflectance array (ADC counts). Each sensor has its own white (floor) and black (line) level and its own noise. Its light spot is sampled at several points across the array, so a sensor half over the line reads about half way between its two levels.""" def __init__(self, sim, rng, white=(310, 280, 350, 300, 330), black=(3550, 3300, 3700, 3400, 3250), noise=8.0, spot_m=0.010, points=7): self.sim = sim self.rng = rng self.white = white self.black = black self.noise = noise # counts self.spot_m = spot_m self.points = points def coverage(self, left_m): """The fraction of the light spot (centred left_m to the left of the robot's axis) that is on paint.""" sim = self.sim x, y, theta = sim.pose() forward = sim.line_sensor.forward_m hits = 0 for j in range(self.points): left = left_m + (j / (self.points - 1) - 0.5) * self.spot_m px = x + forward * math.cos(theta) - left * math.sin(theta) py = y + forward * math.sin(theta) + left * math.cos(theta) hits += 1 if sim.world.on_line(px, py) else 0 return hits / self.points def raw(self): out = [] for i, left in enumerate(self.sim.line_sensor.offsets): level = self.white[i] + self.coverage(left) * (self.black[i] - self.white[i]) level += self.rng.gauss(0.0, self.noise) out.append(int(min(4095, max(0, round(level))))) return out class LineCalibration: """Per-sensor min (floor) and max (line) from a sweep over both, and the 0..1 weights they give.""" def __init__(self, n): self.lo = [None] * n self.hi = [None] * n def observe(self, raw): for i, v in enumerate(raw): if self.lo[i] is None or v < self.lo[i]: self.lo[i] = v if self.hi[i] is None or v > self.hi[i]: self.hi[i] = v def ok(self, min_span): """Did every sensor see both the floor and the line? (A span that is too small means it never saw one of them.)""" return all(l is not None and h - l >= min_span for l, h in zip(self.lo, self.hi)) def weights(self, raw): """wi = (raw - lo) / (hi - lo), limited to 0..1.""" out = [] for v, l, h in zip(raw, self.lo, self.hi): w = (v - l) / (h - l) if h > l else 0.0 out.append(min(1.0, max(0.0, w))) return out def line_position(weights, pitch_m, floor=0.05, lost=0.2): """The weighted average of Robotics Fundamentals 5 section 4: sum(w x) / sum(w), x positive to the RIGHT of the centre. Weights below `floor` count as zero. Returns metres, or None when sum(w) < `lost` (no sensor sees the line).""" n = len(weights) total = 0.0 moment = 0.0 for i, w in enumerate(weights): if w < floor: w = 0.0 total += w moment += w * (i - (n - 1) / 2) * pitch_m if total < lost: return None return moment / total class LineTracker: """line_position with a plan for losing the line: hold the last position for hold_s, then report it lost and remember which side it was last seen on (that is the way to search).""" def __init__(self, pitch_m, floor=0.05, lost=0.2, hold_s=0.1): self.pitch_m = pitch_m self.floor = floor self.lost = lost self.hold_s = hold_s self.last_m = None # the last position the line was really seen at self.last_t = None self.status = "lost" # "on", "holding" or "lost" def update(self, weights, now_s): pos = line_position(weights, self.pitch_m, self.floor, self.lost) if pos is not None: self.last_m, self.last_t, self.status = pos, now_s, "on" return pos if self.last_t is not None and now_s - self.last_t <= self.hold_s: self.status = "holding" return self.last_m # short gap: carry on with the last position self.status = "lost" return None def side(self): """Which side the line was last seen on: 'left', 'right', or None if it was never seen.""" if self.last_m is None: return None return "right" if self.last_m > 0 else "left" class LinearCalibration: """reading = gain x true + offset, and the inverse true = (reading - offset) / gain.""" def __init__(self, gain, offset): self.gain = gain self.offset = offset def apply(self, reading): return (reading - self.offset) / self.gain @classmethod def two_point(cls, t1, r1, t2, r2): """Exact at both points: the readings r1 and r2 at the true values t1 and t2.""" gain = (r2 - r1) / (t2 - t1) return cls(gain, r1 - gain * t1) @classmethod def fit(cls, true_values, readings): """Least squares: the line through all the points that makes the sum of squared errors smallest. The true values are the known ones (the x axis); the readings carry the noise.""" mt, mr = mean(true_values), mean(readings) sxx = sum((t - mt) ** 2 for t in true_values) sxy = sum((t - mt) * (r - mr) for t, r in zip(true_values, readings)) gain = sxy / sxx return cls(gain, mr - gain * mt) def residuals(self, true_values, readings): """Calibrated value minus true value, for each point.""" return [self.apply(r) - t for t, r in zip(true_values, readings)] class IRSensor: """Test rig: an infrared triangulation sensor. ADC counts of V = k / (d + d0) plus noise: strongly nonlinear in distance. It reads the TRUE range from the simulator and turns it into counts.""" def __init__(self, sim, rng, k_vm=0.27, d0_m=0.02, noise_v=0.010, vref_v=3.3, bits=12): self.sim = sim self.rng = rng self.k_vm = k_vm self.d0_m = d0_m self.noise_v = noise_v self.vref_v = vref_v self.full = (1 << bits) - 1 def counts(self): d = self.sim.range_sensors["front"].true_range(self.sim.pose()) volts = 0.0 if d is None else self.k_vm / (d + self.d0_m) volts += self.rng.gauss(0.0, self.noise_v) return int(min(self.full, max(0, round(volts / self.vref_v * self.full)))) def delta_ticks(new, old, bits): """The signed change of a counter that wraps around at 2**bits. Right while the true change is under 2**(bits-1).""" full = 1 << bits return ((new - old + full // 2) % full) - full // 2 class WrappingTicks: """Wraps a hardware object so that ticks() wraps around at 2**bits, like a real counter register.""" def __init__(self, hw, bits): self.hw = hw self.mask = (1 << bits) - 1 def ticks(self): left, right = self.hw.ticks() return (left & self.mask, right & self.mask) def __getattr__(self, name): return getattr(self.hw, name) class SpeedEstimator: """Speed from encoder counts: the change over the last `window` passes divided by the REAL time they took, then an exponential filter (alpha = 1: none). Counts that wrap around (bits) are unwrapped first.""" def __init__(self, metres_per_tick, window=1, alpha=1.0, bits=None): self.mpt = metres_per_tick self.window = window self.alpha = alpha self.bits = bits self.last = None # the previous raw count self.total = 0 # unwrapped total of ticks self.hist = deque(maxlen=window + 1) # (total, time) of the last window + 1 passes self.speed_mps = 0.0 def update(self, ticks, now_s): if self.last is None: self.last = ticks step = delta_ticks(ticks, self.last, self.bits) if self.bits else ticks - self.last self.last = ticks self.total += step self.hist.append((self.total, now_s)) old_total, old_t = self.hist[0] if now_s > old_t: raw = (self.total - old_total) * self.mpt / (now_s - old_t) self.speed_mps += self.alpha * (raw - self.speed_mps) return self.speed_mps @property def distance_m(self): return self.total * self.mpt class Health: """Watches one Sensor and names what is wrong with it: 'stuck', 'bad' or 'stale' (None when it looks fine). stuck: the same value, exactly, stuck_n valid readings in a row. A real sensor with noise never does that. bad: bad_max invalid readings in a row (None, NaN, out of range). stale: no valid reading for the sensor's timeout.""" def __init__(self, sensor, stuck_n=20, bad_max=10): self.sensor = sensor self.stuck_n = stuck_n self.bad_max = bad_max self.last = None self.same_run = 0 self.fault = None def update(self): """Call once after every sensor.read().""" s = self.sensor if s.valid and not s.clamped: self.same_run = self.same_run + 1 if s.value == self.last else 0 self.last = s.value if self.same_run >= self.stuck_n: self.fault = "stuck" elif s.bad_run >= self.bad_max: self.fault = "bad" elif s.stale(): self.fault = "stale" else: self.fault = None return self.fault class FilteredHardware: """Wraps a hardware object. range_m() returns the cleaned range: validated, filtered and health-checked. It returns None when the range cannot be trusted, so the robot's own timeout logic takes over. The rest passes through.""" def __init__(self, hw, pipeline, low=0.02, high=4.0, timeout_s=0.11, stuck_n=20): self.hw = hw self.pipeline = pipeline self.sensor = Sensor(hw.range_m, hw.now_s, low, high, timeout_s, name="range") self.health = Health(self.sensor, stuck_n) def range_m(self): raw = self.sensor.read() fault = self.health.update() if raw is not None and fault != "stuck": self.pipeline.update(raw) # only good, moving readings go in return None if fault else self.pipeline.value def __getattr__(self, name): return getattr(self.hw, name) def approx(a, b, tol=1e-9): return abs(a - b) <= tol def test_constant_input_passes_through(): for f in (MovingAverage(5), MedianFilter(5), ExpFilter(0.25), Pipeline(MedianFilter(5), ExpFilter(0.5))): for _ in range(30): y = f.update(2.5) assert approx(y, 2.5, 1e-12), type(f).__name__ def test_moving_average_step(): f = MovingAverage(5) for _ in range(10): f.update(0.0) out = [f.update(1.0) for _ in range(6)] assert all(approx(a, b) for a, b in zip(out, [0.2, 0.4, 0.6, 0.8, 1.0, 1.0])), out def test_median_step(): f = MedianFilter(5) for _ in range(10): f.update(0.0) out = [f.update(1.0) for _ in range(6)] assert out == [0.0, 0.0, 1.0, 1.0, 1.0, 1.0], out def test_exponential_step(): f = ExpFilter(0.25) f.update(0.0) out = [f.update(1.0) for _ in range(3)] assert all(approx(a, b) for a, b in zip(out, [0.25, 0.4375, 0.578125])), out def test_spike_median_ignores_it_average_smears_it(): med, ma = MedianFilter(5), MovingAverage(5) for f in (med, ma): for _ in range(10): f.update(1.0) stream = [50.0, 1.0, 1.0, 1.0, 1.0, 1.0] med_out = [med.update(x) for x in stream] ma_out = [ma.update(x) for x in stream] assert all(approx(y, 1.0) for y in med_out), med_out assert all(approx(y, 1.0 + 49.0 / 5) for y in ma_out[:5]) and approx(ma_out[5], 1.0), ma_out def test_bad_readings_are_ignored(): makers = (lambda: MovingAverage(5), lambda: MedianFilter(5), lambda: ExpFilter(0.3), lambda: Pipeline(MedianFilter(3), ExpFilter(0.5))) for make in makers: a, b = make(), make() for x in (1.0, 2.0, 3.0): a.update(x) b.update(x) before = a.value for bad in (None, float("nan"), float("inf"), -float("inf")): assert a.update(bad) == before, (type(a).__name__, bad) assert approx(a.update(4.0), b.update(4.0)), type(a).__name__ def test_first_bad_reading_gives_none(): assert MovingAverage(3).update(float("nan")) is None assert ExpFilter(0.5).update(None) is None def test_exponential_time_constant(): f = ExpFilter.from_tau(0.2, 0.02) assert approx(f.alpha, 0.02 / 0.22) and approx(f.tau_s(0.02), 0.2) def test_slew_limiter(): s = SlewLimiter(1.0) s.update(0.0, 0.1) up = [s.update(10.0, 0.1) for _ in range(3)] assert all(approx(a, b) for a, b in zip(up, [0.1, 0.2, 0.3])), up assert approx(s.update(0.25, 0.1), 0.25) def test_schmitt_does_not_chatter(): s = Schmitt(0.25, 0.30) out = [s.update(x) for x in (0.5, 0.28, 0.26, 0.249, 0.26, 0.29, 0.31, 0.29)] assert out == [False, False, False, True, True, True, False, False], out def test_debouncer_needs_n_in_a_row(): d = Debouncer(3) out = [int(d.update(x)) for x in (1, 1, 0, 1, 1, 1, 0, 0, 1, 0, 0, 0)] assert out == [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 0], out def test_counter_wrap(): assert delta_ticks(5, 250, 8) == 11 and delta_ticks(250, 5, 8) == -11 and delta_ticks(100, 90, 8) == 10 def test_fuse(): x, var = fuse([(1.0, 4.0), (3.0, 4.0)]) assert approx(x, 2.0) and approx(var, 2.0), (x, var) # equal variances: the plain mean, half the variance x, var = fuse([(None, 1.0), (5.0, 9.0)]) assert approx(x, 5.0) and approx(var, 9.0), (x, var) # a missing reading is skipped assert fuse([]) == (None, None) and fuse([(float("nan"), 1.0)]) == (None, None) def test_calibration_hits_its_points(): c = LinearCalibration.two_point(10.0, 12.4, 100.0, 102.9) assert approx(c.apply(12.4), 10.0) and approx(c.apply(102.9), 100.0) exact = LinearCalibration.fit([1.0, 2.0, 3.0], [5.0, 7.0, 9.0]) # reading = 2 true + 3 assert approx(exact.gain, 2.0) and approx(exact.offset, 3.0) assert all(approx(e, 0.0) for e in exact.residuals([1.0, 2.0, 3.0], [5.0, 7.0, 9.0])) def test_line_position_and_lost_line(): assert line_position([0, 0, 0, 0, 0], 0.015) is None assert approx(line_position([0, 0, 1, 0, 0], 0.015), 0.0) assert approx(line_position([0, 0, 0, 1, 0], 0.015), 0.015) assert approx(line_position([0, 1, 1, 0, 0], 0.015), -0.0075) t = LineTracker(0.015, hold_s=0.1) assert t.update([0, 0, 0, 1, 0], 0.0) == 0.015 and t.status == "on" assert t.update([0, 0, 0, 0, 0], 0.05) == 0.015 and t.status == "holding" assert t.update([0, 0, 0, 0, 0], 0.20) is None and t.status == "lost" and t.side() == "right" def test_sensor_rejects_garbage_and_goes_stale(): clock = ManualClock() s = Sensor(lambda: float("nan"), clock.now_s, 0.02, 4.0, 0.05) assert s.read() is None and not s.valid and s.stale() values = iter([1.0, float("nan"), float("nan"), float("nan")]) s = Sensor(lambda: next(values), clock.now_s, 0.02, 4.0, 0.05) s.read() for _ in range(3): clock.t += 0.02 s.read() assert not s.valid and s.value == 1.0 and s.stale() and s.bad_run == 3 def test_health_names_a_stuck_sensor(): clock = ManualClock() s = Sensor(lambda: 1.0, clock.now_s, 0.02, 4.0, 0.05) h = Health(s, stuck_n=5) faults = [] for _ in range(8): s.read() faults.append(h.update()) clock.t += 0.02 assert faults == [None] * 5 + ["stuck"] * 3, faults def test_filters_stay_inside_the_data(): rng = random.Random(7) ma, med = MovingAverage(7), MedianFilter(7) window = deque(maxlen=7) for _ in range(500): x = rng.uniform(-3.0, 9.0) window.append(x) for f in (ma, med): y = f.update(x) assert min(window) - 1e-9 <= y <= max(window) + 1e-9 def run_tests(namespace): """Run every function called test_* in the namespace; print one line each.""" failures = 0 names = sorted(n for n, f in namespace.items() if n.startswith("test_") and callable(f)) for name in names: try: namespace[name]() except AssertionError as err: print("FAIL", name, err) failures += 1 else: print("ok ", name) print("%d tests, %d failed" % (len(names), failures)) return failures if __name__ == "__main__": run_tests(globals())

13. Challenges

Work these with Python, the simulator and the module of this chapter. The solutions are text files with the working, and the code in them was run, so the output shown is what it printed.

Challenge 1: Choose the filter for a worse sensor
A cheaper range sensor has a noise of 8 mm (Gaussian) and 4 % of wrong echoes (uniform between 0.05 m and 3.0 m). The robot is parked 0.93 m from a wall. The requirements: the total filter delay at most 4 samples at 50 Hz, and the lowest possible root-mean-square error against the true range over 6000 readings. (a) Using the formulas of sections 4 and 5, predict for the pipeline “median of N, then exponential filter with α” with N = 3, 5, 7 and the α that uses up the rest of the delay budget (the delays add): the delay, the chance that the median fails, and the noise left after the exponential filter alone (σ × √(α / (2 − α)), ignoring the median’s own smoothing). (b) Measure the rms error and the number of outputs more than 25 mm off for the same pipelines on the simulator with FaultyHardware, and for a 9-point moving average and the raw readings. (c) Choose one, and say how close the prediction was.
View solution
Challenge 2: A counter that wraps, and a loop that stalls
The encoders of a robot have 12-bit counters. (a) Work out the longest time between two readings for which the modular difference is right at the top speed of 0.4 m/s (Programming for Robotics 1 section 6 gives 0.340 mm per tick). (b) With WrappingTicks(sim, 12) and SpeedEstimator(..., bits=12), drive straight for 3 m at duty 0.5 with a 20 ms loop and compare the estimated distance with the true one; count how many times the counter wrapped. (c) Drive at duty 1.0 with the loop running every 1.5 s and then every 2.0 s (calling sim.advance with that interval, a loop that stalls) and show what the estimator reports in each case against the true distance, and explain the second result.
View solution
Challenge 3: Fusion with a sensor that freezes
The robot drives at 0.1 m/s (duty 0.25) towards a wall from 1.63 m away for 8 s, with two range sensors at the same mount: A (5 mm noise) and B (2 mm noise, added with add_range_sensor). At t = 4 s sensor B freezes on its last value. Compute three estimates of the range at every pass: (i) A alone, (ii) the inverse-variance fusion of A and B, (iii) the same fusion but with B wrapped in a Sensor and a Health monitor (stuck_n = 10), skipping B when it has a fault. Report the rms error of each against the true range for t < 4 s and for t ≥ 4 s, predict the error of (ii) after the freeze with the weights, and say what (iii) does in the 0.2 s before the freeze is noticed.
View solution

Quick reference

  • The chain: hardware, then Sensor (valid? in range? age?), then filters (median first, then average or exponential), then Health (stuck, bad, stale), then the robot, where None for too long means stop. A bad input is ignored, never averaged in.
  • Sensor: read() returns the value or None and never raises; out-of-range is invalid, or clamped and flagged with clamp=True; age = now − time of the newest valid value; stale when age > timeout.
  • Moving average: y = total / count, total += new − oldest, with a ring buffer: O(1). Delay (N − 1) / 2 samples; noise × 1 / √N (independent readings only). Median (N odd): ignores up to (N − 1) / 2 wild values; fails with probability sum C(N, k) pk (1 − p)N−k, k from (N + 1) / 2; delay (N − 1) / 2 on a step.
  • Exponential: y += α (x − y); τ = dt (1 − α) / α, α = dt / (τ + dt); delay (1 − α) / α samples; noise × √(α / (2 − α)). α = 2 / (N + 1) matches an N-point average in delay and in noise. Start at the first reading.
  • Others: slew limiter (at most rate × dt per pass: a ramp for commands, a spike guard for measurements); Schmitt trigger (two thresholds, a gap of several sigmas); debouncer (N readings in a row, N passes of delay).
  • Choosing: measure a noise record: mean, sigma, robust sigma = 1.4826 × MAD, outlier rate beyond 4 robust sigmas, histogram. Outliers: median first. Noise: smooth, and count the lag D × v × dt. Update a filter when there is a new reading, not when the loop runs.
  • Fusion: x = sum(xi / vi) / sum(1 / vi), v = 1 / sum(1 / vi): always better than the best input if the variances are right; equal weights can be worse than the better sensor; biases and common errors are not averaged away.
  • Line: w = (raw − lo) / (hi − lo) limited to 0..1; xi = (i − (n − 1) / 2) × pitch, positive to the right; position = sum(w x) / sum(w); lost if sum(w) < 0.2; a noise floor of 0.05 to 0.1; check the calibration span; hold the last position briefly, then report lost and the side.
  • Calibration: reading = gain × true + offset; two points: gain = (r2 − r1) / (t2 − t1); least squares: gain = Sxy / Sxx, offset = mean r − gain × mean t; average many readings per point, keep a check point, look at the residuals. Infrared: fit 1 / reading against distance.
  • Encoders: change of an N-bit counter = ((new − old + 2N−1) mod 2N) − 2N−1, valid below 2N−1 ticks per reading. Speed resolution = metres per tick / dt (= 1 / (CPR × dt) rev/s); spread = resolution × √(f (1 − f)); a longer window and an exponential filter trade delay for steadiness; use real timestamps.
  • Health: stuck (the same value stuck_n times), bad (bad_max invalid in a row), stale (age > timeout); a fault makes range_m() return None, and the robot’s own timeout gives the safe stop. Log both the cause and the stop.
  • Tests: constant, step, spike, garbage and a random property test; the simulator for noise ratios and delays; fault injection (FaultyHardware) for the failures the easy noise never shows.

Next: Programming for Robotics 5, Driving: Motors and Odometry

This chapter made the robot’s senses trustworthy: readings that say whether they are valid, filters whose delay is known, a speed that does not flicker by one tick, a calibration, and a monitor that stops the robot when a sensor lies. Programming for Robotics 5 (planned; not yet written) turns to the other half of the loop, the robot’s body. It writes the motor driver code (ramping with the slew limiter of this chapter, speed limits, the dead zone), uses the smoothed wheel speeds of section 9 to know what each wheel is really doing, and updates the robot’s pose from the encoder counts, the odometry whose drift Programming for Robotics 3 measured against the formula. It then drives straight lines and turns by an angle with it, and calibrates the wheel diameter and the track width from measured runs, so that the robot goes where it is told. The hardware notes for the build robot’s motors are there too.