#!/usr/bin/env python3
"""Interactive waterfall viewer for rtl_power CSV output.
Usage:
python3 waterfall.py [--dir DIR] [--host HOST] [--port PORT]
Then open http://127.0.0.1:8000 in a browser.
"""
import argparse
import glob
import json
import os
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
def parse_rtl_power(path):
"""Parse an rtl_power CSV into a waterfall matrix.
Returns dict with frequency axis (MHz), time labels, and 2D power grid.
"""
sweeps = [] # list of (label, {freq_hz: power_db})
freqs = set()
start_low = None
current = None
current_label = None
with open(path, "r") as fh:
for line in fh:
parts = [p.strip() for p in line.split(",")]
if len(parts) < 7:
continue
date, time_s = parts[0], parts[1]
try:
f_low = float(parts[2])
f_high = float(parts[3])
f_step = float(parts[4])
# parts[5] = samples
values = [float(v) for v in parts[6:] if v != ""]
except ValueError:
continue
if start_low is None:
start_low = f_low
# A new sweep starts when the segment wraps back to the range start.
if f_low == start_low:
if current:
sweeps.append((current_label, current))
current = {}
current_label = f"{date} {time_s}"
n = len(values)
span = f_high - f_low
for i, val in enumerate(values):
# Bin center frequency.
freq = f_low + (i + 0.5) * (span / n) if n else f_low
freq = round(freq)
current[freq] = val
freqs.add(freq)
if current:
sweeps.append((current_label, current))
freq_axis = sorted(freqs)
freq_index = {f: i for i, f in enumerate(freq_axis)}
grid = []
times = []
for label, mapping in sweeps:
row = [None] * len(freq_axis)
for f, val in mapping.items():
row[freq_index[f]] = val
grid.append(row)
times.append(label)
return {
"file": os.path.basename(path),
"freqs_mhz": [round(f / 1e6, 6) for f in freq_axis],
"times": times,
"power": grid,
}
def list_csvs(directory):
files = sorted(glob.glob(os.path.join(directory, "*.csv")))
return [os.path.basename(f) for f in files]
PAGE = """
rtl_power waterfall
"""
class Handler(BaseHTTPRequestHandler):
directory = "."
def log_message(self, *args):
pass
def _send(self, code, body, ctype):
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/":
self._send(200, PAGE.encode(), "text/html; charset=utf-8")
elif parsed.path == "/files":
body = json.dumps(list_csvs(self.directory)).encode()
self._send(200, body, "application/json")
elif parsed.path == "/data":
qs = parse_qs(parsed.query)
name = os.path.basename(qs.get("file", [""])[0])
path = os.path.join(self.directory, name)
if not name or not os.path.isfile(path):
self._send(404, b'{"error":"not found"}', "application/json")
return
body = json.dumps(parse_rtl_power(path)).encode()
self._send(200, body, "application/json")
else:
self._send(404, b"not found", "text/plain")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--dir", default=".", help="directory with rtl_power CSVs")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, default=8000)
args = ap.parse_args()
Handler.directory = os.path.abspath(args.dir)
server = ThreadingHTTPServer((args.host, args.port), Handler)
print(f"Serving {Handler.directory} at http://{args.host}:{args.port}")
print("Ctrl+C to stop.")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nstopped")
if __name__ == "__main__":
main()