Source code for openbricks.drivers.qtr

# SPDX-License-Identifier: MIT
"""
Pololu QTR / QTRX reflectance sensor arrays (analog outputs).

A row of IR emitter/phototransistor pairs a few millimetres above the
mat: dark line = high reading, light mat = low. Unlike a pair of
colour sensors, the array gives a CONTINUOUS line position — a
weighted centroid across all elements — so a follower steers on a
real analog error instead of edge-crossings.

Wiring (QTRX-HD-15A on ESP32-S3): the analog outputs go to ADC1 pins
(GPIO 1..10 — ADC2 fights the radios). Any subset of the array's
channels works; pass the pins left-to-right as mounted, and set
``pitch_mm`` to the spacing of the channels you actually wired
(4 mm for adjacent QTRX-HD channels, 8 mm if every other one).
CTRL (emitter enable) may be tied high or given a pin.

Readings are ratiometric to whatever height and mat you mounted over,
so the array MUST be calibrated once per session: sweep it across the
line while ``calibrate()`` runs. Reading before calibration raises —
an uncalibrated centroid is a plausible-looking wrong number.

Steering is the program's job, from the element readings. The three
canonical patterns (``ambient()`` is 0 black .. 100 white after
calibration):

* Edge following — pick the element the line's edge should sit
  under: ``steer = KP * (50 - r[i].ambient())`` for a RIGHT edge
  (element i just inside the right edge reads darker as the robot
  drifts left), ``steer = KP * (r[i].ambient() - 50)`` for a LEFT
  edge. The error is proportional within about one pitch of that
  element and rails beyond.
* Centre following — ``steer = KP_MM * r.position()`` (mm, positive
  = line right of centre); ``position()`` is ``None`` when the line
  is outside the window, so the program keeps the side it was last
  seen on and steers that way.
* Intersection — ``r.all_dark()``. Branch marker — the far-side
  elements (the ones beyond the edge you follow) going dark.

Example (the front window on GPIO 1..8, a two-element rear array on
GPIO 9/10, each with its own calibration file)::

    from openbricks.drivers.qtr import QTRArray, QTRLineSensor

    front = QTRLineSensor(channels=8)        # -28..+28 mm, 8 mm pitch
    rear = QTRArray(pins=(9, 10), pitch_mm=8.0)
    front.load_calibration("/qtr_front.cal")
    rear.load_calibration("/qtr_rear.cal")
    while True:
        r = front.read()
        steer = KP * (50 - r[5].ambient())   # right edge under +12 mm
        if r.all_dark():
            break                            # intersection
        if r[6].dark() or r[7].dark():
            pass                             # branch on the far side
        if rear.read().all_dark():
            pass                             # the rear crossed it too
"""

import time

from openbricks import pins as _pins

_FULL_SCALE = 1000


