132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
"""Read-only digital logic probe for a conditioned ESP32-C3 GPIO input.
|
|
|
|
The ESP32 pin is always configured as an input with no internal pull. This is
|
|
for observation only: it must receive a signal already conditioned to the
|
|
ESP32-C3 input range. It cannot make an unknown coffee-machine wire safe.
|
|
"""
|
|
|
|
from machine import Pin, time_pulse_us
|
|
import time
|
|
|
|
|
|
class DigitalProbe:
|
|
"""Measure level, rising-edge frequency, pulse widths, and transitions."""
|
|
|
|
def __init__(self, pin_id, reporter=None):
|
|
self.pin_id = int(pin_id)
|
|
self.reporter = reporter
|
|
# Do not enable a pull resistor: the observed circuit must determine
|
|
# the level, and the probe must not intentionally bias it.
|
|
self.signal = Pin(self.pin_id, Pin.IN)
|
|
self._edge_count = 0
|
|
|
|
def _emit(self, message):
|
|
print(message)
|
|
if self.reporter is not None:
|
|
self.reporter(message)
|
|
|
|
def _on_rising_edge(self, _pin):
|
|
# Keep IRQ work deliberately minimal. Do not print or allocate here.
|
|
self._edge_count += 1
|
|
|
|
def state(self):
|
|
"""Print and return the present digital level (0 or 1)."""
|
|
value = self.signal.value()
|
|
self._emit("GPIO%d = %d" % (self.pin_id, value))
|
|
return value
|
|
|
|
def frequency(self, seconds=2):
|
|
"""Count rising edges during a fixed window and return approximate Hz."""
|
|
duration_ms = int(float(seconds) * 1000)
|
|
if duration_ms <= 0:
|
|
raise ValueError("seconds must be greater than zero")
|
|
|
|
self._edge_count = 0
|
|
start = time.ticks_ms()
|
|
self.signal.irq(trigger=Pin.IRQ_RISING, handler=self._on_rising_edge)
|
|
try:
|
|
time.sleep_ms(duration_ms)
|
|
finally:
|
|
self.signal.irq(handler=None)
|
|
|
|
elapsed_ms = max(1, time.ticks_diff(time.ticks_ms(), start))
|
|
count = self._edge_count
|
|
hz = (count * 1000.0) / elapsed_ms
|
|
self._emit("Freq: %.2f Hz (%d rising edges / %.3fs)" % (
|
|
hz, count, elapsed_ms / 1000.0))
|
|
return hz
|
|
|
|
def pulse(self, timeout_us=1_000_000):
|
|
"""Measure one HIGH then one LOW pulse; return their timing data.
|
|
|
|
``time_pulse_us`` returns -2 when a requested level never arrives and
|
|
-1 when that level does not end before the timeout.
|
|
"""
|
|
timeout_us = int(timeout_us)
|
|
if timeout_us <= 0:
|
|
raise ValueError("timeout_us must be greater than zero")
|
|
|
|
high_us = time_pulse_us(self.signal, 1, timeout_us)
|
|
low_us = time_pulse_us(self.signal, 0, timeout_us)
|
|
if high_us < 0 or low_us < 0:
|
|
self._emit("No complete pulse: HIGH=%d LOW=%d" % (high_us, low_us))
|
|
self._emit("-2 = level did not arrive; -1 = pulse did not end in time")
|
|
return None
|
|
|
|
period_us = high_us + low_us
|
|
if period_us <= 0:
|
|
self._emit("No complete pulse")
|
|
return None
|
|
|
|
hz = 1_000_000.0 / period_us
|
|
duty = (high_us * 100.0) / period_us
|
|
result = {
|
|
"high_us": high_us,
|
|
"low_us": low_us,
|
|
"period_us": period_us,
|
|
"frequency_hz": hz,
|
|
"duty_percent": duty,
|
|
}
|
|
self._emit("HIGH: %dus LOW: %dus" % (high_us, low_us))
|
|
self._emit("Freq: %.2f Hz Duty: %.1f%%" % (hz, duty))
|
|
return result
|
|
|
|
def watch(self, seconds=5, max_events=200):
|
|
"""Poll and print transitions with microsecond timestamps.
|
|
|
|
This is intended for short diagnostic captures. Narrow pulses can be
|
|
missed because output and polling run in Python; use ``frequency`` or
|
|
dedicated capture hardware for high-rate or timing-critical signals.
|
|
"""
|
|
duration_ms = int(float(seconds) * 1000)
|
|
max_events = int(max_events)
|
|
if duration_ms <= 0:
|
|
raise ValueError("seconds must be greater than zero")
|
|
if max_events <= 0:
|
|
raise ValueError("max_events must be greater than zero")
|
|
|
|
last = self.signal.value()
|
|
start_us = time.ticks_us()
|
|
deadline_ms = time.ticks_add(time.ticks_ms(), duration_ms)
|
|
events = 0
|
|
while time.ticks_diff(deadline_ms, time.ticks_ms()) > 0:
|
|
value = self.signal.value()
|
|
if value == last:
|
|
continue
|
|
now_us = time.ticks_us()
|
|
self._emit("%9dus %d" % (time.ticks_diff(now_us, start_us), value))
|
|
last = value
|
|
events += 1
|
|
if events >= max_events:
|
|
self._emit("Capture stopped after %d events" % max_events)
|
|
break
|
|
return events
|
|
|
|
|
|
def open_probe(pin_id, reporter=None):
|
|
"""Create a probe and print the interactive REPL commands."""
|
|
probe = DigitalProbe(pin_id, reporter)
|
|
probe._emit("ESP32-C3 read-only logic probe on GPIO%d" % probe.pin_id)
|
|
probe._emit("Use: probe.state(), probe.frequency(), probe.pulse(), probe.watch()")
|
|
return probe
|