sabato 4 luglio 2026

Stima Keystone in FigSpec F60-CL

In accoppiata al precedente post qui si effettua una stima dell'errore Keystone sul sensore del FigSpec F60-CL (viene stimato l'errore spaziale al posto di quello spettrale indicato da Smile Effect)

Il grafico sottostante e' il risultato dell'elaborazione  

In sintesi sembra che da 390 a 750 nm l'errore sia accettabile e stabile intorno a 3 px.Da 750 nm in poi l'errore esplode ma siamo in zona di assorbimento atmosferico per cui il segnale e' molto basso e quindi diciamo che e' ragionevole trovare questo tipo di errore ed non e' dipendente dallo strumento ma dalla fisica dell'atmosfera

Rimuovendo la coda si ha questo grafico 

l'interpretazione potrebbe essere che il sensore (ma non ho documentazione della ditta in merito) sia internamente diviso in due sotto sensori (uno visibile ed uno NIR) con una zona di giunzione intorno a 550 nm  

 

 

#!/usr/bin/env python3
"""
Estimates keystone effect: for a fixed physical spatial edge in the scene,
tracks how its detected column (sample) position shifts across bands
(wavelengths). No keystone -> position stays constant. Keystone present ->
position drifts smoothly with wavelength.

Method:
1. Average many lines together per band to build a stable (bands, samples)
spatial profile (reduces along-track noise, keeps the across-track
structure).
2. Pick a reference band (highest-contrast by default) and find its
strongest spatial edge (max |gradient|), sub-pixel refined.
3. For every other band, search a small window around that same position
for the local edge, sub-pixel refined, so we always track the SAME
physical edge rather than jumping to a different one.
4. Plot/report edge column position vs wavelength.

Usage:
python3 detect_keystone.py figspec.hdr figspec.spe --out keystone_diagnostic.png

Works on raw DN or radiance data (recommended) - reflectance is fine too
since spatial edges are a scene/optics property, not atmosphere-dependent
like the smile-detection absorption feature.
"""

import os
import re
import argparse
import numpy as np



def auto_offset_and_lines(path, bands, samples, itemsize, forced_lines=None,
forced_offset=None):
per_line_bytes = bands * samples * itemsize
file_size = os.path.getsize(path)
if forced_offset is not None and forced_lines is not None:
return forced_offset, forced_lines
expected_lines = forced_lines if forced_lines is not None else file_size // per_line_bytes
expected = per_line_bytes * expected_lines
if file_size == expected:
return 0, expected_lines
remainder = file_size % per_line_bytes
lines = file_size // per_line_bytes
print(f" auto-detected offset={remainder} bytes, lines={lines} "
f"(file size {file_size} doesn't match header dims exactly)")
return remainder, lines


def subpixel_extreme(y, i0, find_max=True):
"""Parabolic sub-pixel refinement around discrete extreme at index i0."""
if i0 <= 0 or i0 >= len(y) - 1:
return float(i0)
y0, y1, y2 = y[i0 - 1], y[i0], y[i0 + 1]
denom = (y0 - 2 * y1 + y2)
if abs(denom) < 1e-9:
return float(i0)
delta = 0.5 * (y0 - y2) / denom
delta = np.clip(delta, -1, 1)
return i0 + delta


def main():
ap = argparse.ArgumentParser()
ap.add_argument("hdr_path")
ap.add_argument("data_path")
ap.add_argument("--out", default="keystone_diagnostic.png")
ap.add_argument("--line-stride", type=int, default=5,
help="Use every Nth line when averaging (speed/memory tradeoff)")
ap.add_argument("--ref-band", type=int, default=None,
help="Reference band index to define the target edge "
"(default: auto-pick band with strongest overall edge contrast)")
ap.add_argument("--search-window-px", type=int, default=8,
help="Half-width (in samples) of the search window around the "
"reference edge position, used for every other band")
ap.add_argument("--wl-min", type=float, default=None,
help="Ignore bands below this wavelength (nm) in the fit/report")
ap.add_argument("--wl-max", type=float, default=None,
help="Ignore bands above this wavelength (nm) in the fit/report")
ap.add_argument("--split-nm", type=float, default=None,
help="If the data looks like a step (not a smooth curve), compare mean "
"positions before/after this wavelength instead of trusting the "
"quadratic fit (e.g. --split-nm 570 to test a detector splice)")
ap.add_argument("--offset", type=int, default=None)
ap.add_argument("--lines", type=int, default=None)
args = ap.parse_args()