[docs] class QTRElement: """One calibrated element reading, as returned by :meth:`QTRArray.read`. ``value`` is the 0 (mat) .. 1000 (line) int; ``dark()`` / ``white()`` answer per element exactly like :meth:`QTRChannel.dark`. Deliberately NOT an int subclass: MicroPython cannot reflect-compare ``int`` against one, so ``max()`` over such readings raises on the hub while passing on the desktop — numeric code uses ``.value`` instead (``max(e.value for e in readings)``).""" def __init__(self, value, threshold): self.value = value self._threshold = threshold
[docs] def dark(self): """True when this element is over the line.""" return self.value >= self._threshold
[docs] def white(self): """True when this element is over the mat.""" return self.value < self._threshold
[docs] def ambient(self): """Reflected brightness as 0 (black) .. 100 (white) — the Pybricks scale, inverted from ``value``. The edge follower's error: ``50 - ambient()`` is zero exactly when the element straddles the black/white boundary.""" return (_FULL_SCALE - self.value) * 100 // _FULL_SCALE
def __repr__(self): return "QTRElement(%d, %s)" % ( self.value, "dark" if self.dark() else "white")
[docs] class QTRReading: """One snapshot of the whole array, as returned by :meth:`QTRArray.read`. Behaves like the list of :class:`QTRElement` it holds (index, iterate, ``len``) and carries every aggregate view of exactly this snapshot — so a control tick reads once and derives everything from the same coherent sample:: r = qtr.read() r[0].dark() # per element r[7].ambient() # 0 black .. 100 white (edge error) r.position() # global centroid, mm r.left_edge_position() # the line's left edge, mm r.right_edge_position() # the line's right edge, mm r.leftmost_position() # fork clusters r.rightmost_position() r.dark_count() # how many elements on the line r.all_dark() # whole array on the line r.max() # brightest value (phantom gate) """ def __init__(self, elements, array): self.elements = elements self._array = array def __len__(self): return len(self.elements) def __getitem__(self, i): return self.elements[i] def __iter__(self): return iter(self.elements)
[docs] def values(self): """The plain 0..1000 ints, left to right.""" return [e.value for e in self.elements]
[docs] def max(self): """The brightest calibrated value in this snapshot — the follower's off-mat phantom gate.""" best = 0 for e in self.elements: if e.value > best: best = e.value return best
[docs] def dark_count(self): return self._array.dark_count(self)
[docs] def all_dark(self): """Every element over the line — with the branch flag dark too, the whole crossing is under the robot (the follower's stop condition).""" for e in self.elements: if not e.dark(): return False return True
[docs] def position(self): return self._array.position(self)
[docs] def left_edge_position(self): return self._array.left_edge_position(self)
[docs] def right_edge_position(self): return self._array.right_edge_position(self)
[docs] def leftmost_position(self): return self._array.leftmost_position(self)
[docs] def rightmost_position(self): return self._array.rightmost_position(self)
def __repr__(self): return "QTRReading(%s)" % ",".join( str(e.value) for e in self.elements)
[docs] class QTRArray: """Analog QTR/QTRX reflectance array on ESP32 ADC pins. Args: pins: ADC-capable GPIO numbers, LEFT to RIGHT as mounted. pitch_mm: physical spacing between the wired channels. ctrl: optional emitter-control GPIO (QTRX CTRL). Driven high at construction (emitters on). ``None`` = tied high. dark_threshold: calibrated value (0..1000) above which an element counts as "over the line" for ``position()`` / ``dark_count()``. Each array owns its pins for the run: a second array (or :class:`QTRChannel`) naming a pin this one already holds is refused at construction, by GPIO number and both arrays. Two arrays on disjoint pins coexist — the front window on GPIO 1..8 and a rear pair on GPIO 9/10 — and each calibrates and stores its own file (:meth:`save_calibration` records the wiring, so the files cannot be swapped by accident). """ def __init__(self, pins, pitch_mm=4.0, ctrl=None, dark_threshold=300, positions_mm=None): """``positions_mm`` (optional): explicit per-element x coordinates in mm, left to right, strictly increasing — for wirings whose channels are NOT evenly spaced (e.g. the skip pattern QTRX ch 15,13,12,11,9,7,5,4,3,1: spacings 8/4/4/8/8/8/4/4/8 mm span a 56 mm window on ten pins). The origin is wherever the caller puts it; centring the tuple on 0 keeps ``position()`` symmetric. When given, ``pitch_mm`` only seeds the off-array edge saturation via the MEAN spacing; all real geometry comes from the positions.""" if len(pins) < 1: raise ValueError("at least one element required") role = "QTR array on GPIO %s" % ",".join(str(int(p)) for p in pins) for p in pins: _pins.check(p, role, output=False) self._check_adc_capable(p) for p in pins: _pins.claim(p, role) self._pins = tuple(int(p) for p in pins) self._threshold = int(dark_threshold) self._adcs = [self._make_adc(p) for p in pins] self._ctrl = None if ctrl is not None: from machine import Pin self._ctrl = Pin(ctrl, Pin.OUT, value=1) # emitters on n = len(self._adcs) if positions_mm is not None: if len(positions_mm) != n: raise ValueError( "positions_mm has %d entries for %d pins" % (len(positions_mm), n)) xs = [float(x) for x in positions_mm] for i in range(1, n): if xs[i] <= xs[i - 1]: raise ValueError( "positions_mm must be strictly increasing " "left to right, got %r" % (positions_mm,)) self._x_mm = xs self._pitch = ((xs[-1] - xs[0]) / (n - 1) if n > 1 else float(pitch_mm)) else: self._pitch = float(pitch_mm) # Element x-positions in mm, centre of the wired span at 0. self._x_mm = [(i - (n - 1) / 2.0) * self._pitch for i in range(n)] self._cal_min = None self._cal_max = None @property def positions_mm(self): """The element x coordinates in mm, left to right — the same frame as :meth:`position`. Name elements by where they sit: ``r[qtr.positions_mm.index(12.0)]`` is the element 12 mm right of centre.""" return tuple(self._x_mm) @staticmethod def _check_adc_capable(pin): """Refuse pins with no usable ADC, BY NAME, at construction. ``pins.check`` validates the board's GPIO map but not analog capability, and ``machine.ADC`` fails with a bare ValueError — after the user already soldered the harness (bench 2026-08-06: five channels landed on GPIO 38-42, which have no ADC at all on the S3). """ chip = _pins._detect_chip() if chip == "esp32s3": if not 1 <= pin <= 10: if 11 <= pin <= 20: raise ValueError( "GPIO %d is ADC2 on the ESP32-S3 — unusable " "for the QTR array: ADC2 is shared with the " "radio (BLE is up during `openbricks run`) " "and unreliable by errata. Use ADC1, GPIO " "1-10." % pin) raise ValueError( "GPIO %d has no ADC on the ESP32-S3 — analog " "inputs must be ADC1, GPIO 1-10." % pin) elif chip == "esp32": if not 32 <= pin <= 39: raise ValueError( "GPIO %d is not an ADC1 pin on the classic ESP32 " "— use GPIO 32-39 (ADC2 is shared with the " "radio)." % pin) @staticmethod def _make_adc(pin): from machine import ADC, Pin adc = ADC(Pin(pin)) # Full 0..~3.3 V range; the QTRX output swings to its supply. # Ports without attenuation control (unix) simply skip it. try: adc.atten(ADC.ATTN_11DB) except AttributeError: pass return adc def _read_u16(self): return [adc.read_u16() for adc in self._adcs] # ---- calibration -------------------------------------------------
[docs] def calibrate(self, duration_ms=3000, poll_ms=5): """Learn each element's mat/line extremes. Sweep the array across the line while this runs (rotate the robot, or slide it by hand). Extends any previous calibration rather than replacing it, so repeated calls refine. """ if self._cal_min is None: self._cal_min = [65535] * len(self._adcs) self._cal_max = [0] * len(self._adcs) deadline = time.ticks_add(time.ticks_ms(), int(duration_ms)) while time.ticks_diff(deadline, time.ticks_ms()) > 0: for i, v in enumerate(self._read_u16()): if v < self._cal_min[i]: self._cal_min[i] = v if v > self._cal_max[i]: self._cal_max[i] = v time.sleep_ms(poll_ms) self._check_calibration()
def _check_calibration(self): if self._cal_min is None: raise RuntimeError( "QTR array is not calibrated — call calibrate() while " "sweeping the array across the line (an uncalibrated " "centroid is a plausible-looking wrong number)") # An element whose span never opened up saw only mat (or only # line, or is unwired): its normalized reading would be noise # amplified to full scale. Name it. flat = [i for i in range(len(self._adcs)) if self._cal_max[i] - self._cal_min[i] < 1024] if flat: raise RuntimeError( "QTR calibration saw no line/mat contrast on " "element(s) %s (left=0) — that channel is unwired, " "or the sweep never carried it across the line" % ",".join(str(i) for i in flat))
[docs] def save_calibration(self, path): """Persist the calibration to a file on the hub filesystem, so one sweep (``examples/qtr_calibrate.py``) serves every later run. The wiring is stored with it: a calibration is per- element min/max, so loading it onto different pins would silently mis-scale every reading. One file per array — the front window and a rear pair each keep their own (``"/qtr_front.cal"``, ``"/qtr_rear.cal"``).""" self._check_calibration() import json with open(path, "w") as f: json.dump({"pins": list(self._pins), "min": self._cal_min, "max": self._cal_max}, f)
[docs] def load_calibration(self, path): """Load a calibration saved by :meth:`save_calibration`. Raises with the remedy when the file is missing (run the calibrate script), corrupt, or was recorded for DIFFERENT wiring. Calibration is height- and mat-dependent — resweep after remounting the array or changing mats.""" import json try: with open(path) as f: data = json.load(f) except OSError: raise RuntimeError( "no saved QTR calibration at %s — run " "examples/qtr_calibrate.py once (sweep the array " "across the line while it runs)" % path) except ValueError: raise RuntimeError( "saved QTR calibration at %s is corrupt — delete it " "and re-run examples/qtr_calibrate.py" % path) if tuple(data.get("pins", ())) != self._pins: raise RuntimeError( "saved QTR calibration at %s was recorded for pins %s " "but this array is wired to %s — per-element min/max " "does not transfer across wiring; re-run " "examples/qtr_calibrate.py" % (path, tuple(data.get("pins", ())), self._pins)) self._cal_min = [int(v) for v in data["min"]] self._cal_max = [int(v) for v in data["max"]] self._check_calibration()
# ---- reading -----------------------------------------------------
[docs] def read(self): """One calibrated snapshot: a :class:`QTRReading` holding a :class:`QTRElement` per array element, left to right — ``r[i].value`` 0 (mat) .. 1000 (line), ``r[i].dark()`` / ``r[i].white()`` / ``r[i].ambient()``, and the aggregate views (``r.position()``, ``r.dark_count()``, ...) computed from exactly this sample.""" self._check_calibration() out = [] for i, v in enumerate(self._read_u16()): span = self._cal_max[i] - self._cal_min[i] n = (v - self._cal_min[i]) * _FULL_SCALE // span if n < 0: n = 0 elif n > _FULL_SCALE: n = _FULL_SCALE out.append(QTRElement(n, self._threshold)) return QTRReading(out, self)
[docs] def dark_count(self, readings=None): """How many elements are over the line — the intersection / stop-bar signal (a full-width bar darkens most of the array, a branch stub only one side).""" if readings is None: readings = self.read() return sum(1 for r in readings if r.dark())
[docs] def position(self, readings=None): """Line centre in mm relative to the array centre; positive = line is to the RIGHT. ``None`` when no element sees the line — it is outside the window, and the program steers toward the side it last saw the line on (keep that yourself from the sign of the last non-``None`` value). """ if len(self._adcs) < 2: raise RuntimeError( "position() needs at least 2 elements — a single " "detector channel has no centroid (use QTRChannel" ".dark() for a flag)") readings = self.read() if readings is None else readings weight_sum = 0 moment = 0.0 seen = False for r, x in zip(readings, self._x_mm): if r.dark(): seen = True weight_sum += r.value moment += r.value * x if not seen: return None return moment / weight_sum
def _cluster_position(self, readings, rightmost): """Centroid of the leftmost (or rightmost) contiguous dark cluster, with one sub-threshold neighbour each side joining the weighting — the same between-element interpolation position() has. ``None`` when nothing is dark.""" n = len(readings) order = range(n - 1, -1, -1) if rightmost else range(n) first = None for i in order: if readings[i].dark(): first = i break if first is None: return None last = first step = -1 if rightmost else 1 while 0 <= last + step < n and readings[last + step].dark(): last += step i0, i1 = (last, first) if rightmost else (first, last) lo = i0 - 1 if i0 > 0 else 0 hi = i1 + 1 if i1 + 1 < n else i1 weight_sum = 0 moment = 0.0 for i in range(lo, hi + 1): weight_sum += readings[i].value moment += readings[i].value * self._x_mm[i] return moment / weight_sum
[docs] def left_edge_position(self, readings=None): """Millimetre position of the line's LEFT edge — the white→black boundary on the left side of the leftmost dark cluster, in the same frame as :meth:`position` (positive = right of the array centre). ``None`` when nothing is dark. The crossing is interpolated between the last white element and the first dark one: the point where the calibrated value passes ``dark_threshold``. An edge follower keeps THIS at 0, so the robot straddles the boundary — half the array over mat, half over line — instead of centring on the line itself. That also makes line width irrelevant to steering. When the leftmost element is itself dark the true edge is off-array to the left; the estimate saturates half a pitch beyond that element, so the error keeps its sign and magnitude instead of vanishing. """ if len(self._adcs) < 2: raise RuntimeError( "left_edge_position() needs at least 2 elements — a " "single detector channel has no edge to interpolate " "(use QTRChannel.dark() for a flag)") readings = self.read() if readings is None else readings first = None for i, r in enumerate(readings): if r.dark(): first = i break if first is None: return None if first == 0: return self._x_mm[0] - self._pitch / 2 v0 = readings[first - 1].value v1 = readings[first].value frac = (self._threshold - v0) / (v1 - v0) # Interpolate over the ACTUAL spacing between these two # elements — the pitch on a uniform array, the local gap on # a positions_mm one. span = self._x_mm[first] - self._x_mm[first - 1] return self._x_mm[first - 1] + frac * span
[docs] def right_edge_position(self, readings=None): """Mirror of :meth:`left_edge_position`: the line's RIGHT edge — the black→white boundary on the right side of the rightmost dark cluster, in the same mm frame. ``None`` when nothing is dark; a dark rightmost element saturates half a pitch off-array to the right. A right-edge follower keeps THIS at 0 — the mirror-image track discipline of the left-edge follower, same sign convention (the P law is symmetric between the two).""" if len(self._adcs) < 2: raise RuntimeError( "right_edge_position() needs at least 2 elements — a " "single detector channel has no edge to interpolate " "(use QTRChannel.dark() for a flag)") readings = self.read() if readings is None else readings n = len(self._adcs) last = None for i in range(n - 1, -1, -1): if readings[i].dark(): last = i break if last is None: return None if last == n - 1: return self._x_mm[n - 1] + self._pitch / 2 v0 = readings[last + 1].value v1 = readings[last].value frac = (self._threshold - v0) / (v1 - v0) span = self._x_mm[last + 1] - self._x_mm[last] return self._x_mm[last + 1] - frac * span
[docs] def leftmost_position(self, readings=None): """Centroid of the LEFTMOST contiguous dark cluster, in mm (same frame as :meth:`position`), or ``None`` when nothing is dark. At a branch the array sees TWO dark regions and the global centroid lands between them — steering into the gap. The leftmost cluster is the left line's own centre: a follower that must take the left fork steers on this while its branch flag says a second line is present. Computed the same way every tick, so switching between the two is jump-free.""" readings = self.read() if readings is None else readings return self._cluster_position(readings, rightmost=False)
[docs] def rightmost_position(self, readings=None): """Mirror of :meth:`leftmost_position`: the RIGHTMOST contiguous dark cluster's centre — the right fork, for a route policy that takes it.""" readings = self.read() if readings is None else readings return self._cluster_position(readings, rightmost=True)
[docs] def emitters(self, on): """Drive the CTRL pin (no-op when CTRL is tied high).""" if self._ctrl is not None: self._ctrl.value(1 if on else 0)
[docs] class QTRChannel(QTRArray): """One reflectance element with the array's calibrate/read contract — for a DETECTOR channel wired apart from the line cluster (a branch / marker flag). Kept out of :class:`QTRArray` on purpose: a flag element folded into the steering centroid would yank the position toward every marker it passes. Steer on the array; DECIDE on this. Example (GPIO 9, one of the two ADC1 pins the 8-channel front window leaves free):: branch = QTRChannel(pin=9) branch.calibrate(duration_ms=3000) # same sweep as the array if branch.dark(): ... # marker under the flag channel """ def __init__(self, pin, dark_threshold=300): QTRArray.__init__(self, pins=(pin,), dark_threshold=dark_threshold)
[docs] def value(self): """Calibrated reading, 0 (mat) .. 1000 (marker/line).""" return self.read()[0].value
[docs] def dark(self): """True when the element is over a marker/line.""" return self.value() >= self._threshold
[docs] def white(self): """True when the element is over the mat.""" return not self.dark()
[docs] class QTRLineSensor(QTRArray): """THE bench line sensor: one QTRX-HD-15A window with its geometry pre-wired — construct it, pick an element, follow:: qtr = QTRLineSensor() # ten channels, GPIO 1..10 r = qtr.read() steer = KP * (50 - r[7].ambient()) # right edge under +16 mm Two layouts (detailed tables in docs/hardware.md), selected with ``channels``: * ``channels=10`` (default): QTRX channels 1, 3, 4, 5, 7, 9, 11, 12, 13, 15 left-to-right onto GPIO 1..10 in order — a 56 mm window at spacings 8/4/4/8/8/8/4/4/8 mm (the pattern is a palindrome, so board orientation only changes the channel LABELS, never the geometry). Right edge under index 7 (+16 mm), left edge under index 2 (-16 mm). * ``channels=8``: QTRX channels 1, 3, 5, 7, 9, 11, 13, 15 (every other one, 8 mm pitch) onto GPIO 1..8 — the same 56 mm window on eight pins, leaving GPIO 9 and 10 (the last two ADC1 pins on the ESP32-S3) for a SECOND array. Right edge under index 5 (+12 mm), left edge under index 2 (-12 mm). Both leave two mat-side elements beyond the followed edge for the branch watch. Different wiring: use :class:`QTRArray` directly with your own pins/positions. """ PINS = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) POSITIONS_MM = (-28.0, -20.0, -16.0, -12.0, -4.0, 4.0, 12.0, 16.0, 20.0, 28.0) PINS_8 = (1, 2, 3, 4, 5, 6, 7, 8) POSITIONS_MM_8 = (-28.0, -20.0, -12.0, -4.0, 4.0, 12.0, 20.0, 28.0) def __init__(self, channels=10, dark_threshold=300): if channels == 10: pins, positions = self.PINS, self.POSITIONS_MM elif channels == 8: pins, positions = self.PINS_8, self.POSITIONS_MM_8 else: raise ValueError( "channels must be 8 (GPIO 1-8, leaving 9/10 for a " "second array) or 10 (GPIO 1-10), got %r" % (channels,)) QTRArray.__init__(self, pins=pins, positions_mm=positions, dark_threshold=dark_threshold)