Update scanner

examples/scanner.py

@@ -36,12 +36,18 @@
 
 # use reverse engineering tactics for a better "snapshot"
 radio.address_width = 2
-radio.open_rx_pipe(0, b"\x55\x55")
-radio.open_rx_pipe(1, b"\xaa\xaa")
-radio.open_rx_pipe(2, b"\xa0\xaa")
-radio.open_rx_pipe(3, b"\x0a\xaa")
-radio.open_rx_pipe(4, b"\xa5\xaa")
-radio.open_rx_pipe(5, b"\x5a\xaa")
+# The worst possible addresses. These are designed to confuse the radio into thinking
+# the RF signal's preamble is part of the packet/payload.
+noise_addresses = [
+ b"\x55\x55",
+ b"\xaa\xaa",
+ b"\xa0\xaa",
+ b"\x0a\xaa",
+ b"\xa5\xaa",
+ b"\x5a\xaa",
+]
+for pipe, address in enumerate(noise_addresses):
+ radio.open_rx_pipe(pipe, address)
 
 
 def scan(timeout: int = 30):
@@ -60,6 +66,7 @@ def scan(timeout: int = 30):
  print("\n" + "~" * 126)
 
  signals = [0] * 126 # store the signal count for each channel
+ sweeps = 0 # keep track of the number of sweeps made through all channels
  curr_channel = 0
  start_timer = time.monotonic() # start the timer
  while time.monotonic() - start_timer < timeout:
@@ -75,15 +82,21 @@ def scan(timeout: int = 30):
  # clear the RX FIFO if a signal was detected/captured
  if found_signal:
  radio.flush_rx() # flush the RX FIFO because it asserts the RPD flag
+ endl = False
+ if curr_channel >= 125:
+ sweeps += 1
+ if sweeps % 100 == 0:
+ endl = True
  curr_channel = curr_channel + 1 if curr_channel < 125 else 0
 
  # output the signal counts per channel
  sig_cnt = signals[curr_channel]
  print(
  ("%X" % min(15, sig_cnt)) if sig_cnt else "-",
  sep="",
- end="" if curr_channel < 125 else "\r",
+ end="" if curr_channel < 125 else ("\n" if endl else "\r"),
  )
+
  # finish printing results and end with a new line
  while curr_channel < len(signals) - 1:
  curr_channel += 1

examples/scanner_curses.py