meta = parse_hdr(args.hdr_path)
samples, lines, bands = meta["samples"], meta["lines"], meta["bands"]
dtype = meta["dtype"]
interleave = meta["interleave"]
wavelengths = meta["wavelengths"]
pixel_size = meta["pixel_size"]

if interleave != "bil":
raise ValueError("This script currently assumes BIL interleave.")

itemsize = dtype.itemsize
offset, lines = auto_offset_and_lines(
args.data_path, bands, samples, itemsize,
forced_lines=args.lines, forced_offset=args.offset)
print(f"Using offset={offset}, lines={lines}")

cube = np.memmap(args.data_path, dtype=dtype, mode="r", offset=offset,
shape=(lines, bands, samples))

print(f"Averaging every {args.line_stride}th line "
f"({lines // args.line_stride} lines used) to build a per-band spatial profile...")
idx_lines = np.arange(0, lines, args.line_stride)
accum = np.zeros((bands, samples), dtype=np.float64)
for i, li in enumerate(idx_lines):
accum += cube[li, :, :]
if i % 200 == 0:
print(f" {i}/{len(idx_lines)}", end="\r")
mean_spatial = (accum / len(idx_lines)).astype(np.float32) # (bands, samples)
print(f"\nDone averaging. Shape: {mean_spatial.shape}")

# spatial gradient per band (along samples axis)
grad = np.gradient(mean_spatial, axis=1) # (bands, samples)

if args.ref_band is None:
contrast = np.sum(np.abs(grad), axis=1)
ref_band = int(np.argmax(contrast))
print(f"Auto-selected reference band: {ref_band} "
f"({wavelengths[ref_band]:.1f} nm), contrast={contrast[ref_band]:.1f}")
else:
ref_band = args.ref_band

ref_profile_grad = np.abs(grad[ref_band])
ref_i0 = int(np.argmax(ref_profile_grad))
ref_pos = subpixel_extreme(ref_profile_grad, ref_i0, find_max=True)
print(f"Reference edge position (band {ref_band}): column {ref_pos:.2f}")

win = args.search_window_px
edge_positions = np.full(bands, np.nan)
saturated = np.zeros(bands, dtype=bool)

for b in range(bands):
lo = max(0, int(round(ref_pos)) - win)
hi = min(samples, int(round(ref_pos)) + win + 1)
local_grad = np.abs(grad[b, lo:hi])
if len(local_grad) < 3 or local_grad.max() < 1e-6:
continue
i0_local = int(np.argmax(local_grad))
if i0_local == 0 or i0_local == len(local_grad) - 1:
saturated[b] = True # hit the window boundary - unreliable
pos_local = subpixel_extreme(local_grad, i0_local, find_max=True)
edge_positions[b] = lo + pos_local

valid = ~np.isnan(edge_positions) & ~saturated
if args.wl_min is not None:
valid &= wavelengths >= args.wl_min
if args.wl_max is not None:
valid &= wavelengths <= args.wl_max

n_saturated = int(saturated.sum())
if n_saturated > 0:
print(f"\nWARNING: {n_saturated}/{bands} bands hit the search-window edge "
f"(unreliable detections, likely low SNR) - excluded from the fit.")
sat_wls = wavelengths[saturated]
if len(sat_wls) > 0:
print(f" saturated band wavelength range: {sat_wls.min():.1f} - {sat_wls.max():.1f} nm")

wl_valid = wavelengths[valid]
pos_valid = edge_positions[valid]

if len(pos_valid) < 10:
print("Too few valid bands to estimate keystone reliably.")
return

