Overview
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ucls_infra_gateway_v3.py
-------------------------------------------------------------------------------------------------
Universal, SAFE‑BY‑DEFAULT interoperability gateway — now with SIGNAL RX integration
and dash‑ready publishers.
Includes:
• OpenADR 2.0b VEN registration (DRY‑RUN by default), event polling (read‑only)
• IEC‑61850 path helpers (LN/DO/DA traversal) + browse (read‑only/sim)
• TM Forum Open APIs (OSS/BSS) read‑only GETs (TMF622/632/633/639/641)
• Telecom adapters: SNMP, NETCONF, gNMI (read‑only; stubs if libs absent)
• Edge adapters: Modbus/TCP, OPC UA, MQTT subscriber (read‑only)
• PLC linguistics codec + monitor (sim)
• NEW: Spectrum & Signal RX (SDR) — scan / analyze IQ / demod FM (RX‑only)
• NEW: Publishers for dashboards — MQTT & WebSocket client push
Safety rails:
- LIVE mode OFF by default (UCLS_LIVE=0) → simulation/dry‑run only
- Host/device allowlists for any live I/O:
* Grid/Telecom/OSS: UCLS_WHITELIST="host1,1.2.3.4,..."
* SDR devices: UCLS_SDR_WHITELIST="rtlsdr,soapy:driver=rtlsdr"
- STRICTLY READ‑ONLY. No control/TX. No decryption of restricted/encrypted traffic.
- Signed JSON audit record for every operation.
CLI EXAMPLES
# Simulated panorama
python ucls_infra_gateway_v3.py demo
# OpenADR CreatePartyRegistration (DRY‑RUN unless LIVE & whitelisted)
UCLS_LIVE=1 UCLS_WHITELIST="vtn.example.com" \
python ucls_infra_gateway_v3.py openadr-create --vtn https://vtn.example.com/oadr \
--ven VEN_123 --profile 2.0b --transport https://ven.example.com/callback --token "***"
# IEC‑61850 helpers
python ucls_infra_gateway_v3.py iec-parse --path "LD0/MMXU1.TotW.mag.f"
python ucls_infra_gateway_v3.py iec-browse --host substation.sim --path "LD0/MMXU1.TotW.mag.f"
# TM Forum (read‑only)
UCLS_LIVE=1 UCLS_WHITELIST="oss.example.com" \
python ucls_infra_gateway_v3.py tmf-get --base https://oss.example.com/tmf-api \
--api tmf622-product-order --id 123 --token "***"
# NEW — Signal scan (RX only). Publishers optional.
UCLS_LIVE=1 UCLS_SDR_WHITELIST="rtlsdr,soapy:driver=rtlsdr" \
python ucls_infra_gateway_v3.py signal-scan --device rtlsdr --center 100e6 --span 20e6 --step 2e6 \
--seconds 3 --samp-rate 2.4e6 --mqtt-broker mqtt.local --mqtt-topic ucls/spectrum \
--ws-url ws://localhost:8765
# NEW — Analyze IQ file (RX only)
python ucls_infra_gateway_v3.py signal-analyze-iq --file wifi.iq --samp-rate 20e6 --center 2.437e9
# NEW — Demod FM from IQ (RX only), write WAV
python ucls_infra_gateway_v3.py signal-demod-fm --file fm.iq --samp-rate 240000 --deemph 75e-6 --audio-out fm.wav
"""
from __future__ import annotations
import argparse, json, os, re, socket, time, hmac, hashlib, sys, struct, math, pathlib
from dataclasses import dataclass, asdict
from typing import Any, Dict, List, Optional, Tuple
# ============================== SAFETY & AUDIT ===============================
def is_live() -> bool:
return os.environ.get("UCLS_LIVE","0") in ("1","true","TRUE","yes","YES")
def whitelist() -> List[str]:
raw = os.environ.get("UCLS_WHITELIST","")
return [h.strip() for h in raw.split(",") if h.strip()]
def is_host_allowed(host_or_url: str) -> bool:
host = host_or_url
m = re.match(r"^https?://([^/:]+)", host_or_url, re.I)
if m: host = m.group(1)
allowed = whitelist()
if not allowed: return False
try:
ip = socket.gethostbyname(host)
except Exception:
ip = None
return host in allowed or (ip and ip in allowed)
def sdr_allowlist() -> List[str]:
raw = os.environ.get("UCLS_SDR_WHITELIST","")
return [x.strip() for x in raw.split(",") if x.strip()]
def device_allowed(dev: str) -> bool:
w = sdr_allowlist()
return any(dev.lower().startswith(p.lower()) for p in w) if w else False
AUDIT_KEY = os.environ.get("UCLS_AUDIT_KEY","ucls-dev-key").encode()
@dataclass
class AuditRecord:
ts: float
actor: str
action: str
target: str
payload: Dict[str, Any]
result_preview: str
signature: str = ""
def sign_and_print(action: str, target: str, payload: Dict[str,Any], result: Any):
rec = AuditRecord(
time.time(), os.environ.get("USER","ucls"), action, target,
payload,
(json.dumps(result, ensure_ascii=False)[:300] + "…") if len(str(result)) > 320 else (result if isinstance(result,str) else json.dumps(result,ensure_ascii=False))
)
body = json.dumps({
"ts": rec.ts, "actor": rec.actor, "action": rec.action, "target": rec.target,
"payload": rec.payload, "result_preview": rec.result_preview
}, sort_keys=True, ensure_ascii=False).encode()
rec.signature = hmac.new(AUDIT_KEY, body, hashlib.sha256).hexdigest()
print(json.dumps(asdict(rec), ensure_ascii=False))
# ============================== OPTIONAL LIBS ================================
# HTTP
try:
import requests
HAS_REQUESTS = True
except Exception:
HAS_REQUESTS = False
# SNMP
try:
from pysnmp.hlapi import (
SnmpEngine, UdpTransportTarget, CommunityData,
ContextData, ObjectType, ObjectIdentity, getCmd,
UsmUserData, usmHMACSHAAuthProtocol, usmHMACMD5AuthProtocol,
usmAesCfb128Protocol, usmDESPrivProtocol
)
HAS_PYSNMP = True
except Exception:
HAS_PYSNMP = False
# NETCONF
try:
from ncclient import manager as nc_manager
HAS_NCCLIENT = True
except Exception:
HAS_NCCLIENT = False
# gNMI (stubbed)
HAS_GNMI = False
# Modbus
try:
from pymodbus.client import ModbusTcpClient
HAS_PYMODBUS = True
except Exception:
HAS_PYMODBUS = False
# OPC UA
try:
from opcua import Client as OPCUAClient
HAS_OPCUA = True
except Exception:
HAS_OPCUA = False
# MQTT pub/sub
try:
import paho.mqtt.client as paho_mqtt
HAS_PAHO = True
except Exception:
HAS_PAHO = False
# WebSocket client (prefer websocket-client for sync push)
try:
import websocket as ws_client # websocket-client
HAS_WSCLIENT = True
except Exception:
HAS_WSCLIENT = False
# NumPy/SciPy for signal analysis
try:
import numpy as np
HAS_NUMPY = True
except Exception:
HAS_NUMPY = False
try:
from scipy.signal import welch, butter, lfilter, decimate
HAS_SCIPY = True
except Exception:
HAS_SCIPY = False
# SDR frontends
try:
import rtlsdr # pyrtlsdr
HAS_RTL = True
except Exception:
HAS_RTL = False
try:
import SoapySDR # type: ignore
from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_CF32
HAS_SOAPY = True
except Exception:
HAS_SOAPY = False
# ============================== OPENADR (REG) ================================
class OpenADR:
@staticmethod
def create_party_registration(vtn: str, ven_id: str, profile: str, transport_address: str, token: Optional[str]) -> Dict[str,Any]:
payload = {
"oadrCreatePartyRegistration": {
"venID": ven_id, "oadrProfileName": profile,
"oadrTransportAddress": transport_address,
"oadrTransportName": "simpleHttp", "oadrXmlSignature": False
}
}
if is_live() and is_host_allowed(vtn) and HAS_REQUESTS:
r = requests.post(f"{vtn}/CreatePartyRegistration", json=payload,
headers={"Authorization": f"Bearer {token}"} if token else {}, timeout=15)
result = {"status": r.status_code, "body": (r.text[:400]+"…")}
else:
result = {"dry_run": True, "target": f"{vtn}/CreatePartyRegistration", "payload": payload}
sign_and_print("openadr-create", vtn, {"ven": ven_id, "profile": profile}, result)
return result
@staticmethod
def query_registration(vtn: str, ven_id: str, token: Optional[str]) -> Dict[str,Any]:
payload = {"oadrQueryRegistration": {"venID": ven_id}}
if is_live() and is_host_allowed(vtn) and HAS_REQUESTS:
r = requests.post(f"{vtn}/QueryRegistration", json=payload,
headers={"Authorization": f"Bearer {token}"} if token else {}, timeout=15)
result = {"status": r.status_code, "body": (r.text[:400]+"…")}
else:
result = {"dry_run": True, "target": f"{vtn}/QueryRegistration", "payload": payload}
sign_and_print("openadr-query", vtn, {"ven": ven_id}, result)
return result
@staticmethod
def cancel_party_registration(vtn: str, ven_id: str, token: Optional[str]) -> Dict[str,Any]:
payload = {"oadrCancelPartyRegistration": {"venID": ven_id}}
if is_live() and is_host_allowed(vtn) and HAS_REQUESTS:
r = requests.post(f"{vtn}/CancelPartyRegistration", json=payload,
headers={"Authorization": f"Bearer {token}"} if token else {}, timeout=15)
result = {"status": r.status_code, "body": (r.text[:400]+"…")}
else:
result = {"dry_run": True, "target": f"{vtn}/CancelPartyRegistration", "payload": payload}
sign_and_print("openadr-cancel", vtn, {"ven": ven_id}, result)
return result
class OpenADREvents:
def poll(self, vtn: str, ven: str, token: Optional[str]) -> Dict[str,Any]:
if not is_live() or not HAS_REQUESTS:
data = {"vtn": vtn, "ven": ven, "events":[{"id":"evt-1","start":"2025-08-18T20:00:00Z","duration":"PT30M","level":1}]}
else:
if not is_host_allowed(vtn): raise PermissionError("Host not allowed")
r = requests.get(f"{vtn}/Events?ven={ven}", headers={"Authorization": f"Bearer {token}"} if token else {}, timeout=10)
data = {"status": r.status_code, "body": (r.text[:400]+"…")}
sign_and_print("openadr-events", vtn, {"ven": ven}, data)
return data
# ============================== IEC‑61850 HELPERS ============================
class IEC61850Path:
PATH_RE = re.compile(r"^(?P<ld>[^/]+)/(?P<ln>[A-Z]{4}\d+)\.(?P<rest>.+)$")
@staticmethod
def parse(path: str) -> Dict[str,Any]:
m = IEC61850Path.PATH_RE.match(path)
if not m: raise ValueError("Invalid IEC‑61850 path (expect 'LDX/LN.DO.DA[.sub]')")
ld, ln, rest = m.group("ld"), m.group("ln"), m.group("rest")
parts = rest.split("."); do = parts[0]; da = parts[1:]
return {"ld": ld, "ln": ln, "do": do, "da": da, "raw": path}
@staticmethod
def to_string(parsed: Dict[str,Any]) -> str:
return f"{parsed['ld']}/{parsed['ln']}." + ".".join([parsed["do"], *parsed["da"]])
class IEC61850Browser:
def browse(self, host: str, path: str) -> Dict[str,Any]:
parsed = IEC61850Path.parse(path)
if not is_live():
result = {"path": parsed, "value": 12345.67, "quality":"valid", "ts": time.time(), "simulated": True}
else:
if not is_host_allowed(host): raise PermissionError("Host not in allowlist.")
# Real MMS read would go here.
result = {"path": parsed, "value": None, "quality":"unknown", "ts": time.time()}
sign_and_print("iec-browse", host, {"path": path}, result); return result
# ============================== TM FORUM (READ‑ONLY) =========================
class TMForumAPI:
API_MAP = {
"tmf622-product-order": "/productOrderingManagement/v4/productOrder",
"tmf632-party": "/partyManagement/v5/party",
"tmf633-service-spec": "/serviceCatalogManagement/v4/serviceSpecification",
"tmf639-service": "/serviceInventoryManagement/v4/service",
"tmf641-service-order": "/serviceOrderingManagement/v4/serviceOrder",
}
@staticmethod
def _suffix(base: str) -> str:
return "/tmf-api" if base.rstrip("/").endswith("/tmf-api") else ""
@staticmethod
def get(base: str, api: str, _id: Optional[str], token: Optional[str]) -> Dict[str,Any]:
if api not in TMForumAPI.API_MAP: raise ValueError(f"Unknown TMF API key: {api}")
path = TMForumAPI.API_MAP[api]
url = f"{base.rstrip('/').rstrip(TMForumAPI._suffix(base))}{path}"
url = f"{url}/{_id}" if _id else f"{url}?limit=10"
if is_live() and is_host_allowed(url) and HAS_REQUESTS:
r = requests.get(url, headers={"Authorization": f"Bearer {token}"} if token else {}, timeout=20)
result = {"status": r.status_code, "body": (r.text[:400]+"…")}
else:
result = {"dry_run": True, "url": url}
sign_and_print("tmf-get", url, {"api": api, "id": _id or "list"}, result); return result
# ============================== TELECOM (READ‑ONLY) ==========================
class SNMPMonitor:
def get(self, host: str, oid: str) -> Dict[str,Any]:
if not is_live() or not HAS_PYSNMP:
result = {"host": host, "oid": oid, "value": "sim-uptime-123456"}
sign_and_print("snmp-get-sim", host, {"oid": oid}, result); return result
errorIndication, errorStatus, errorIndex, varBinds = next(getCmd(
SnmpEngine(), CommunityData("public"),
UdpTransportTarget((host, 161), timeout=3, retries=1),
ContextData(), ObjectType(ObjectIdentity(oid))
))
if errorIndication or errorStatus:
result = {"error": str(errorIndication or errorStatus)}
else:
result = {str(name): str(val) for name,val in varBinds}
sign_and_print("snmp-get", host, {"oid": oid}, result); return result
class NETCONFClient:
def get(self, host: str, port: int, user: str, password: str, filter_subtree: Optional[str]=None) -> Dict[str,Any]:
if not is_live() or not HAS_NCCLIENT:
result = {"host": host, "op":"get", "filter": filter_subtree or "<interfaces/>", "xml":"<simulated/>"}
sign_and_print("netconf-get-sim", f"{host}:{port}", {"filter": filter_subtree}, result); return result
if not is_host_allowed(host): raise PermissionError("Host not allowed")
with nc_manager.connect(host=host, port=port, username=user, password=password,
hostkey_verify=True, allow_agent=False, look_for_keys=False, timeout=10) as m:
reply = m.get(('subtree', filter_subtree)) if filter_subtree else m.get()
xml = reply.xml
result = {"host":host, "xml": xml}
sign_and_print("netconf-get", f"{host}:{port}", {"filter": filter_subtree}, result); return result
class GNMIClient:
def get(self, host: str, path: str) -> Dict[str,Any]:
if not is_live() or not HAS_GNMI:
result = {"host": host, "path": path, "value": {"counter": 123456}}
sign_and_print("gnmi-get-sim", host, {"path": path}, result); return result
if not is_host_allowed(host): raise PermissionError("Host not allowed")
result = {"host": host, "path": path, "value": {}}
sign_and_print("gnmi-get", host, {"path": path}, result); return result
# ============================== EDGE (READ‑ONLY) =============================
class ModbusReader:
def read_holding(self, host: str, addr: int, count: int, port: int = 502) -> Dict[str,Any]:
if not is_live() or not HAS_PYMODBUS:
vals = [addr+i for i in range(count)]
result = {"host":host,"addr":addr,"count":count,"values":vals,"simulated":True}
sign_and_print("modbus-read-sim", f"{host}:{port}", {"addr":addr,"count":count}, result); return result
if not is_host_allowed(host): raise PermissionError("Host not allowed")
client = ModbusTcpClient(host=host, port=port); client.connect()
rr = client.read_holding_registers(addr, count, slave=1); client.close()
vals = list(rr.registers) if rr and hasattr(rr,"registers") else []
result = {"host":host,"addr":addr,"count":count,"values":vals}
sign_and_print("modbus-read", f"{host}:{port}", {"addr":addr,"count":count}, result); return result
class OPCUAReader:
def read_node(self, endpoint: str, node: str) -> Dict[str,Any]:
if not is_live() or not HAS_OPCUA:
result = {"endpoint":endpoint,"node":node,"value":"sim-value-42","simulated":True}
sign_and_print("opcua-read-sim", endpoint, {"node":node}, result); return result
host = re.sub(r"^opc\.tcp://","", endpoint).split(":")[0]
if not is_host_allowed(host): raise PermissionError("Endpoint host not allowed")
client = OPCUAClient(endpoint); client.connect()
v = client.get_node(node).get_value(); client.disconnect()
result = {"endpoint":endpoint,"node":node,"value":str(v)}
sign_and_print("opcua-read", endpoint, {"node":node}, result); return result
class MQTTMonitor:
def subscribe(self, broker: str, topic: str, seconds: int = 10) -> Dict[str,Any]:
if not is_live() or not HAS_PAHO:
result = {"broker":broker,"topic":topic,"received":[{"ts":time.time(),"payload":"sim-message"}]}
sign_and_print("mqtt-sub-sim", broker, {"topic":topic}, result); return result
if not is_host_allowed(broker): raise PermissionError("Broker not allowed")
messages: List[Dict[str,Any]] = []
def on_message(client, userdata, msg):
messages.append({"ts":time.time(),"topic":msg.topic,"payload": msg.payload.decode(errors="replace")[:200]})
client = paho_mqtt.Client()
client.on_message = on_message; client.connect(broker, 1883, 60)
client.subscribe(topic); client.loop_start()
t0 = time.time()
while time.time() - t0 < seconds: time.sleep(0.1)
client.loop_stop(); client.disconnect()
result = {"broker":broker,"topic":topic,"received":messages}
sign_and_print("mqtt-sub", broker, {"topic":topic,"seconds":seconds}, result); return result
# ============================== PUBLISHERS (NEW) =============================
def publish_mqtt(obj: Dict[str,Any], broker: Optional[str], topic: Optional[str]) -> Dict[str,Any]:
if not broker or not topic:
return {"published": False, "reason": "no broker/topic"}
if not is_live() or not HAS_PAHO:
res = {"published": False, "simulated": True, "broker": broker, "topic": topic}
sign_and_print("mqtt-publish-sim", broker, {"topic": topic}, res); return res
if not is_host_allowed(broker): raise PermissionError("MQTT broker not allowed")
client = paho_mqtt.Client()
client.connect(broker, 1883, 60)
payload = json.dumps(obj, ensure_ascii=False)
rc = client.publish(topic, payload, qos=0, retain=False)
client.disconnect()
res = {"published": rc.rc == 0, "broker": broker, "topic": topic}
sign_and_print("mqtt-publish", broker, {"topic": topic}, res); return res
def publish_ws(obj: Dict[str,Any], ws_url: Optional[str]) -> Dict[str,Any]:
if not ws_url:
return {"published": False, "reason": "no ws_url"}
payload = json.dumps(obj, ensure_ascii=False)
if not is_live() or not HAS_WSCLIENT:
res = {"published": False, "simulated": True, "ws_url": ws_url}
sign_and_print("ws-publish-sim", ws_url, {}, res); return res
# websocket-client (sync)
ws = ws_client.create_connection(ws_url, timeout=5)
ws.send(payload)
try:
ack = ws.recv()
except Exception:
ack = None
ws.close()
res = {"published": True, "ws_url": ws_url, "ack": (ack[:200]+"…") if isinstance(ack,str) and len(ack)>200 else ack}
sign_and_print("ws-publish", ws_url, {}, res); return res
# ============================== SIGNAL RX (NEW) ==============================
@dataclass
class Band:
name: str; f_lo: float; f_hi: float; service: str; notes: str = ""
# “All frequencies known and unknown”: we include broad ITU ranges + common services.
# For any frequency outside these heuristics, classification returns "Unspecified/Unknown".
BANDS: List[Band] = [
Band("ELF", 3.0, 30.0, "Extremely Low Frequency", "Geophysical; not RX here"),
Band("VLF", 3e3, 30e3, "Very Low Frequency", "Time signals, navigation"),
Band("LF", 30e3, 300e3, "Low Frequency", "Nav beacons"),
Band("MF", 300e3, 3e6, "Medium Frequency", "AM broadcast 530–1710 kHz"),
Band("HF", 3e6, 30e6, "High Frequency", "Shortwave, maritime/aviation"),
Band("VHF", 30e6, 300e6, "Very High Frequency", "FM, airband, VHF services"),
Band("UHF", 300e6, 3e9, "Ultra High Frequency", "TV, PMR, LTE, Wi‑Fi 2.4"),
Band("SHF", 3e9, 30e9, "Super High Frequency", "Wi‑Fi 5/6, radar, sat"),
Band("EHF", 30e9, 300e9, "Extremely High Frequency", "mmWave 5G, sensing"),
# ISM highlights
Band("ISM 433", 433e6, 435e6, "ISM", "SRD/LoRa EU"),
Band("ISM 868", 863e6, 870e6, "ISM", "LoRa EU"),
Band("ISM 902–928", 902e6, 928e6, "ISM", "LoRa/FSK US"),
Band("Wi‑Fi 2.4", 2.400e9, 2.4835e9, "WLAN/BLE", "802.11b/g/n/ax, BLE, Zigbee"),
Band("Wi‑Fi 5", 5.150e9, 5.875e9, "WLAN", "802.11a/n/ac"),
Band("Wi‑Fi 6E", 5.925e9, 7.125e9, "WLAN", "802.11ax 6 GHz"),
# Broadcast
Band("FM broadcast", 87.5e6, 108e6, "Broadcast FM", "WBFM ~200 kHz"),
Band("AM broadcast", 530e3, 1710e3, "Broadcast AM", "10/9 kHz step"),
# GNSS (example)
Band("GPS L1", 1.5754e9, 1.5756e9, "GNSS", "C/A"),
# Cellular coarse
Band("LTE/NR sub‑GHz", 700e6, 1.0e9, "Cellular", "Bands 12/13/20/28 etc."),
Band("LTE/NR mid‑band", 1.7e9, 2.7e9, "Cellular", "1800/1900/2100/2600"),
Band("5G NR n77/n78", 3.3e9, 4.2e9, "5G NR", "region‑specific"),
]
def bands_covering(freq_hz: float) -> List[Band]:
return [b for b in BANDS if b.f_lo <= freq_hz <= b.f_hi]
WIFI2_CHANNELS = {i: 2.412e9 + 5e6*(i-1) for i in range(1,14)}
def guess_service(freq_hz: float, occupied_bw_hz: Optional[float]=None) -> str:
# Broad heuristic. If nothing matches, declare unknown/unspecified (covers “undefined” ranges).
hits = bands_covering(freq_hz)
if 88e6 <= freq_hz <= 108e6 and (occupied_bw_hz or 150e3) >= 150e3:
return "Broadcast FM (WBFM)"
if 2.4e9 <= freq_hz <= 2.4835e9:
near = min(WIFI2_CHANNELS.items(), key=lambda kv: abs(kv[1]-freq_hz))
if abs(near[1]-freq_hz) < 2.5e6: return f"Wi‑Fi 2.4 GHz (ch {near[0]})"
return "BLE/Zigbee/WLAN 2.4 GHz"
if hits:
return ", ".join(sorted({h.service for h in hits}))
return "Unspecified / Unknown (region‑defined)"
# ---- IQ & PSD helpers
def load_iq_file(path: str, dtype: str = "complex64") -> Tuple[Optional["np.ndarray"], float]:
if not HAS_NUMPY: return None, 0.0
p = pathlib.Path(path); data = p.read_bytes()
if dtype == "complex64":
arr = np.frombuffer(data, dtype=np.complex64); return arr, 8.0
elif dtype == "int16":
raw = np.frombuffer(data, dtype=np.int16)
i = raw[0::2].astype(np.float32) / 32768.0; q = raw[1::2].astype(np.float32) / 32768.0
return (i + 1j*q).astype(np.complex64), 4.0
else:
raise ValueError("Unsupported dtype")
def bandwidth_estimate(Pxx: "np.ndarray", f: "np.ndarray", frac: float = 0.5) -> float:
pk = float(Pxx.max()); th = pk * frac
idx = np.where(Pxx >= th)[0]
if idx.size < 2: return 0.0
return float(f[idx[-1]] - f[idx[0]])
def psd_peaks(iq: "np.ndarray", fs: float, nfft: int = 4096) -> Dict[str, Any]:
if not HAS_NUMPY: return {"note": "numpy not available"}
if HAS_SCIPY:
f, Pxx = welch(iq, fs=fs, nperseg=nfft, return_onesided=False, scaling="density")
else:
seg = iq[:nfft] if iq.size >= nfft else np.pad(iq, (0, nfft - iq.size))
P = np.fft.fftshift(np.abs(np.fft.fft(seg))**2) / len(seg)
f = np.fft.fftshift(np.fft.fftfreq(len(seg), d=1.0/fs)); Pxx = P
idx = int(np.argmax(Pxx))
cf = f[idx]; bw_est = bandwidth_estimate(Pxx, f)
return {"center_offset_hz": float(cf), "peak_power": float(Pxx[idx]), "bw_est_hz": float(bw_est)}
# ---- Demods (RX only)
def demod_am(iq: "np.ndarray") -> "np.ndarray":
return np.abs(iq)
def demod_fm(iq: "np.ndarray") -> "np.ndarray":
ph = np.unwrap(np.angle(iq)); d = np.diff(ph)
return np.concatenate([[0.0], d])
def deemphasis(audio: "np.ndarray", fs: float, tau: float = 75e-6) -> "np.ndarray":
if not HAS_SCIPY: return audio
b, a = butter(1, 1/(2*math.pi*tau) / (fs/2)); return lfilter(b, a, audio)
def write_wav(path: str, pcm: "np.ndarray", fs: float):
pcm16 = np.clip(pcm / (np.max(np.abs(pcm)) + 1e-12), -1, 1); pcm16 = (pcm16 * 32767.0).astype(np.int16)
with open(path, "wb") as f:
f.write(b"RIFF"); f.write(struct.pack("<I", 36 + pcm16.nbytes)); f.write(b"WAVEfmt ")
f.write(struct.pack("<IHHIIHH", 16, 1, 1, int(fs), int(fs)*2, 2, 16))
f.write(b"data"); f.write(struct.pack("<I", pcm16.nbytes)); f.write(pcm16.tobytes())
# ---- SDR sources (RX only; simulation fallback)
class SDRSource:
def __init__(self, device: str, center: float, samp_rate: float, gain: Optional[float]=None):
self.device, self.center, self.samp_rate, self.gain = device, center, samp_rate, gain
def __iter__(self): raise NotImplementedError
class SDRSim(SDRSource):
def __iter__(self):
if not HAS_NUMPY:
yield None; return
N = 262144; t = np.arange(N)/self.samp_rate
tones = [ (self.center + d, 0.5) for d in (-0.4e6, 0.0, 0.9e6) ]
x = np.zeros(N, dtype=np.complex64)
for f0, a in tones: x += a * np.exp(1j*2*np.pi*(f0-self.center)*t)
x += (np.random.randn(N)+1j*np.random.randn(N))*0.05
for i in range(0, N, 16384): yield x[i:i+16384]
class RTLSource(SDRSource):
def __iter__(self):
if not (HAS_RTL and is_live() and device_allowed(self.device)):
yield from SDRSim(self.device, self.center, self.samp_rate, self.gain); return
sdr = rtlsdr.RtlSdr(); sdr.sample_rate = self.samp_rate; sdr.center_freq = self.center
if self.gain is not None: sdr.gain = self.gain
for _ in range(64):
data = sdr.read_samples(16384); yield np.array(data, dtype=np.complex64)
sdr.close()
class SoapySource(SDRSource):
def __iter__(self):
if not (HAS_SOAPY and is_live() and device_allowed(self.device)):
yield from SDRSim(self.device, self.center, self.samp_rate, self.gain); return
sdr = SoapySDR.Device(dict(driver=self.device.split(":",1)[-1]))
sdr.setSampleRate(SOAPY_SDR_RX, 0, self.samp_rate); sdr.setFrequency(SOAPY_SDR_RX, 0, self.center)
if self.gain is not None:
try: sdr.setGain(SOAPY_SDR_RX, 0, self.gain)
except Exception: pass
rx = sdr.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CF32); sdr.activateStream(rx)
buff = np.empty(16384, dtype=np.complex64)
for _ in range(64):
sr = sdr.readStream(rx, [buff], len(buff))
if sr.ret > 0: yield buff[:sr.ret].copy()
sdr.deactivateStream(rx); sdr.closeStream(rx)
def get_source(device: str, center: float, samp_rate: float, gain: Optional[float]) -> SDRSource:
if device.lower().startswith("rtlsdr"): return RTLSource(device, center, samp_rate, gain)
if device.lower().startswith("soapy"): return SoapySource(device, center, samp_rate, gain)
return SDRSim(device, center, samp_rate, gain)
# ---- Signal operations
def analyze_window(device: str, center: float, samp_rate: float, seconds: int = 2, gain: Optional[float]=None) -> Dict[str,Any]:
src = get_source(device, center, samp_rate, gain)
if not HAS_NUMPY:
res = {"note":"numpy not available; returning band guess only",
"bands":[b.__dict__ for b in bands_covering(center)],
"service_guess": guess_service(center)}
sign_and_print("spectrum-analyze", device, {"center":center, "fs":samp_rate}, res); return res
blocks = []; t0 = time.time()
for blk in src:
if blk is None: break
blocks.append(blk)
if time.time() - t0 > seconds: break
if not blocks:
res = {"error":"no samples"}; sign_and_print("spectrum-analyze", device, {"center":center,"fs":samp_rate}, res); return res
x = np.concatenate(blocks)
peaks = psd_peaks(x, samp_rate)
service = guess_service(center + peaks.get("center_offset_hz", 0.0), peaks.get("bw_est_hz"))
res = {"center_hz": center, "fs_hz": samp_rate, "bands":[b.__dict__ for b in bands_covering(center)],
"peaks": peaks, "service_guess": service}
sign_and_print("spectrum-analyze", device, {"center":center,"fs":samp_rate}, res); return res
def analyze_iq_file(path: str, fs: float, center: float) -> Dict[str,Any]:
if not HAS_NUMPY:
res = {"note":"numpy not available"}; sign_and_print("iq-analyze", path, {"fs":fs,"center":center}, res); return res
iq, _ = load_iq_file(path, "complex64")
if iq is None:
res = {"error":"failed to load IQ"}; sign_and_print("iq-analyze", path, {"fs":fs,"center":center}, res); return res
peaks = psd_peaks(iq, fs)
service = guess_service(center + peaks.get("center_offset_hz", 0.0), peaks.get("bw_est_hz"))
res = {"file": path, "fs_hz": fs, "center_hz": center, "peaks": peaks, "service_guess": service}
sign_and_print("iq-analyze", path, {"fs":fs,"center":center}, res); return res
def demod_fm_from_file(path: str, fs: float, deemph: float, audio_out: Optional[str]) -> Dict[str,Any]:
if not HAS_NUMPY:
res = {"error":"numpy not available"}; sign_and_print("demod-fm", path, {"fs":fs}, res); return res
iq, _ = load_iq_file(path, "complex64")
if iq is None or iq.size < 8192:
res = {"error":"too few samples"}; sign_and_print("demod-fm", path, {"fs":fs}, res); return res
y = demod_fm(iq)
if HAS_SCIPY:
dec = max(1, int(fs // 48000)); y = decimate(y, dec, zero_phase=True); fs_out = fs/dec; y = deemphasis(y, fs_out, deemph)
else:
fs_out = fs
if audio_out: write_wav(audio_out, y.real, fs_out)
res = {"file": path, "audio_out": audio_out or None, "samples": int(y.size), "fs_audio": fs_out}
sign_and_print("demod-fm", path, {"fs":fs}, res); return res
# ============================== FACADE (Gateway) =============================
class InfraGateway:
def __init__(self):
self.openadr = OpenADR(); self.oadr_events = OpenADREvents()
self.iec = IEC61850Browser()
self.snmp = SNMPMonitor(); self.netconf = NETCONFClient(); self.gnmi = GNMIClient()
self.tmf = TMForumAPI()
self.modbus = ModbusReader(); self.opcua = OPCUAReader(); self.mqtt = MQTTMonitor()
# Signal ops (static wrappers to module‑level functions)
def signal_scan(self, device: str, center: float, span: float, step: float, seconds: int, samp_rate: float, gain: Optional[float]) -> Dict[str,Any]:
# Step across requested span; *any* frequency (known/unknown) allowed — classification handles unknowns.
centers = []; nsteps = max(1, int(span//step)); start = center - span/2.0
for i in range(nsteps+1): centers.append(start + i*step)
results = [analyze_window(device, c, samp_rate, seconds, gain) for c in centers]
out = {"scan": results, "center": center, "span": span, "step": step, "fs": samp_rate}
return out
def signal_analyze_iq(self, path: str, fs: float, center: float) -> Dict[str,Any]:
return analyze_iq_file(path, fs, center)
def signal_demod_fm(self, path: str, fs: float, deemph: float, audio_out: Optional[str]) -> Dict[str,Any]:
return demod_fm_from_file(path, fs, deemph, audio_out)
# ============================== CLI =========================================
def main():
gw = InfraGateway()
p = argparse.ArgumentParser(description="UCLS Infra Gateway v3 — universal, read‑only by default")
sub = p.add_subparsers(dest="cmd")
# Demo
sub.add_parser("demo", help="Run simulated flows")
# OpenADR registration flows
c = sub.add_parser("openadr-create"); c.add_argument("--vtn", required=True); c.add_argument("--ven", required=True)
c.add_argument("--profile", default="2.0b"); c.add_argument("--transport", required=True); c.add_argument("--token")
q = sub.add_parser("openadr-query"); q.add_argument("--vtn", required=True); q.add_argument("--ven", required=True); q.add_argument("--token")
x = sub.add_parser("openadr-cancel"); x.add_argument("--vtn", required=True); x.add_argument("--ven", required=True); x.add_argument("--token")
e = sub.add_parser("openadr-events"); e.add_argument("--vtn", required=True); e.add_argument("--ven", required=True); e.add_argument("--token")
# IEC‑61850
ip = sub.add_parser("iec-parse"); ip.add_argument("--path", required=True)
ib = sub.add_parser("iec-browse"); ib.add_argument("--host", required=True); ib.add_argument("--path", required=True)
# TM Forum
t = sub.add_parser("tmf-get"); t.add_argument("--base", required=True); t.add_argument("--api", required=True, choices=list(TMForumAPI.API_MAP.keys())); t.add_argument("--id"); t.add_argument("--token")
# Telecom
s = sub.add_parser("snmp-get"); s.add_argument("--host", required=True); s.add_argument("--oid", required=True)
n = sub.add_parser("netconf-get"); n.add_argument("--host", required=True); n.add_argument("--port", type=int, default=830); n.add_argument("--user", required=True); n.add_argument("--password", required=True); n.add_argument("--filter-subtree")
g = sub.add_parser("gnmi-get"); g.add_argument("--host", required=True); g.add_argument("--path", required=True)
# Edge
m = sub.add_parser("modbus-read"); m.add_argument("--host", required=True); m.add_argument("--addr", type=int, required=True); m.add_argument("--count", type=int, required=True); m.add_argument("--port", type=int, default=502)
o = sub.add_parser("opcua-read"); o.add_argument("--endpoint", required=True); o.add_argument("--node", required=True)
mq = sub.add_parser("mqtt-sub"); mq.add_argument("--broker", required=True); mq.add_argument("--topic", required=True); mq.add_argument("--seconds", type=int, default=10)
# NEW — Signal RX subcommands (RX‑only)
scan = sub.add_parser("signal-scan")
scan.add_argument("--device", default="rtlsdr", help="rtlsdr | soapy:driver=... | sim")
scan.add_argument("--center", type=float, required=True)
scan.add_argument("--span", type=float, default=10e6)
scan.add_argument("--step", type=float, default=2e6)
scan.add_argument("--seconds", type=int, default=2)
scan.add_argument("--samp-rate", type=float, default=2.4e6)
scan.add_argument("--gain", type=float, default=None)
scan.add_argument("--mqtt-broker"); scan.add_argument("--mqtt-topic")
scan.add_argument("--ws-url")
aiq = sub.add_parser("signal-analyze-iq")
aiq.add_argument("--file", required=True)
aiq.add_argument("--samp-rate", type=float, required=True)
aiq.add_argument("--center", type=float, required=True)
aiq.add_argument("--mqtt-broker"); aiq.add_argument("--mqtt-topic"); aiq.add_argument("--ws-url")
dfm = sub.add_parser("signal-demod-fm")
dfm.add_argument("--file", required=True)
dfm.add_argument("--samp-rate", type=float, required=True)
dfm.add_argument("--deemph", type=float, default=75e-6)
dfm.add_argument("--audio-out")
dfm.add_argument("--mqtt-broker"); dfm.add_argument("--mqtt-topic"); dfm.add_argument("--ws-url")
args = p.parse_args()
if args.cmd in (None, "demo"):
# Simulated panorama touching each module
OpenADR.create_party_registration("https://vtn.sim/oadr","VEN_DEMO","2.0b","https://ven.sim/cb",None)
OpenADREvents().poll("https://vtn.sim/oadr","VEN_DEMO",None)
print(json.dumps(IEC61850Path.parse("LD0/MMXU1.TotW.mag.f"), ensure_ascii=False, indent=2))
gw.iec.browse("substation.sim","LD0/MMXU1.TotW.mag.f")
TMForumAPI.get("https://oss.sim/tmf-api","tmf622-product-order","123",None)
gw.snmp.get("router.sim","1.3.6.1.2.1.1.3.0")
gw.netconf.get("router.sim",830,"user","pass","<interfaces/>")
gw.gnmi.get("router.sim","/interfaces/interface[name=xe-0/0/0]/state")
gw.modbus.read_holding("plc.sim",0,4)
gw.opcua.read_node("opc.tcp://opcua.sim:4840","ns=2;i=10853")
gw.mqtt.subscribe("mqtt.sim","ucls/#",seconds=2)
# Signal scan (sim)
out_scan = gw.signal_scan("sim", 100e6, 10e6, 2e6, 1, 2.4e6, None)
print(json.dumps(out_scan, ensure_ascii=False, indent=2))
return
# --- OpenADR
if args.cmd == "openadr-create":
OpenADR.create_party_registration(args.vtn, args.ven, args.profile, args.transport, args.token); return
if args.cmd == "openadr-query":
OpenADR.query_registration(args.vtn, args.ven, args.token); return
if args.cmd == "openadr-cancel":
OpenADR.cancel_party_registration(args.vtn, args.ven, args.token); return
if args.cmd == "openadr-events":
OpenADREvents().poll(args.vtn, args.ven, args.token); return
# --- IEC‑61850
if args.cmd == "iec-parse":
res = IEC61850Path.parse(args.path); sign_and_print("iec-parse","local",{"path":args.path},res); print(json.dumps(res,ensure_ascii=False,indent=2)); return
if args.cmd == "iec-browse":
gw.iec.browse(args.host, args.path); return
# --- TMF
if args.cmd == "tmf-get":
TMForumAPI.get(args.base, args.api, args.id, args.token); return
# --- Telecom
if args.cmd == "snmp-get":
gw.snmp.get(args.host, args.oid); return
if args.cmd == "netconf-get":
gw.netconf.get(args.host, args.port, args.user, args.password, args.filter_subtree); return
if args.cmd == "gnmi-get":
gw.gnmi.get(args.host, args.path); return
# --- Edge
if args.cmd == "modbus-read":
gw.modbus.read_holding(args.host, args.addr, args.count, args.port); return
if args.cmd == "opcua-read":
gw.opcua.read_node(args.endpoint, args.node); return
if args.cmd == "mqtt-sub":
gw.mqtt.subscribe(args.broker, args.topic, args.seconds); return
# --- SIGNAL RX (NEW) + Publishers
if args.cmd == "signal-scan":
res = gw.signal_scan(args.device, args.center, args.span, args.step, args.seconds, args.samp_rate, args.gain)
# publish if requested
if getattr(args, "mqtt_broker", None) or getattr(args, "ws_url", None):
publish_mqtt(res, args.mqtt_broker, args.mqtt_topic)
publish_ws(res, args.ws_url)
print(json.dumps(res, ensure_ascii=False, indent=2)); return
if args.cmd == "signal-analyze-iq":
res = gw.signal_analyze_iq(args.file, args.samp_rate, args.center)
if getattr(args, "mqtt_broker", None) or getattr(args, "ws_url", None):
publish_mqtt(res, args.mqtt_broker, args.mqtt_topic)
publish_ws(res, args.ws_url)
print(json.dumps(res, ensure_ascii=False, indent=2)); return
if args.cmd == "signal-demod-fm":
res = gw.signal_demod_fm(args.file, args.samp_rate, args.deemph, args.audio_out)
if getattr(args, "mqtt_broker", None) or getattr(args, "ws_url", None):
publish_mqtt(res, args.mqtt_broker, args.mqtt_topic)
publish_ws(res, args.ws_url)
print(json.dumps(res, ensure_ascii=False, indent=2)); return
if __name__ == "__main__":
main()
Notes
- RX‑only by construction; no transmit paths implemented.
- “All frequencies known and unknown”: the scan accepts any
--center/--span; classifier returns a best guess or Unspecified/Unknown (region‑defined) for undefined ranges. - MQTT/WS publishers are optional flags per command; they respect LIVE mode and allowlists.
Key terms in plain language
Open a term for a concise explanation of language used on this page.
Fiber Internet
Internet delivered through strands of glass using light. Fiber commonly supports high capacity, low latency, and strong upload performance, but availability must be confirmed for the exact address.
API
An application programming interface is a defined way for software systems to exchange data or request functions from one another.
Broadband
A general term for always-on, high-speed Internet access. Broadband can be delivered over fiber, cable, DSL, fixed wireless, cellular, or satellite networks.
Bandwidth
The amount of data a connection can carry in a given time, usually measured in Mbps or Gbps. More bandwidth supports more users, devices, and simultaneous applications.
Latency
The time it takes data to travel between two points. Lower latency improves voice, video meetings, cloud applications, gaming, and other real-time services.
Dedicated Internet Access (DIA)
A business-grade Internet connection with capacity dedicated to the customer rather than shared in the same way as typical consumer broadband. It often includes symmetrical speeds and an SLA.