@@ -0,0 +1,227 @@
+"""A scanner example written in python using the std lib's ncurses wrapper.
+
+This is a good diagnostic tool to check whether you're picking a
+good channel for your application.
+
+See documentation at https://nRF24.github.io/pyRF24
+"""
+
+# pylint: disable=no-member
+import curses
+import time
+from typing import List, Tuple, Any
+
+from pyrf24 import RF24, RF24_1MBPS, RF24_2MBPS, RF24_250KBPS
+
+CSN_PIN = 0 # connected to GPIO8
+CE_PIN = 22 # connected to GPIO22
+radio = RF24(CE_PIN, CSN_PIN)
+
+
+OFFERED_DATA_RATES = ["1 Mbps", "2 Mbps", "250 kbps"]
+AVAILABLE_RATES = [RF24_1MBPS, RF24_2MBPS, RF24_250KBPS]
+TOTAL_CHANNELS = 126
+CACHE_MAX = 5 # the depth of history to calculate peaks
+
+# To detect noise, we'll use the worst addresses possible (a reverse engineering
+# tactic). These addresses are designed to confuse the radio into thinking that the
+# RF signal's preamble is part of the packet/payload.
+noise_address = [
+ b"\x55\x55",
+ b"\xaa\xaa",
+ b"\x0a\xaa",
+ b"\xa0\xaa",
+ b"\x00\xaa",
+ b"\xab\xaa",
+]
+
+
+class ChannelHistory:
+ def __init__(self) -> None:
+ #: FIFO for tracking peak decays
+ self._history: List[bool] = [False] * CACHE_MAX
+ #: for the total signal counts
+ self.total: int = 0
+
+ def push(self, value: bool) -> int:
+ """Push a scan result's value into history while returning the sum of cached
+ signals found. This function also increments the total signal count accordingly.
+ """
+ self._history = self._history[1:] + [value]
+ self.total += value
+ return self._history.count(True)
+
+
+#: An array of histories for each channel
+stored = [ChannelHistory() for _ in range(TOTAL_CHANNELS)]
+
+
+class ProgressBar: # pylint: disable=too-few-public-methods
+ """This represents a progress bar using a curses window object."""
+
+ def __init__( # pylint: disable=too-many-arguments,invalid-name
+ self,
+ x: int,
+ y: int,
+ cols: int,
+ std_scr: Any, # type: curses.window,
+ label: str,
+ color: int,
+ ):
+ self.x, self.y, self.width, self.win, self.color = (x, y, cols, std_scr, color)
+ self.win.move(self.y, self.x)
+ self.win.attron(curses.color_pair(self.color))
+ self.win.addstr(label) # always labeled in MHz (4 digits)
+ for _ in range(self.width - 8): # draw the empty bar
+ self.win.addch(curses.ACS_HLINE)
+ self.win.addstr(" - ") # draw the initial signal count
+ self.win.attroff(curses.color_pair(self.color))
+
+ def update(self, completed: int, signal_count: int):
+ """Update the progress bar."""
+ count = " - "
+ if signal_count:
+ count = " %X " % min(0xF, signal_count)
+ filled = (self.width - 8) * completed / CACHE_MAX
+ offset_x = 5
+ self.win.move(self.y, self.x + offset_x)
+ for i in range(offset_x, self.width - 3):
+ bar_filled = i < (filled + offset_x)
+ bar_color = 5 if bar_filled else self.color
+ self.win.attron(curses.color_pair(bar_color))
+ self.win.addch("=" if bar_filled else curses.ACS_HLINE)
+ self.win.attroff(curses.color_pair(bar_color))
+ self.win.attron(curses.color_pair(self.color))
+ self.win.addstr(count)
+ self.win.attroff(curses.color_pair(self.color))
+
+
+def init_display(window) -> List[ProgressBar]:
+ """Creates a table of progress bars (1 for each channel)."""
+ progress_bars: List[ProgressBar] = [None] * TOTAL_CHANNELS
+ bar_w = int(curses.COLS / 6)
+ for i in range(21): # 21 rows
+ for j in range(i, i + (21 * 6), 21): # 6 columns
+ color = 7 if int(j / 21) % 2 else 3
+ progress_bars[j] = ProgressBar(
+ x=bar_w * int(j / 21),
+ y=i + 3,
+ cols=bar_w,
+ std_scr=window,
+ label=f"{2400 + (j)} ",
+ color=color,
+ )
+ return progress_bars
+
+
+def init_radio():
+ """init the radio"""
+ if not radio.begin():
+ raise RuntimeError("Radio hardware not responding!")
+ radio.setAutoAck(False)
+ radio.disableCRC()
+ radio.setAddressWidth(2)
+ for pipe, address in enumerate(noise_address):
+ radio.openReadingPipe(pipe, address)
+ radio.startListening()
+ radio.stopListening()
+ radio.flush_rx()
+
+
+def init_curses():
+ """init the curses interface"""
+ std_scr = curses.initscr()
+ curses.noecho()
+ curses.cbreak()
+ curses.start_color()
+ curses.use_default_colors()
+ curses.init_pair(3, curses.COLOR_YELLOW, -1)
+ curses.init_pair(5, curses.COLOR_MAGENTA, -1)
+ curses.init_pair(7, curses.COLOR_WHITE, -1)
+ return std_scr
+
+
+def deinit_curses(spectrum_passes: int):
+ """de-init the curses interface"""
+ curses.nocbreak()
+ curses.echo()
+ curses.endwin()
+ noisy_channels: int = 0
+ digit_w = len(str(spectrum_passes))
+ for channel, data in enumerate(stored):
+ if data.total:
+ count_padding = " " * (digit_w - len(str(data.total)))
+ percentage = round(data.total / spectrum_passes * 100, 3)
+ print(
+ f" {channel:>3}: {count_padding}{data.total} / {spectrum_passes} ({percentage} %)"
+ )
+ noisy_channels += 1
+ print(
+ f"{noisy_channels} channels detected signals out of {spectrum_passes}",
+ "passes on the entire spectrum.",
+ )
+
+
+def get_user_input() -> Tuple[int, int]:
+ """Get input parameters for the scan from the user."""
+ for i, d_rate in enumerate(OFFERED_DATA_RATES):
+ print(f"{i + 1}. {d_rate}")
+ d_rate = input("Select your data rate [1, 2, 3] (defaults to 1 Mbps) ")
+ duration = input("How long (in seconds) to perform scan? ")
+ while not duration.isdigit():
+ print("Please enter a positive number.")
+ duration = input("How long (in seconds) to perform scan? ")
+ return (
+ max(1, min(3, 1 if not d_rate.isdigit() else int(d_rate))) - 1,
+ abs(int(duration)),
+ )
+
+
+def scan_channel(channel: int) -> bool:
+ """Scan a specified channel and report if a signal was detected."""
+ radio.channel = channel
+ radio.startListening()
+ time.sleep(0.00013)
+ found_signal = radio.testRPD()
+ radio.stopListening()
+ if found_signal or radio.testRPD() or radio.available():
+ radio.flush_rx()
+ return True
+ return False
+
+
+def main():
+ spectrum_passes = 0
+ data_rate, duration = get_user_input()
+ print(f"Scanning for {duration} seconds at {OFFERED_DATA_RATES[data_rate]}")
+ init_radio()
+ radio.setDataRate(AVAILABLE_RATES[data_rate])
+ try:
+ std_scr = init_curses()
+ timer_prompt = "Scanning for {:>3} seconds at " + OFFERED_DATA_RATES[data_rate]
+ std_scr.addstr(0, 0, "Channels are labeled in MHz.")
+ std_scr.addstr(1, 0, "Signal counts are clamped to a single hexadecimal digit.")
+ bars = init_display(std_scr)
+ channel, val = (0, False)
+ end = time.monotonic() + duration
+ while time.monotonic() < end:
+ std_scr.addstr(2, 0, timer_prompt.format(int(end - time.monotonic())))
+ val = scan_channel(channel)
+ cache_sum = stored[channel].push(val)
+ if stored[channel].total:
+ bars[channel].update(cache_sum, stored[channel].total)
+ std_scr.refresh()
+ if channel + 1 == TOTAL_CHANNELS:
+ channel = 0
+ spectrum_passes += 1
+ else:
+ channel += 1
+ finally:
+ radio.powerDown()
+ deinit_curses(spectrum_passes)
+
+
+if __name__ == "__main__":
+ main()
+else:
+ print("Enter 'main()' to run the program.")