coeffs = np.polyfit(wl_valid, pos_valid, 2)
fit_curve = np.polyval(coeffs, wl_valid)

ptp_measured = pos_valid.max() - pos_valid.min()
ptp_fit = fit_curve.max() - fit_curve.min()

print(f"\nEdge column position across {bands} bands:")
print(f" raw measured range: {pos_valid.min():.2f} - {pos_valid.max():.2f} px "
f"(peak-to-valley = {ptp_measured:.2f} px)")
print(f" smooth quadratic fit range: {fit_curve.min():.2f} - {fit_curve.max():.2f} px "
f"(peak-to-valley = {ptp_fit:.2f} px)")
print(f" quadratic fit coefficients (a*wl^2+b*wl+c): {coeffs}")

if args.split_nm is not None:
seg1 = pos_valid[wl_valid < args.split_nm]
seg2 = pos_valid[wl_valid >= args.split_nm]
if len(seg1) > 3 and len(seg2) > 3:
print(f"\nTwo-segment comparison (split at {args.split_nm} nm) - use this if the "
f"data looks like a STEP rather than a smooth curve (e.g. a detector splice):")
print(f" segment 1 (< {args.split_nm}nm, n={len(seg1)}): "
f"mean={seg1.mean():.2f} px, std={seg1.std():.2f} px")
print(f" segment 2 (>= {args.split_nm}nm, n={len(seg2)}): "
f"mean={seg2.mean():.2f} px, std={seg2.std():.2f} px")
print(f" step offset between segments: {seg1.mean() - seg2.mean():.2f} px")

if pixel_size:
print(f" pixel size from header: {pixel_size} -> "
f"keystone p2v ~ {ptp_fit * pixel_size:.3f} (same units as pixel size)")

try:
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
plt.figure(figsize=(11, 5))
excluded = ~valid & ~np.isnan(edge_positions)
if excluded.sum() > 0:
plt.scatter(wavelengths[excluded], edge_positions[excluded], s=6, alpha=0.3,
color="lightgray", label="excluded (window-edge / out of range)")
plt.scatter(wl_valid, pos_valid, s=6, alpha=0.4, label="measured (sub-pixel)", color="steelblue")
plt.plot(wl_valid, fit_curve, color="darkorange", linewidth=2,
label=f"quadratic fit (p2v={ptp_fit:.2f} px)")
plt.axhline(ref_pos, color="gray", linestyle="--", alpha=0.5,
label=f"reference position (band {ref_band})")
plt.xlabel("Wavelength (nm)")
plt.ylabel("Detected edge column (samples)")
plt.title("Keystone estimate: spatial edge position vs wavelength")
plt.legend()
plt.tight_layout()
plt.savefig(args.out, dpi=120)
print(f"\nSaved plot to {args.out}")
except ImportError:
print("\n(matplotlib not installed - numeric results above still usable)")


if __name__ == "__main__":
main()


 

Stima Smile in FigSpec FS60-CL

Per stimare lo smile effect in FigSpec FS60-CL e' stato usato il file raw dei digital numbers cercando di isolare lo spostamento  dell minimo dell'assorbimento dell'Ossigeno a 760 nm

 

 

prima di tutto si vede che il grafico non e' simmetrico indice di problemi a livello ottico. In generale pero' lo smile e' stimato inferiore ad 1 banda (indicativamente 0.2) quindi e' trascurabile

L'aspetto da sottilineare (non legata allo smile) e' che sembra ci sia un offset di calibrazione spettrale di circa 2 nm (l'assorbimento reale e' di letteratura a 760 nm mentre il FigSpec lo vede a 762 nm) 


python3 detect_smile.py figspec.hdr figspec.spe --out smile_diagnostic.png

 

