rf-scan-ft232h.py
Files / cc1101-scanner / rf-scan-ft232h.py
Files in cc1101-scanner/
Directorycc1101-scanner/
- 70-ftdi-ft232h.rules 196 bytes
- keeloq-capture-ft232h.py 9.5 KB
- keeloq-check.py 9.6 KB
- LICENSE 1.0 KB
- rf-scan-ft232h.py 41.0 KB
From CC1101 on an FT232H: a 433 MHz scanner in the terminal. MIT licence: LICENSE.
#!/usr/bin/env python3"""See what the CC1101 hears, on a CJMCU-232H (FT232H) board: a terminal screen with a spectrumsweep and a listen mode that shows pulses and decodes remote codes. Receive only: it never transmits.
sweep steps across a band reading the CC1101's signal strength (RSSI): bars with peak hold, a waterfall of past sweeps, and the strongest signals seen listen sits on one frequency (OOK, 650 kHz filter, as keeloq-capture-ft232h.py) and samples the demodulated data at 50 kHz: signal history, the last frame's pulses and bits, and every code heard more than once (fixed-code remotes: EV1527, PT2262 and the like), plus KeeLoq presses. The timings go to /tmp/rf-listen.log, in the format keeloq-check.py reads
The CC1101 tunes 300-348, 387-464 and 779-928 MHz. My E07-M1101D and its antenna are built for433 MHz, so the other bands hear only strong signals. Wiring: keeloq-capture-ft232h.py.
Usage, in this folder, with pyftdi in .venv: .venv/bin/python -I rf-scan-ft232h.py [--band 433] [--listen 433.92] --band 433, 315, 300-348, 387-464, 779-928, all, or any START-STOP in MHz --text plain lines instead of the screen; --seconds N stops after N s .venv/bin/python -I rf-scan-ft232h.py --replay /tmp/rf-listen.log .venv/bin/python -I rf-scan-ft232h.py --self-test [--preview]"""import argparseimport cursesimport importlib.utilimport localeimport mathimport queueimport randomimport reimport statisticsimport sysimport threadingimport timefrom collections import dequefrom pathlib import Path
HERE = Path(__file__).resolve().parent_spec = importlib.util.spec_from_file_location("keeloq_capture", HERE / "keeloq-capture-ft232h.py")cap = importlib.util.module_from_spec(_spec)_spec.loader.exec_module(cap)kc = cap.kc
XTAL = 26e6CC_BANDS = ((300.0, 348.0), (387.0, 464.0), (779.0, 928.0)) # MHz the CC1101 tunes with a 26 MHz crystalBANDS = [ # key for --band, label, start and stop MHz; b steps through them ("433", "433 MHz remotes", 432.0, 436.0), ("315", "315 MHz remotes", 313.0, 317.0), ("300-348", "low band", 300.0, 348.0), ("387-464", "middle band", 387.0, 464.0), ("779-928", "868/915 band", 779.0, 928.0), ("all", "all 3 bands", 300.0, 928.0),]RSSI = 0x34 # status registerSETTLE = 0.0012 # s after SRX before RSSI is real: calibration ends about 1 ms in (probed 2026-11-02)DBM_LO, DBM_HI = -110, -20 # graph rangeLISTEN_MDMCFG4 = cap.REGS[0x10] # 650 kHz filter, Flipper's AM650LISTEN_CHUNK = cap.RATE // 5 // 8 # bytes per read: 0.2 s of samplesPAIR_MIN, PAIR_MAX = 100, 3000 # µs: a bit's high and low both lie in betweenMIN_BITS = 12MAX_PENDING_US = 3_000_000 # cut the data even without a gap after this longLOG = Path("/tmp/rf-listen.log")LEFT = 5 # columns for the labels left of the graphsBLOCKS = " ▁▂▃▄▅▆▇█"RAMP = " .:-=+*#%@"
# Receive filter widths: 26 MHz / (8 (4 + M) 2^E), E and M 0 to 3, so 58 to 812 kHz;# MDMCFG4 = E << 6 | M << 4 | DRATE_E 7 (as in REGS)FILTERS = sorted((XTAL / (8 * (4 + m) * 2 ** e) / 1e3, e << 6 | m << 4 | 0x07) for e in range(4) for m in range(4))
def supported(mhz): return any(lo <= mhz <= hi for lo, hi in CC_BANDS)
def freq_word(mhz): w = round(mhz * 1e6 * 65536 / XTAL) return [w >> 16 & 0xFF, w >> 8 & 0xFF, w & 0xFF]
def filter_for(khz): """The narrowest receive filter at least khz wide, or the widest: (kHz, MDMCFG4).""" return next((f for f in FILTERS if f[0] >= khz), FILTERS[-1])
def dbm(raw): return (raw - 256 if raw >= 128 else raw) / 2 - 74
def frac(v): return min(max((v - DBM_LO) / (DBM_HI - DBM_LO), 0.0), 1.0)
def fmt(v): return "–" if v is None else f"{v:.0f} dBm"
def hms(t): return time.strftime("%H:%M:%S", time.localtime(t))
class Plan: """One sweep: cols columns from lo to hi MHz. A column measures more than one point when its step is wider than the widest filter, so wide bands leave no gaps; columns outside the CC1101's bands measure nothing."""
def __init__(self, lo, hi, cols): self.lo, self.hi, self.cols = lo, hi, max(cols, 2) self.step = (hi - lo) / (self.cols - 1) self.bw, self.mdmcfg4 = filter_for(self.step * 1e3) n = max(1, math.ceil(self.step * 1e3 / FILTERS[-1][0])) self.points = [] for i in range(self.cols): pts = (self.freq(i) + (k + 0.5 - n / 2) * self.step / n for k in range(n)) self.points.append([f for f in pts if supported(f)])
def freq(self, i): return self.lo + i * self.step
def column(self, mhz): return min(max(round((mhz - self.lo) / self.step), 0), self.cols - 1)
class Radio: """The CC1101 on the FT232H. Only the worker thread uses it."""
def __init__(self, url): self.spi, self.port, self.sampler = cap.connect(url) self.mdmcfg4 = cap.REGS[0x10]
def status(self, addr): return self.port.exchange([0xC0 | addr, 0], duplex=True)[1]
def tune(self, mhz, mdmcfg4): if mdmcfg4 != self.mdmcfg4: self.port.write([cap.SIDLE, 0x10, mdmcfg4]) self.mdmcfg4 = mdmcfg4 self.port.write([cap.SIDLE, 0x4D, *freq_word(mhz)]) # strobe, then a burst write of FREQ2 to FREQ0 self.port.write([cap.SRX]) # calibrates (MCSM0), then receives
def measure(self, mhz, mdmcfg4): self.tune(mhz, mdmcfg4) time.sleep(SETTLE) return dbm(self.status(RSSI))
def close(self): try: self.port.write([cap.SIDLE]) finally: self.spi.terminate()
class Worker(threading.Thread): """Owns the radio: sweeps or listens, whichever the screen last asked for in .want, and queues (want, payload, extra) results. Each result carries the request it answers, so stale ones drop."""
def __init__(self, radio): super().__init__(daemon=True) self.radio, self.want = radio, None self.results, self.stop = queue.Queue(), threading.Event()
def run(self): tuned = None try: while not self.stop.is_set(): want = self.want if want is None: time.sleep(0.01) elif want[0] == "sweep": plan, tuned, t, levels = want[1], None, time.monotonic(), [] for points in plan.points: if self.want is not want or self.stop.is_set(): break levels.append(max((self.radio.measure(f, plan.mdmcfg4) for f in points), default=None)) else: self.results.put((want, levels, time.monotonic() - t)) else: if tuned is not want: self.radio.tune(want[1], LISTEN_MDMCFG4) time.sleep(0.002) tuned = want data = bytes(self.radio.sampler.read(LISTEN_CHUNK)) self.results.put((want, data, dbm(self.radio.status(RSSI)))) except Exception as e: # USB unplugged and the like self.results.put((None, f"{type(e).__name__}: {e}", None))
class SweepView: """Sweep results for one plan: latest levels, peak hold, waterfall."""
def __init__(self, plan, label): self.plan, self.label = plan, label self.clear()
def clear(self): self.count, self.took = 0, 0.0 self.levels = [None] * self.plan.cols self.hold = [None] * self.plan.cols self.hold_t = [0.0] * self.plan.cols self.water = deque(maxlen=300) # (time, levels), newest first
def add(self, levels, took, when=None): when = time.time() if when is None else when self.levels, self.took, self.count = levels, took, self.count + 1 self.water.appendleft((when, levels)) for i, v in enumerate(levels): if v is not None and (self.hold[i] is None or v > self.hold[i]): self.hold[i], self.hold_t[i] = v, when
def floor(self): vals = [v for v in self.levels if v is not None] return statistics.median(vals) if vals else DBM_LO
def peaks(self, levels, above=10, n=5): """The strongest columns at least `above` dB over the noise floor, one per signal (columns within 200 kHz count as one): [(MHz, dBm, column)].""" floor = self.floor() cols = sorted((i for i, v in enumerate(levels) if v is not None and v >= floor + above), key=lambda i: -levels[i]) width = max(1, round(0.2 / self.plan.step)) out = [] for i in cols: if all(abs(i - j) > width for _, _, j in out): out.append((self.plan.freq(i), levels[i], i)) if len(out) == n: break return out
def near(x, ref): return abs(x - ref) <= 0.35 * ref
def cut_point(d): """Where pending timings can be cut without splitting a frame: before the last long silence or long noise high (no remote frame holds either). None if there is no such place.""" for i in range(len(d) - 1, 0, -1): if d[i] <= -cap.GAP_US or d[i] > PAIR_MAX: return i return None
def find_frames(d): """Fixed-code frames in raw timings (+µs high, -µs low): runs of at least MIN_BITS (high, low) pairs where one is short (1 TE) and the other long (2-4 TE), as EV1527, PT2262 and most other PWM remotes send. Bit 1 = long high. Returns [(bits, TE µs, timings)]."""
def pair_ok(k): h, l = d[k], -d[k + 1] return (h > 0 and l > 0 and PAIR_MIN <= h <= PAIR_MAX and PAIR_MIN <= l <= PAIR_MAX and max(h, l) >= 1.6 * min(h, l))
runs, run, k = [], [], 0 while k < len(d) - 1: if pair_ok(k): run.append(k) k += 2 continue if run: runs.append(run) run = [] k += 2 if d[k] > 0 > d[k + 1] else 1 # step to the next high if run: runs.append(run)
out = [] for run in runs: if len(run) < MIN_BITS - 1: continue s = statistics.median(min(d[k], -d[k + 1]) for k in run) l = statistics.median(max(d[k], -d[k + 1]) for k in run) part = [] for k in run + [None]: # split where a pair doesn't fit the run's timing if k is not None and near(min(d[k], -d[k + 1]), s) and near(max(d[k], -d[k + 1]), l): part.append(k) continue if part: fr = frame(d, part, s, l) if len(fr[0]) >= MIN_BITS: out.append(fr) part = [] return out
def frame(d, part, s, l): bits = "".join("1" if d[k] > -d[k + 1] else "0" for k in part) timings = d[part[0] : part[-1] + 2] k = part[0] # Noise fills the remote's gap before a frame and swallows the first bit's high (my rig, # 2026-11-02): when the high before the run is too long for a bit, read that bit from its low if k >= 2 and d[k - 2] > 1.35 * l and d[k - 1] < 0: low = -d[k - 1] if near(low, l): bits, timings = "0" + bits, [round(s), -low] + timings elif near(low, s): bits, timings = "1" + bits, [round(l), -low] + timings te = round(statistics.median(min(d[j], -d[j + 1]) for j in part)) return bits, te, timings
def hexcode(bits): return f"0x{int(bits, 2):0{(len(bits) + 3) // 4}X}"
def describe(bits): """Best guess at the protocol, from the bit count and pattern.""" n, v = len(bits), int(bits, 2) trits = "".join({"00": "0", "11": "1", "01": "F"}.get(bits[i : i + 2], "?") for i in range(0, n, 2)) out = [] if n == 24: out.append(f"EV1527 id 0x{v >> 4:05X} button {v & 0xF:X}") if n % 2 == 0 and n >= 16 and "?" not in trits: out.append(f"PT2262 {trits}") if n == 12: out.append("12 bits, like CAME or Nice FLO") if 64 <= n <= 66: out.append("KeeLoq length") return " · ".join(out) or f"{n}-bit fixed code"
def code_line(bits, c): return (f"{c['n']:>4}× {hexcode(bits):<19} {len(bits):>2} bits {describe(bits):<34} " f"TE {c['te']:>4} µs {c['mhz']:.3f} MHz {fmt(c['rssi']):>8} last {hms(c['last'])}")
def keeloq_line(fr): return (f"{hms(fr['t'])} KeeLoq serial 0x{fr['serial']:07X} button 0x{fr['button']:X} " f"hop 0x{fr['hop']:08X} TE {fr['te']} µs {fr['mhz']:.3f} MHz")
class ListenView: """Listen mode: cuts the sampled data where no frame can be split, finds frames, counts codes."""
def __init__(self, mhz, log=None): self.mhz, self.log = mhz, log self.pending = [] self.chunks = 0 self.clear()
def clear(self): self.codes = {} # bits -> n, te, first, last, mhz, rssi self.keeloq, self.keeloq_seen = [], set() self.last = None # (bits, te, timings, when) of the latest frame self.history = deque(maxlen=2000) # per chunk: [RSSI, frames found] self.rssi, self.top = None, (None, 0.0)
def retune(self, mhz): self.flush() self.mhz = mhz
def add(self, data, rssi, when=None): """One chunk of samples and the RSSI read after it. Returns new sightings: [(kind, what)].""" when = time.time() if when is None else when self.chunks += 1 self.rssi = rssi if rssi is not None and (self.top[0] is None or rssi > self.top[0]): self.top = (rssi, when) self.history.append([rssi, 0]) new = cap.to_durations(data) if self.pending and new and (self.pending[-1] > 0) == (new[0] > 0): self.pending[-1] += new.pop(0) # the run carries on across the chunk boundary self.pending += new cut = cut_point(self.pending) if cut is None and sum(map(abs, self.pending)) > MAX_PENDING_US: cut = next((i for i in range(len(self.pending) - 1, 0, -1) if self.pending[i] <= -PAIR_MAX), len(self.pending)) if not cut: return [] seg, self.pending = self.pending[:cut], self.pending[cut:] return self.process(seg, when)
def flush(self): seg, self.pending = self.pending, [] return self.process(seg) if seg else []
def process(self, seg, when=None): when = time.time() if when is None else when if self.log: self.log.write("[I][remote.raw]: Received Raw: " + ", ".join(map(str, seg)) + "\n") recent = [r for r, _ in list(self.history)[-3:] if r is not None] events = [] for bits, te, timings in find_frames(seg): c = self.codes.setdefault(bits, {"n": 0, "first": when}) c.update(n=c["n"] + 1, te=te, last=when, mhz=self.mhz, rssi=max(recent, default=None)) self.last = (bits, te, timings, when) if self.history: self.history[-1][1] += 1 events.append(("code", bits)) for burst in cap.bursts(seg): for fr in kc.decode_raw(burst): key = (fr["serial"], fr["button"], fr["hop"]) if key not in self.keeloq_seen: self.keeloq_seen.add(key) self.keeloq.append({**fr, "t": when, "mhz": self.mhz}) events.append(("keeloq", self.keeloq[-1])) return events
# ---- screen -------------------------------------------------------------------------------------
STYLE = {"bar": 0, "hold": curses.A_BOLD, "mark": curses.A_BOLD, "dim": 0, "hi": curses.A_BOLD, "warn": curses.A_BOLD, "ramp": [0] * 5}
def init_style(): try: curses.curs_set(0) except curses.error: pass if not curses.has_colors(): return curses.start_color() try: curses.use_default_colors() bg = -1 except curses.error: bg = curses.COLOR_BLACK for n, c in enumerate((curses.COLOR_GREEN, curses.COLOR_YELLOW, curses.COLOR_RED, curses.COLOR_CYAN, curses.COLOR_BLUE), 1): curses.init_pair(n, c, bg) p = curses.color_pair STYLE.update(bar=p(1), hold=p(2) | curses.A_BOLD, mark=p(3) | curses.A_BOLD, dim=curses.A_DIM, hi=p(1) | curses.A_BOLD, warn=p(2) | curses.A_BOLD, ramp=[p(5), p(4), p(1), p(2), p(3) | curses.A_BOLD])
class Grid: """A pretend curses window that keeps the text: for the self-test and --preview."""
def __init__(self, h, w): self.h, self.w = h, w self.erase()
def getmaxyx(self): return self.h, self.w
def erase(self): self.rows = [[" "] * self.w for _ in range(self.h)]
def addstr(self, y, x, text, attr=0): for i, ch in enumerate(text): if x + i < self.w: self.rows[y][x + i] = ch
def text(self): return "\n".join("".join(r).rstrip() for r in self.rows)
def put(win, y, x, text, attr=0): h, w = win.getmaxyx() if 0 <= y < h and 0 <= x < w: try: win.addstr(y, x, text[: w - x], attr) except curses.error: # the bottom-right cell pass
def ticks(plan, min_gap=10): """Round-number axis ticks at least min_gap columns apart: [(column, label)].""" for tick in (0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 20, 50, 100): if tick / plan.step >= min_gap: break dec = 0 if tick >= 1 else 1 if tick >= 0.1 else 2 first = math.ceil(plan.lo / tick - 1e-9) out = [] for k in range(first, math.floor(plan.hi / tick + 1e-9) + 1): out.append((plan.column(k * tick), f"{k * tick:.{dec}f}")) return out
def draw_sweep(win, sv, marker, note=""): h, w = win.getmaxyx() plan, S = sv.plan, STYLE put(win, 0, 0, f"CC1101 sweep · {sv.label} {plan.lo:.3f}-{plan.hi:.3f} MHz · step {plan.step * 1e3:.0f} kHz · " f"filter {plan.bw:.0f} kHz · {sv.took:.2f} s per sweep · #{sv.count}", S["hi"]) bars = max(5, (h - 6) * 45 // 100) water = h - 6 - bars for r in range(0, bars, 3): # dBm labels, top down put(win, 1 + r, 0, f"{DBM_LO + (DBM_HI - DBM_LO) * (bars - 1 - r) / bars:>4.0f}┤", S["dim"]) mc = plan.column(marker) for i in range(min(plan.cols, w - LEFT)): v, hv = sv.levels[i], sv.hold[i] units = frac(v) * bars * 8 if v is not None else 0 hold_r = min(int(frac(hv) * bars), bars - 1) if hv is not None else -1 for r in range(bars): # from the bottom fill = min(max(int(units - r * 8), 0), 8) if fill: put(win, bars - r, LEFT + i, BLOCKS[fill], S["bar"]) elif r == hold_r: put(win, bars - r, LEFT + i, "·", S["hold"]) elif i == mc: put(win, bars - r, LEFT + i, "│", S["mark"]) tk = ticks(plan) axis = ["─"] * plan.cols for c, _ in tk: axis[c] = "┴" put(win, bars + 1, LEFT, "".join(axis), S["dim"]) put(win, bars + 1, LEFT + mc, "▲", S["mark"]) put(win, bars + 2, 0, "MHz", S["dim"]) end = -1 for c, label in tk: x = max(c - len(label) // 2, 0) if x > end and x + len(label) <= plan.cols: put(win, bars + 2, LEFT + x, label) end = x + len(label) v, hv = sv.levels[mc], sv.hold[mc] text = f"▲ {marker:.3f} MHz · now {fmt(v)} · hold {fmt(hv)}" if hv is not None: text += f" at {hms(sv.hold_t[mc])}" text += f" · floor {sv.floor():.0f} dBm · Enter listens here" put(win, bars + 3, 0, text, S["mark"]) if note: put(win, bars + 3, len(text) + 3, note, S["warn"]) floor = sv.floor() now = time.time() for k in range(min(water, len(sv.water))): # newest at the top, 4 dB per step over the floor y = bars + 4 + k t, levels = sv.water[k] if k % 5 == 0: put(win, y, 0, f"{min(now - t, 999):>3.0f}s", S["dim"]) for i, lv in enumerate(levels[: w - LEFT]): idx = min(max(int((lv - floor - 4) / 4), 0), 9) if lv is not None else 0 if idx: put(win, y, LEFT + i, RAMP[idx], S["ramp"][idx // 2]) pk = sv.peaks(sv.hold) put(win, h - 2, 0, "strongest: " + (" · ".join(f"{f:.3f} MHz {db:.0f} dBm {hms(sv.hold_t[c])}" for f, db, c in pk) if pk else "nothing 10 dB over the floor yet")) put(win, h - 1, 0, "Tab/Enter listen at ▲ · ←→ [ ] move ▲ · p ▲ to strongest · z x zoom · b band · c clear · q quit", S["dim"])
def pulses(timings, bits, width): """The frame's highs and lows squeezed into width columns, and its bits under each pair.""" total = sum(abs(t) for t in timings) upc = max(total / width, 1) # µs per column high = [0.0] * width t = 0 for d in timings: a, b = t, t + abs(d) if d > 0: for c in range(int(a // upc), min(int(b // upc) + 1, width)): high[c] += max(0.0, min(b, (c + 1) * upc) - max(a, c * upc)) t = b wave = "".join("█" if x >= 0.5 * upc else "▁" for x in high) row, t = [" "] * width, 0 for n, bit in enumerate(bits): c = int(t / upc) if c < width: row[c] = bit t += abs(timings[2 * n]) + abs(timings[2 * n + 1]) return wave, "".join(row), upc
def draw_listen(win, lv, note=""): h, w = win.getmaxyx() S = STYLE log = lv.log.name if lv.log else "none" put(win, 0, 0, f"CC1101 listen · {lv.mhz:.3f} MHz · OOK, 650 kHz filter · data sampled at {cap.RATE // 1000} kHz · " f"log {log}", S["hi"]) n = min(40, w - 40) fill = round(frac(lv.rssi) * n) if lv.rssi is not None else 0 put(win, 1, 0, "RSSI") put(win, 1, LEFT, "█" * fill, S["bar"]) put(win, 1, LEFT + fill, "░" * (n - fill), S["dim"]) text = f" {fmt(lv.rssi)} · loudest {fmt(lv.top[0])}" + (f" at {hms(lv.top[1])}" if lv.top[0] is not None else "") put(win, 1, LEFT + n, text) if note: put(win, 1, LEFT + n + len(text) + 3, note, S["warn"]) put(win, 3, 0, "Signal strength, one column per 0.2 s, newest on the right; • = a frame decoded", S["dim"]) hist = list(lv.history)[-(w - LEFT) :] for r in range(3): # a 3-row sparkline, 24 steps from DBM_LO to DBM_HI put(win, 6 - r, 0, f"{DBM_LO + (DBM_HI - DBM_LO) * r / 3:>4.0f}┤", S["dim"]) for i, (rs, _) in enumerate(hist): f = min(max(int(frac(rs) * 24) - r * 8, 0), 8) if rs is not None else 0 if f: put(win, 6 - r, LEFT + i, BLOCKS[f], S["bar"]) for i, (_, nf) in enumerate(hist): if nf: put(win, 7, LEFT + i, "•", S["hi"]) if lv.last: bits, te, timings, when = lv.last c = lv.codes.get(bits, {"n": 1}) put(win, 9, 0, f"Last frame {hms(when)}: {hexcode(bits)} · {len(bits)} bits · {describe(bits)} · TE {te} µs · " f"heard {c['n']}×", S["hi"]) wave, row, upc = pulses(timings, bits, w - LEFT - 1) put(win, 10, 0, "data", S["dim"]) put(win, 10, LEFT, wave, S["bar"]) put(win, 11, 0, "bits", S["dim"]) put(win, 11, LEFT, row) put(win, 12, LEFT, f"one column = {upc:.0f} µs", S["dim"]) else: put(win, 9, 0, "No frame yet: hold a remote button for a second or two, 30 cm or more from the antenna", S["dim"]) y = 14 codes = sorted(((b, c) for b, c in lv.codes.items() if c["n"] >= 2), key=lambda bc: -bc[1]["last"]) once = sum(1 for c in lv.codes.values() if c["n"] == 1) put(win, y, 0, f"Codes heard more than once ({len(codes)}; {once} more heard once, likely noise or a cut-off frame):", S["hi"]) room = h - y - 2 # rows between this heading and the keys line; codes come first kq = min(len(lv.keeloq), 3, max(room - 1 - len(codes), 1 if room >= 4 else 0)) if lv.keeloq else 0 for k, (bits, c) in enumerate(codes[: room - (kq + 1 if kq else 0)]): put(win, y + 1 + k, 0, code_line(bits, c), S["hi"] if k == 0 else 0) keeloq = lv.keeloq[-kq:] if kq else [] if keeloq: y = h - 2 - len(keeloq) put(win, y, 0, f"KeeLoq presses ({len(lv.keeloq)}; rolling code, every press differs):", S["hi"]) for k, fr in enumerate(keeloq): put(win, y + 1 + k, 0, keeloq_line(fr)) put(win, h - 1, 0, "Tab sweep · ←→ ±10 kHz · [ ] ±100 kHz · c clear · q quit", S["dim"])
# ---- the app ------------------------------------------------------------------------------------
class Failure(Exception): pass
def pick_marker(lo, hi): return next((f for f in (433.92, 315.0, 868.35, 915.0) if lo <= f <= hi), (lo + hi) / 2)
def parse_band(text): for i, (key, label, lo, hi) in enumerate(BANDS): if text == key: return i, label, lo, hi m = re.fullmatch(r"(\d+(?:\.\d+)?)-(\d+(?:\.\d+)?)", text) if not m or not 300 <= float(m[1]) < float(m[2]) <= 928: raise SystemExit(f"🔴 --band {text}: use 433, 315, 300-348, 387-464, 779-928, all, or START-STOP " "within 300-928 MHz\n\n🔴 FAILED") return -1, "custom", float(m[1]), float(m[2])
class App: def __init__(self, worker, band, listen=None, log=None): self.worker = worker self.band_i, self.label, self.lo, self.hi = band self.marker = pick_marker(self.lo, self.hi) self.cols = 100 self.sweep = None self.listen = ListenView(listen or self.marker, log) self.mode = "listen" if listen else "sweep" self.note, self.note_t = "", 0.0
def say(self, text): self.note, self.note_t = text, time.time()
def start(self): if self.mode == "listen": self.go_listen(self.listen.mhz) else: self.go_sweep()
def go_sweep(self): sv = self.sweep if sv is None or sv.plan.cols != self.cols or (sv.plan.lo, sv.plan.hi) != (self.lo, self.hi): self.sweep = SweepView(Plan(self.lo, self.hi, self.cols), self.label) self.marker = min(max(self.marker, self.lo), self.hi) self.mode = "sweep" self.worker.want = ("sweep", self.sweep.plan)
def go_listen(self, mhz): mhz = round(mhz, 3) if not supported(mhz): self.say(f"{mhz:.3f} MHz is outside the CC1101's bands") return self.listen.retune(mhz) self.mode = "listen" self.worker.want = ("listen", mhz)
def resize(self, width): self.cols = max(width - LEFT - 1, 20) if self.mode == "sweep" and self.sweep: self.go_sweep()
def zoom(self, factor): span = min(max((self.hi - self.lo) * factor, 1.0), 628.0) self.lo = round(min(max(self.marker - span / 2, 300.0), 928.0 - span), 3) self.hi, self.label = round(self.lo + span, 3), "zoom" self.go_sweep()
def key(self, ch): """Handle one key; False means quit.""" if ch in (ord("q"), ord("Q")): return False if ch == ord("\t"): self.go_listen(self.marker) if self.mode == "sweep" else self.go_sweep() elif ch == ord("c"): (self.sweep if self.mode == "sweep" else self.listen).clear() elif self.mode == "sweep": plan = self.sweep.plan moves = {curses.KEY_LEFT: -1, curses.KEY_RIGHT: 1, ord("["): -10, ord("]"): 10} if ch in (10, 13, curses.KEY_ENTER): self.go_listen(self.marker) elif ch in moves: self.marker = plan.freq(min(max(plan.column(self.marker) + moves[ch], 0), plan.cols - 1)) elif ch == ord("p"): pk = self.sweep.peaks(self.sweep.hold, n=1) if pk: self.marker = pk[0][0] else: self.say("no peak yet") elif ch == ord("z"): self.zoom(1 / 4) elif ch == ord("x"): self.zoom(4) elif ch == ord("b"): self.band_i = (self.band_i + 1) % len(BANDS) _, self.label, self.lo, self.hi = BANDS[self.band_i] self.marker = pick_marker(self.lo, self.hi) self.go_sweep() else: moves = {curses.KEY_LEFT: -0.01, curses.KEY_RIGHT: 0.01, ord("["): -0.1, ord("]"): 0.1} if ch in moves: self.go_listen(self.listen.mhz + moves[ch]) return True
def pump(self): """Take the worker's results: (anything new, listen events).""" changed, events = False, [] while True: try: want, payload, extra = self.worker.results.get_nowait() except queue.Empty: return changed, events if want is None: raise Failure(payload) if want is not self.worker.want: continue if want[0] == "sweep": self.sweep.add(payload, extra) else: events += self.listen.add(payload, extra) changed = True
def draw(self, win): h, w = win.getmaxyx() note = self.note if time.time() - self.note_t < 4 else "" if h < 20 or w < 60: put(win, 0, 0, "Make the window at least 60 × 20") elif self.mode == "sweep": draw_sweep(win, self.sweep, self.marker, note) else: draw_listen(win, self.listen, note)
def run_tui(stdscr, app, seconds): init_style() stdscr.timeout(100) end = time.monotonic() + seconds if seconds else math.inf app.resize(stdscr.getmaxyx()[1]) app.start() while time.monotonic() < end: ch = stdscr.getch() if ch == curses.KEY_RESIZE: app.resize(stdscr.getmaxyx()[1]) elif ch != -1 and not app.key(ch): break app.pump() stdscr.erase() app.draw(stdscr) stdscr.refresh()
def run_text(app, seconds): end = time.monotonic() + seconds if seconds else math.inf app.start() if app.mode == "sweep": p = app.sweep.plan print(f"🌱 sweeping {p.lo:.3f}-{p.hi:.3f} MHz in {p.cols} steps of {p.step * 1e3:.0f} kHz, " f"filter {p.bw:.0f} kHz (Ctrl+C stops)") else: print(f"🌱 listening at {app.listen.mhz:.3f} MHz: hold a remote button for a second or two (Ctrl+C stops)") last_line = last_beat = time.time() while time.monotonic() < end: time.sleep(0.05) changed, events = app.pump() now = time.time() if app.mode == "sweep" and changed and now - last_line >= 1: sv = app.sweep pk = sv.peaks(sv.levels, n=3) print(f"{hms(now)} floor {sv.floor():.0f} dBm · " + (" · ".join(f"🟢 {f:.3f} MHz {db:.0f} dBm" for f, db, _ in pk) if pk else "nothing 10 dB over the floor")) last_line = now for kind, what in events: if kind == "keeloq": print(f"🟢 {keeloq_line(what)}") continue c = app.listen.codes[what] if c["n"] >= 2 and now - c.get("printed", 0) >= 1: print(f"{hms(now)} 🟢 {hexcode(what)} · {len(what)} bits · {describe(what)} · TE {c['te']} µs · " f"RSSI {fmt(c['rssi'])} · heard {c['n']}×") c["printed"] = now if app.mode == "listen" and now - last_beat >= 5: print(f"{hms(now)} 🪵 listening at {app.listen.mhz:.3f} MHz · RSSI {fmt(app.listen.rssi)} · " f"loudest {fmt(app.listen.top[0])}") last_beat = now
def summary(app=None, lv=None, everything=False): sv = app.sweep if app else None lv = lv or app.listen if sv and sv.count: pk = sv.peaks(sv.hold) print(f"🪵 {sv.count} sweeps of {sv.plan.lo:.3f}-{sv.plan.hi:.3f} MHz, floor {sv.floor():.0f} dBm; strongest:") for f, db, c in pk: print(f" {f:.3f} MHz {db:.0f} dBm {hms(sv.hold_t[c])}") if not pk: print(" nothing 10 dB over the floor") if lv.chunks or everything: codes = sorted(lv.codes.items(), key=lambda bc: -bc[1]["n"]) shown = [bc for bc in codes if everything or bc[1]["n"] >= 2] print(f"🪵 codes heard{'' if everything else ' more than once'}: {len(shown)}") for bits, c in shown: print(" " + code_line(bits, c)) if len(codes) > len(shown): print(f" (+{len(codes) - len(shown)} heard only once)") for fr in lv.keeloq: print(" " + keeloq_line(fr)) if lv.log: print(f"🪵 timings saved to {lv.log.name}")
def read_raw(path): """Raw timing blocks from a log: this script's, keeloq-capture-ft232h.py's or ESPHome's.""" blocks, cur = [], None for line in Path(path).read_text(errors="replace").splitlines(): if "remote.raw" not in line: continue text = line.rsplit("]: ", 1)[-1] if "Received Raw:" in text: if cur: blocks.append(cur) cur, text = [], text.split("Received Raw:", 1)[1] if cur is not None: cur += [int(n) for n in re.findall(r"-?\d+", text)] if cur: blocks.append(cur) return blocks
# ---- self-test ----------------------------------------------------------------------------------
# A real frame from my AliExpress copy remote (keeloq-capture-ft232h.py, 2026-11-02), with the noise# before it: the 45 ms high is the gap before the frame filled with noise, swallowing the first bit's highREAL = [1040, -120, 640, -400, 120, -140, 45720, -1000, 300, -1080, 280, -1080, 300, -1040, 1000, -340, 340, -1040, 960, -380, 1000, -380, 980, -380, 300, -1040, 320, -1040, 300, -1040, 340, -1040, 320, -1020, 320, -1040, 340, -1020, 1000, -360, 320, -1020, 1020, -360, 340, -1020, 1000, -360, 320, -1020, 360, -1020, 320, -1020, 340, -10500]
def noise(rng, seconds): """Idle CC1101 output as my rig showed it (2026-11-02): mostly high, with short dips.""" out, total = [], 0.0 while total < seconds * 1e6: h = rng.uniform(3000, 60000) if rng.random() < 0.12 else rng.uniform(100, 1500) l = rng.uniform(100, 400) if rng.random() < 0.9 else rng.uniform(400, 1500) out += [round(h), -round(l)] total += h + l return out
def ev1527(rng, code, te, repeats): """A held EV1527 button: sync (1 TE high, 31 TE low), then 24 bits; 1 = 3 TE high + 1 TE low.""" out = [] for _ in range(repeats): out += [te, -31 * te] if rng.random() < 0.5: # noise fills the gap and swallows the first bit's high, as on my rig dip = rng.randint(100, 600) out[-1:] = [-dip, 31 * te - dip] for n in range(23, -1, -1): out += [3 * te, -te] if code >> n & 1 else [te, -3 * te] return out
def keeloq_press(serial, button, repeats): mk = (kc.load_keys(kc.KEYFILE) if kc.KEYFILE.exists() else kc.TEST_KEYS)[0][0] dev = kc.normal(button << 28 | serial, mk) hop = kc.encrypt(button << 28 | (serial & 0x3FF) << 16 | 0x0101, dev) v = (button << 28 | serial) << 32 | hop frame = [410, -390] * 11 + [400, -4000] for n in range(66): frame += [400, -820] if kc.bit(v, n) else [810, -400] frame[-1] = -15000 return frame * repeats
def to_chunks(rng, timings): """Raw timings -> 50 kHz samples with jitter and single-sample glitches -> LISTEN_CHUNK reads.""" s = list("".join(("1" if d > 0 else "0") * round(abs(d) * rng.uniform(0.92, 1.08) * cap.RATE / 1e6) for d in timings)) for i in rng.sample(range(len(s)), len(s) // 2000): s[i] = "1" if s[i] == "0" else "0" s = "".join(s) + "0" * (-len(s) % 8) data = int(s, 2).to_bytes(len(s) // 8, "big") return [data[i : i + LISTEN_CHUNK] for i in range(0, len(data), LISTEN_CHUNK)]
def demo_views(rng, lv, cols): plan = Plan(432.0, 436.0, cols) sv = SweepView(plan, "433 MHz remotes") t0 = time.time() - 30 for k in range(40): levels = [rng.gauss(-92, 1.2) for _ in range(plan.cols)] if k % 3 == 0: # the remote, on and off c = plan.column(433.92) for j in range(max(c - 2, 0), min(c + 3, plan.cols)): levels[j] = -45 - 6 * abs(j - c) + rng.gauss(0, 1) sv.add(levels, 0.18, t0 + k * 0.2) return sv
def self_test(preview=False): rng = random.Random(7) assert freq_word(433.92) == [0x10, 0xB0, 0x71], freq_word(433.92) assert filter_for(650) == (650.0, cap.REGS[0x10]), filter_for(650) assert round(filter_for(30)[0], 1) == 58.0 and filter_for(5000)[0] == 812.5 wide = Plan(779.0, 928.0, 110) assert all(len(p) == 2 for p in wide.points[1:-1]), "wide steps need 2 points per column" every = Plan(300.0, 928.0, 120) assert every.points[every.column(400)] and not every.points[every.column(600)] print(f"🟢 tuning maths: 433.92 MHz = 10 B0 71, {len(FILTERS)} filters 58-812 kHz, wide bands gap-free, " "columns between the CC1101's bands stay empty")
[(bits, te, _)] = find_frames(REAL) assert (hexcode(bits), describe(bits)) == ("0x0B80A8", "EV1527 id 0x0B80A button 8"), (bits, describe(bits)) print(f"🟢 my AliExpress remote's frame decodes: {hexcode(bits)}, {describe(bits)}, TE {te} µs") assert describe("00" * 4 + "11" * 4 + "01" * 4).startswith("EV1527") and "PT2262 000011" in describe( "00" * 4 + "11" * 2 + "01" * 6)
code = 0x5A3C91 stream = (noise(rng, 2) + ev1527(rng, code, 350, 10) + noise(rng, 1) + ev1527(rng, code, 350, 6) + noise(rng, 1) + keeloq_press(0x0ABCDEF, 0x2, 2) + noise(rng, 1)) lv, t = ListenView(433.92), time.monotonic() chunks = to_chunks(rng, stream) for k, data in enumerate(chunks): lv.add(data, -90.0 + (40 if 10 <= k <= 20 else 0), 1000.0 + k * 0.2) lv.flush() took = time.monotonic() - t want = format(code, "024b") repeated = {b: c["n"] for b, c in lv.codes.items() if c["n"] >= 2} assert repeated.get(want, 0) >= 12, f"EV1527 found {repeated.get(want, 0)} of 16 times: {repeated}" assert describe(want) == "EV1527 id 0x5A3C9 button 1" stray = {hexcode(b): n for b, n in repeated.items() if b != want and len(b) < 64} assert not stray, f"codes noise made up: {stray}" assert [f["serial"] for f in lv.keeloq] == [0x0ABCDEF], lv.keeloq assert took < len(chunks) * 0.2 / 4, f"{took:.1f} s to decode {len(chunks) * 0.2:.0f} s of data" print(f"🟢 {len(chunks) * 0.2:.0f} s of noisy samples: EV1527 0x5A3C91 found {lv.codes[want]['n']} of 16 times, " f"the KeeLoq press decoded, no made-up codes, {took:.2f} s to decode")
quiet = ListenView(433.92) for data in to_chunks(rng, noise(rng, 20)): quiet.add(data, -91.0) quiet.flush() fake = {hexcode(b): c["n"] for b, c in quiet.codes.items() if c["n"] >= 2} assert not fake, f"20 s of noise made up codes: {fake}" print(f"🟢 20 s of noise alone: no code heard twice ({len(quiet.codes)} single noise runs ignored)")
for h, w in ((20, 60), (24, 80), (45, 130), (60, 200)): sv = demo_views(rng, lv, w - LEFT - 1) for draw, view in ((draw_sweep, sv), (draw_listen, lv)): g = Grid(h, w) draw(g, view, 433.92) if draw is draw_sweep else draw(g, view) text = g.text() assert ("433.9" in text) if draw is draw_sweep else ("0x5A3C91" in text), f"{draw.__name__} {w}x{h}:\n{text}" if preview and (h, w) == (45, 130): print(f"\n---- {draw.__name__} {w} x {h} ----\n{text}\n") print("🟢 both screens draw at 60 x 20 up to 200 x 60") print("\n🟢 SUCCESS")
def main(): locale.setlocale(locale.LC_ALL, "") ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--band", default="433") ap.add_argument("--listen", type=float, metavar="MHZ", help="start in listen mode at MHZ") ap.add_argument("--text", action="store_true", help="plain lines instead of the screen") ap.add_argument("--seconds", type=float, help="stop after this long") ap.add_argument("--url", default="ftdi://ftdi:232h/1") ap.add_argument("--replay", type=Path, metavar="LOG", help="decode a saved log instead of the radio") ap.add_argument("--self-test", action="store_true") ap.add_argument("--preview", action="store_true", help="with --self-test: print both screens") args = ap.parse_args() if args.self_test: self_test(args.preview) return 0 if args.replay: lv = ListenView(433.92) for block in read_raw(args.replay): lv.process(block) summary(lv=lv, everything=True) print("\n🟢 SUCCESS") return 0 band = parse_band(args.band) if args.listen and not supported(args.listen): raise SystemExit(f"🔴 {args.listen} MHz is outside the CC1101's bands {CC_BANDS}\n\n🔴 FAILED") radio = Radio(args.url) worker = Worker(radio) worker.start() log = LOG.open("w", buffering=1) app = App(worker, band, args.listen, log) failure = None try: if args.text: run_text(app, args.seconds) else: curses.wrapper(run_tui, app, args.seconds) except KeyboardInterrupt: pass except Failure as e: failure = str(e) finally: worker.stop.set() worker.join(2) app.listen.flush() log.close() try: radio.close() except Exception: pass print() summary(app) if failure: print(f"\n🔴 FAILED: {failure}") return 1 print("\n🟢 SUCCESS") return 0
if __name__ == "__main__": sys.exit(main())Text, photos and drawings: CC BY 4.0. Code: MIT. Share and adapt them, crediting HomeLab Series (blog.jacq.cc). Licence and privacy