#!/usr/bin/env python3
"""
Estimates spectral smile by tracking the position of a known, narrow
atmospheric absorption feature (default: O2-A band ~760-762nm) across the
spatial (sample/across-track) dimension of an ENVI BIL hyperspectral cube.

If the sensor has no smile, the absorption minimum should sit at the same
wavelength for every column. If it drifts smoothly from one edge to the
other, that drift IS the smile curve.

Usage:
python3 detect_smile.py figspec_reflectance_rid_savgol.hdr \
figspec_reflectance_rid_savgol.img \
--out smile_diagnostic.png

Optional: --feature-nm 760 --search-window-nm 25 (O2-A, default)
--feature-nm 940 --search-window-nm 40 (water vapor, cross-check)
"""

import os
import re
import argparse
import numpy as np




def main():
ap = argparse.ArgumentParser()
ap.add_argument("hdr_path")
ap.add_argument("data_path")
ap.add_argument("--out", default="smile_diagnostic.png")
ap.add_argument("--feature-nm", type=float, default=760.0,
help="Center wavelength of the known absorption feature (default: O2-A 760nm)")
ap.add_argument("--search-window-nm", type=float, default=25.0,
help="Half-width of the search window around feature-nm")
ap.add_argument("--line-stride", type=int, default=5,
help="Use every Nth line when averaging (speed/memory tradeoff)")
args = ap.parse_args()

meta = parse_hdr(args.hdr_path)
samples, lines, bands = meta["samples"], meta["lines"], meta["bands"]
dtype = meta["dtype"]
interleave = meta["interleave"]
wavelengths = meta["wavelengths"]

if interleave != "bil":
raise ValueError("This script currently assumes BIL interleave.")

cube = np.memmap(args.data_path, dtype=dtype, mode="r",
shape=(lines, bands, samples))

print(f"Averaging every {args.line_stride}th line "
f"({lines // args.line_stride} lines used) to build a per-column mean spectrum...")
idx_lines = np.arange(0, lines, args.line_stride)
accum = np.zeros((bands, samples), dtype=np.float64)
for i, li in enumerate(idx_lines):
accum += cube[li, :, :]
if i % 200 == 0:
print(f" {i}/{len(idx_lines)}", end="\r")
mean_spectrum = (accum / len(idx_lines)).astype(np.float32) # (bands, samples)
print(f"\nDone averaging. Shape: {mean_spectrum.shape}")

lo_nm = args.feature_nm - args.search_window_nm
hi_nm = args.feature_nm + args.search_window_nm
band_mask = (wavelengths >= lo_nm) & (wavelengths <= hi_nm)
band_idxs = np.where(band_mask)[0]
if len(band_idxs) < 5:
raise ValueError(f"Too few bands in [{lo_nm},{hi_nm}]nm window; "
f"check --feature-nm/--search-window-nm vs your wavelength range.")
print(f"Search window: bands {band_idxs[0]}-{band_idxs[-1]} "
f"({wavelengths[band_idxs[0]]:.1f}-{wavelengths[band_idxs[-1]]:.1f} nm)")

sub_wavelengths = np.full(samples, np.nan)
for col in range(samples):
window_vals = mean_spectrum[band_idxs, col].astype(np.float64)
if np.all(window_vals <= 1e-6):
continue # degenerate/empty column
local_min = np.argmin(window_vals)
# sub-pixel parabolic refinement (skip if at window edge)
if 0 < local_min < len(window_vals) - 1:
y0, y1, y2 = window_vals[local_min - 1], window_vals[local_min], window_vals[local_min + 1]
denom = (y0 - 2 * y1 + y2)
delta = 0.5 * (y0 - y2) / denom if abs(denom) > 1e-9 else 0.0
delta = np.clip(delta, -1, 1)
else:
delta = 0.0
band_global = band_idxs[local_min]
# interpolate wavelength at sub-pixel band position
if 0 <= band_global + int(np.sign(delta)) < bands:
w0 = wavelengths[band_global]
w_next = wavelengths[band_global + (1 if delta >= 0 else -1)]
sub_wl = w0 + delta * (w_next - w0)
else:
sub_wl = wavelengths[band_global]
sub_wavelengths[col] = sub_wl

valid = ~np.isnan(sub_wavelengths)
cols = np.arange(samples)[valid]
vals = sub_wavelengths[valid]

if len(vals) < 10:
print("Too few valid columns to estimate smile reliably.")
return

# quadratic fit (typical smile shape) for a smooth summary curve
coeffs = np.polyfit(cols, vals, 2)
fit_curve = np.polyval(coeffs, cols)

ptp_measured = vals.max() - vals.min()
ptp_fit = fit_curve.max() - fit_curve.min()

print(f"\nAbsorption feature position across {samples} columns:")
print(f" raw measured range: {vals.min():.2f} - {vals.max():.2f} nm "
f"(peak-to-valley = {ptp_measured:.2f} nm)")
print(f" smooth quadratic fit range: {fit_curve.min():.2f} - {fit_curve.max():.2f} nm "
f"(peak-to-valley = {ptp_fit:.2f} nm)")
print(f" quadratic fit coefficients (a*x^2+b*x+c): {coeffs}")

band_spacing_nm = np.mean(np.diff(wavelengths))
print(f" approx. band spacing: {band_spacing_nm:.2f} nm/band")
print(f" smile in band units: ~{ptp_fit / band_spacing_nm:.2f} bands peak-to-valley")

try:
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
plt.figure(figsize=(11, 5))
plt.scatter(cols, vals, s=6, alpha=0.4, label="measured (sub-pixel)", color="steelblue")
plt.plot(cols, fit_curve, color="darkorange", linewidth=2,
label=f"quadratic fit (p2v={ptp_fit:.2f} nm)")
plt.axhline(args.feature_nm, color="gray", linestyle="--", alpha=0.5,
label=f"nominal {args.feature_nm:.0f} nm")
plt.xlabel("Sample (across-track column)")
plt.ylabel(f"Detected absorption minimum (nm)")
plt.title("Spectral smile estimate")
plt.legend()
plt.tight_layout()
plt.savefig(args.out, dpi=120)
print(f"\nSaved plot to {args.out}")
except ImportError:
print("\n(matplotlib not installed - numeric results above still usable)")


if __name__ == "__main__":
main()



mercoledì 1 luglio 2026

Allarme prossimita' ADBS

Uno degli incubi mentre volo con il drone e' di imbattermi in Pegaso 1, l'elisoccorso della Toscana

La quota di volo di Pegaso minima (tranne quando decolla ed atterra) sembra essere di 1000 piedi (circa 300 m) quindi ben al di sopra dei 120 m di quota massima dei droni 

l'idea e' comunque di avere una ground station con allarme di prossimita' usando un dongle RTL-SDR impostando la propria posizione ed un raggio di 2 miglia nautiche

su Debian 13 si usa (scaricabile tramite apt)  

dump1090-fa    --net   --net-sbs-port 30003   --net-ro-port 30002   --net-http-port 8080   --lat 43.7696 --lon 11.2558  

il comando soprastante 

usando il programma Python (grazie Claude) si leggono i dati in uscita dump1009-fa e si selezionano solo gli aeromobili in 3 miglia nautiche di distanza e separazione verticale inferiore a 1500 piedi

 python3 adsb_proximity_alarm.py sbs --host 127.0.0.1 --port 30003     --lat 43.7696 --lon 11.2558 --elev-ft 150     --radius-nm 3 --vsep-ft 1500 -v 

 

 


#!/usr/bin/env python3
"""
adsb_proximity_alarm.py

Proximity alarm for ADS-B traffic received via an RTL-SDR + dump1090
(dump1090-fa, dump1090-mutability, readsb, or the dump1090 fork bundled
with Stratux).

It watches live aircraft positions and raises an alert when any aircraft
comes within a configurable horizontal distance AND vertical separation
of a reference point (e.g. your ground station / launch site).

Two input modes are supported, pick whichever your setup exposes:

1. SBS / BaseStation feed (raw TCP text feed, classic dump1090 port 30003)
Most dump1090 builds expose this. Stratux's internal dump1090 usually
does too, though it may not be reachable outside localhost unless you
tunnel it (e.g. `ssh -L 30003:localhost:30003 pi@stratux.local`).

2. aircraft.json polling (HTTP), as served by dump1090-fa/readsb/tar1090,
typically at http://<host>/dump1090-fa/data/aircraft.json or similar.
Check your install; the path varies by distro/build.

Usage examples
--------------
SBS mode, ground station at 43.7696N 11.2558E (Florence area), alarm inside
3 NM / 1500 ft:

python3 adsb_proximity_alarm.py sbs \\
--host 127.0.0.1 --port 30003 \\
--lat 43.7696 --lon 11.2558 --elev-ft 150 \\
--radius-nm 3 --vsep-ft 1500

JSON polling mode:

python3 adsb_proximity_alarm.py json \\
--url http://127.0.0.1/dump1090-fa/data/aircraft.json \\
--lat 43.7696 --lon 11.2558 --radius-nm 3 --vsep-ft 1500

No extra dependencies beyond the standard library and `requests`
(only needed for JSON mode: pip install requests).
"""

import argparse
import math
import socket
import sys
import time
import urllib.request
import json as jsonlib
from dataclasses import dataclass, field
from datetime import datetime, timezone


# --------------------------------------------------------------------------
# Geometry helpers
# --------------------------------------------------------------------------

EARTH_RADIUS_NM = 3440.065 # mean earth radius in nautical miles


def haversine_nm(lat1, lon1, lat2, lon2):
"""Great-circle distance in nautical miles."""
phi1, phi2 = math.radians(lat1), math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dlambda = math.radians(lon2 - lon1)
a = (math.sin(dphi / 2) ** 2
+ math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2)
return 2 * EARTH_RADIUS_NM * math.asin(math.sqrt(a))


def bearing_deg(lat1, lon1, lat2, lon2):
"""Initial bearing from point 1 to point 2, in degrees (0=N, 90=E)."""
phi1, phi2 = math.radians(lat1), math.radians(lat2)
dlambda = math.radians(lon2 - lon1)
y = math.sin(dlambda) * math.cos(phi2)
x = (math.cos(phi1) * math.sin(phi2)
- math.sin(phi1) * math.cos(phi2) * math.cos(dlambda))
return (math.degrees(math.atan2(y, x)) + 360) % 360


def compass(deg):
dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
return dirs[int((deg + 22.5) // 45) % 8]


# --------------------------------------------------------------------------
# Aircraft state
# --------------------------------------------------------------------------

@dataclass
class Aircraft:
icao: str
callsign: str = ""
lat: float = None
lon: float = None
alt_ft: float = None
last_seen: float = field(default_factory=time.monotonic)
last_alert: float = 0.0

def has_position(self):
return self.lat is not None and self.lon is not None


class AircraftTable:
"""Keeps track of recently seen aircraft and prunes stale ones."""

def __init__(self, stale_after_s=60):
self.aircraft = {}
self.stale_after_s = stale_after_s

def update(self, icao, callsign=None, lat=None, lon=None, alt_ft=None):
ac = self.aircraft.get(icao)
if ac is None:
ac = Aircraft(icao=icao)
self.aircraft[icao] = ac
if callsign:
ac.callsign = callsign.strip()
if lat is not None:
ac.lat = lat
if lon is not None:
ac.lon = lon
if alt_ft is not None:
ac.alt_ft = alt_ft
ac.last_seen = time.monotonic()
return ac

def prune(self):
now = time.monotonic()
dead = [icao for icao, ac in self.aircraft.items()
if now - ac.last_seen > self.stale_after_s]
for icao in dead:
del self.aircraft[icao]


# --------------------------------------------------------------------------
# Alerting
# --------------------------------------------------------------------------

RED = "\033[91m"
YELLOW = "\033[93m"
GREEN = "\033[92m"
RESET = "\033[0m"


def alert(ac: Aircraft, ref_lat, ref_lon, dist_nm, vsep_ft, bell=True):
brg = bearing_deg(ref_lat, ref_lon, ac.lat, ac.lon)
ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
tag = ac.callsign if ac.callsign else ac.icao
msg = (f"{RED}[ALERT {ts}Z]{RESET} {tag} (ICAO {ac.icao}) "
f"{dist_nm:.1f} NM {compass(brg)} ({brg:5.1f}°), "
f"alt {ac.alt_ft:.0f} ft, vertical sep {vsep_ft:.0f} ft")
print(msg)
if bell:
sys.stdout.write("\a")
sys.stdout.flush()


def status_line(ac: Aircraft, dist_nm, vsep_ft):
tag = ac.callsign if ac.callsign else ac.icao
print(f"{YELLOW} near{RESET} {tag:8s} {dist_nm:5.1f} NM "
f"alt {ac.alt_ft:6.0f} ft vsep {vsep_ft:6.0f} ft")


# --------------------------------------------------------------------------
# Core proximity check
# --------------------------------------------------------------------------

def check_proximity(ac: Aircraft, args):
if not ac.has_position() or ac.alt_ft is None:
return
dist_nm = haversine_nm(args.lat, args.lon, ac.lat, ac.lon)
vsep_ft = abs(ac.alt_ft - args.elev_ft)

if dist_nm <= args.radius_nm and vsep_ft <= args.vsep_ft:
now = time.monotonic()
if now - ac.last_alert >= args.cooldown:
alert(ac, args.lat, args.lon, dist_nm, vsep_ft, bell=not args.no_bell)
ac.last_alert = now
elif args.verbose and dist_nm <= args.radius_nm * 2.0:
# Show near-miss traffic even if outside the hard alert ring,
# useful for situational awareness.
status_line(ac, dist_nm, vsep_ft)


# --------------------------------------------------------------------------
# SBS (BaseStation) feed parser
# --------------------------------------------------------------------------
#
# Format reference (dump1090 SBS output), comma separated fields:
# MSG,type,session,aircraft,hex,flight,date,time,date,time,
# callsign,altitude,gspeed,track,lat,lon,vrate,squawk,alert,emergency,spi,onground
#
# Relevant message types:
# MSG,1 -> callsign (field 10)
# MSG,3 -> airborne position: altitude(11), lat(14), lon(15)
# MSG,4 -> velocity
# MSG,5/6/7/8 -> various, mostly ignored here

def run_sbs(args):
table = AircraftTable()
print(f"Connecting to SBS feed at {args.host}:{args.port} ...")
while True:
try:
with socket.create_connection((args.host, args.port), timeout=10) as sock:
print(f"{GREEN}Connected.{RESET} Watching for traffic within "
f"{args.radius_nm} NM / {args.vsep_ft} ft of "
f"{args.lat:.4f},{args.lon:.4f} (elev {args.elev_ft} ft)...")
buf = b""
sock.settimeout(5)
last_prune = time.monotonic()
while True:
try:
chunk = sock.recv(4096)
except socket.timeout:
chunk = b""
if chunk == b"" and time.monotonic() - last_prune > 30:
# no data for a while but connection alive; keep looping
pass
buf += chunk
while b"\n" in buf:
line, buf = buf.split(b"\n", 1)
line = line.decode(errors="ignore").strip()
if line:
handle_sbs_line(line, table, args)
if time.monotonic() - last_prune > 15:
table.prune()
last_prune = time.monotonic()
except (ConnectionRefusedError, OSError) as e:
print(f"Connection error ({e}); retrying in 5s...")
time.sleep(5)


def handle_sbs_line(line, table, args):
fields = line.split(",")
if len(fields) < 2 or fields[0] != "MSG":
return
msg_type = fields[1]
if len(fields) < 5:
return
icao = fields[4].strip()
if not icao:
return

callsign = None
lat = lon = alt_ft = None

try:
if msg_type == "1" and len(fields) > 10:
callsign = fields[10]
elif msg_type == "3" and len(fields) > 15:
alt_ft = float(fields[11]) if fields[11] else None
lat = float(fields[14]) if fields[14] else None
lon = float(fields[15]) if fields[15] else None
elif msg_type in ("5", "6", "7", "8") and len(fields) > 11 and fields[11]:
alt_ft = float(fields[11])
except ValueError:
return

ac = table.update(icao, callsign=callsign, lat=lat, lon=lon, alt_ft=alt_ft)
check_proximity(ac, args)


# --------------------------------------------------------------------------
# aircraft.json polling
# --------------------------------------------------------------------------

def run_json(args):
table = AircraftTable()
print(f"Polling {args.url} every {args.poll_interval}s ...")
print(f"Watching for traffic within {args.radius_nm} NM / {args.vsep_ft} ft of "
f"{args.lat:.4f},{args.lon:.4f} (elev {args.elev_ft} ft)...")
last_prune = time.monotonic()
while True:
try:
with urllib.request.urlopen(args.url, timeout=5) as resp:
data = jsonlib.loads(resp.read())
except Exception as e:
print(f"Fetch error ({e}); retrying in {args.poll_interval}s...")
time.sleep(args.poll_interval)
continue

aircraft_list = data.get("aircraft", data if isinstance(data, list) else [])
for entry in aircraft_list:
icao = (entry.get("hex") or entry.get("icao") or "").strip()
if not icao:
continue
callsign = entry.get("flight") or entry.get("callsign")
lat = entry.get("lat")
lon = entry.get("lon")
alt_ft = entry.get("alt_baro", entry.get("altitude"))
try:
alt_ft = float(alt_ft) if alt_ft not in (None, "ground") else None
except (TypeError, ValueError):
alt_ft = None

ac = table.update(icao, callsign=callsign, lat=lat, lon=lon, alt_ft=alt_ft)
check_proximity(ac, args)

if time.monotonic() - last_prune > 15:
table.prune()
last_prune = time.monotonic()

time.sleep(args.poll_interval)


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------

def build_parser():
# Shared options live on a "parent" parser so they can be passed either
# before or after the sbs/json subcommand, e.g. both of these work:
# adsb_proximity_alarm.py --lat 43.7 --lon 11.2 sbs --port 30003
# adsb_proximity_alarm.py sbs --port 30003 --lat 43.7 --lon 11.2
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--lat", type=float, required=True,
help="Reference latitude (your ground station / launch site)")
common.add_argument("--lon", type=float, required=True,
help="Reference longitude")
common.add_argument("--elev-ft", type=float, default=0.0,
help="Reference point elevation, feet MSL (default 0)")
common.add_argument("--radius-nm", type=float, default=3.0,
help="Horizontal alert radius in nautical miles (default 3)")
common.add_argument("--vsep-ft", type=float, default=1500.0,
help="Vertical separation threshold in feet (default 1500)")
common.add_argument("--cooldown", type=float, default=30.0,
help="Seconds before re-alerting on the same aircraft (default 30)")
common.add_argument("--no-bell", action="store_true",
help="Disable terminal bell on alert")
common.add_argument("-v", "--verbose", action="store_true",
help="Also print traffic within 2x the alert radius")

p = argparse.ArgumentParser(
description="ADS-B proximity alarm using RTL-SDR / dump1090 feeds")
sub = p.add_subparsers(dest="mode", required=True)

sbs = sub.add_parser("sbs", parents=[common],
help="Connect to raw SBS/BaseStation TCP feed")
sbs.add_argument("--host", default="127.0.0.1")
sbs.add_argument("--port", type=int, default=30003)

js = sub.add_parser("json", parents=[common],
help="Poll an aircraft.json HTTP endpoint")
js.add_argument("--url", required=True,
help="Full URL to aircraft.json")
js.add_argument("--poll-interval", type=float, default=1.0,
help="Seconds between polls (default 1.0)")

return p


def main():
args = build_parser().parse_args()
try:
if args.mode == "sbs":
run_sbs(args)
else:
run_json(args)
except KeyboardInterrupt:
print("\nStopped.")


if __name__ == "__main__":
main()


Cicala

Alla fine sono riuscito a vedere una cicala viva (per adesso avevo visto solo le mute  o come dicono quelli bravi esuvie) ed ho capito perch...