Overview
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ucls_infra_gateway_v2.py
-------------------------------------------------------------------------------------------------
Universal, SAFE-BY-DEFAULT interoperability gateway for:
• OpenADR 2.0b / IEEE 2030.5 (SEP 2.0) — VEN registration flows (dry-run by default)
• IEC-61850 — path helpers (LN/DO/DA traversal) + read-only browse
• TM Forum Open APIs (OSS/BSS) — read-only GETs (TMF622/632/633/639/641)
• Telecom — SNMP, NETCONF, gNMI (read-only; stubs if libs absent)
• Grid telemetry — DNP3 read (stub), OpenADR event poll (read-only)
• Edge devices — Modbus/TCP read, OPC UA read, MQTT subscribe (safe)
• Powerline communication (PLC) data — linguistics codec (encode/decode) + RX monitor (sim)
HARD GUARANTEES
- LIVE mode OFF by default (UCLS_LIVE=0) → dry-run/simulation only.
- Host allowlist REQUIRED for any live I/O: UCLS_WHITELIST="host1,1.2.3.4,..."
- Strict READ-ONLY for grid/telecom/OSS except OpenADR *registration*:
• Registration payloads are constructed and logged. Sending requires LIVE+allowlist.
- Signed JSON audit for every operation to stdout (HMAC-SHA256).
USAGE (CLI)
python ucls_infra_gateway_v2.py demo
# OpenADR VEN registration (DRY-RUN unless LIVE=1 and host allow-listed)
UCLS_LIVE=1 UCLS_WHITELIST="vtn.example.com" \
python ucls_infra_gateway_v2.py openadr-create --vtn https://vtn.example.com/oadr \
--ven VEN_123 --profile 2.0b --transport https://ven.example.com/callback --token "***"
# IEC-61850 path helpers
python ucls_infra_gateway_v2.py iec-parse --path "LD0/MMXU1.TotW.mag.f"
python ucls_infra_gateway_v2.py iec-browse --host substation.sim --path "LD0/MMXU1.TotW.mag.f"
# TM Forum Open APIs (read-only)
UCLS_LIVE=1 UCLS_WHITELIST="oss.example.com" \
python ucls_infra_gateway_v2.py tmf-get --base https://oss.example.com/tmf-api \
--api tmf622-product-order --id 123 --token "***"
# Modbus/OPC UA (read-only; sim if libs missing)
python ucls_infra_gateway_v2.py modbus-read --host plc.sim --addr 0 --count 4
python ucls_infra_gateway_v2.py opcua-read --endpoint opc.tcp://opcua.sim:4840 --node "ns=2;i=10853"
# MQTT monitor (subscribe; sim unless LIVE and paho-mqtt present; prints messages to audit)
python ucls_infra_gateway_v2.py mqtt-sub --broker mqtt.sim --topic "ucls/#" --seconds 5
DISCLAIMER
You are responsible for authorization and compliance in your environment.
Extending to CONTROL operations is out of scope in this tool and must undergo
rigorous safety review and testing.
"""
from __future__ import annotations
import argparse, json, os, re, socket, time, hmac, hashlib
from dataclasses import dataclass, asdict
from typing import Any, Dict, List, Optional, Tuple
# --------------------------- SAFE RUNTIME CONTROLS ---------------------------
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:
# Accept host or URL; extract hostname when URL is provided
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)
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)[:240] + "…") if len(str(result))>260 else str(result))
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 (GRACEFUL) ------------------------
# HTTP client for OpenADR/TM Forum (requests optional: live calls only when available)
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 (stub in this build)
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
try:
import paho.mqtt.client as mqtt
HAS_PAHO = True
except Exception:
HAS_PAHO = False
# --------------------------- OPENADR: VEN REGISTRATION -----------------------
class OpenADR:
"""
OpenADR 2.0b VEN registration flows — DRY-RUN by default.
Endpoints vary; many VTNs expose XML over HTTP(S).
We construct canonical payloads; only send if LIVE and host allow-listed.
"""
@staticmethod
def create_party_registration(vtn: str, ven_id: str, profile: str, transport_address: str, token: str) -> Dict[str,Any]:
payload = {
"oadrCreatePartyRegistration": {
"venID": ven_id, "oadrProfileName": profile,
"oadrTransportAddress": transport_address,
"oadrTransportName": "simpleHttp",
"oadrXmlSignature": False
}
}
action = "openadr-create"
if is_live() and is_host_allowed(vtn) and HAS_REQUESTS:
r = requests.post(f"{vtn}/CreatePartyRegistration",
json=payload, headers={"Authorization": f"Bearer {token}"}, timeout=15)
result = {"status": r.status_code, "body": (r.text[:200]+"…")}
else:
result = {"dry_run": True, "target": f"{vtn}/CreatePartyRegistration", "payload": payload}
sign_and_print(action, vtn, {"ven": ven_id, "profile": profile}, result)
return result
@staticmethod
def query_registration(vtn: str, ven_id: str, token: str) -> Dict[str,Any]:
payload = {"oadrQueryRegistration": {"venID": ven_id}}
action = "openadr-query"
if is_live() and is_host_allowed(vtn) and HAS_REQUESTS:
r = requests.post(f"{vtn}/QueryRegistration",
json=payload, headers={"Authorization": f"Bearer {token}"}, timeout=15)
result = {"status": r.status_code, "body": (r.text[:200]+"…")}
else:
result = {"dry_run": True, "target": f"{vtn}/QueryRegistration", "payload": payload}
sign_and_print(action, vtn, {"ven": ven_id}, result)
return result
@staticmethod
def cancel_party_registration(vtn: str, ven_id: str, token: str) -> Dict[str,Any]:
payload = {"oadrCancelPartyRegistration": {"venID": ven_id}}
action = "openadr-cancel"
if is_live() and is_host_allowed(vtn) and HAS_REQUESTS:
r = requests.post(f"{vtn}/CancelPartyRegistration",
json=payload, headers={"Authorization": f"Bearer {token}"}, timeout=15)
result = {"status": r.status_code, "body": (r.text[:200]+"…")}
else:
result = {"dry_run": True, "target": f"{vtn}/CancelPartyRegistration", "payload": payload}
sign_and_print(action, vtn, {"ven": ven_id}, result)
return result
# --------------------------- IEC-61850: PATH HELPERS + BROWSE ----------------
class IEC61850Path:
"""
Helpers for IEC-61850 logical path parsing: LD/LN.DO.DA[.FC][.Sub]
Example: LD0/MMXU1.TotW.mag.f
Returns dict: {ld, ln, do_path:[...], da_path:[...]}
"""
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/LNName.DO.DA[.sub]'.")
ld = m.group("ld")
ln = m.group("ln")
rest = m.group("rest")
parts = rest.split(".")
# Heuristic: DO may be 1 element (e.g., TotW), DA may be the rest (mag.f)
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:
tail = ".".join([parsed["do"]] + parsed["da"])
return f"{parsed['ld']}/{parsed['ln']}.{tail}"
class IEC61850Browser:
"""
Read-only browse: in real deployments, bind to MMS and read DA values.
Here we simulate unless LIVE+allowlist and a MMS client is injected.
"""
def __init__(self):
self._mms = None # placeholder for real MMS client
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.")
# Here you would invoke MMS read using parsed components (omitted).
result = {"path": parsed, "value": None, "quality":"unknown", "ts": time.time()}
sign_and_print("iec-browse", host, {"path": path}, result)
return result
# --------------------------- TM Forum Open APIs (READ-ONLY) ------------------
class TMForumAPI:
"""
Read-only GET wrapper for TMF APIs. Examples:
• TMF622 Product Ordering → /tmf-api/productOrderingManagement/v4/productOrder/{id}
• TMF632 Party Management → /tmf-api/partyManagement/v5/party/{id}
• TMF633 Service Catalog → /tmf-api/serviceCatalogManagement/v4/serviceSpecification/{id}
• TMF639 Service Inventory → /tmf-api/serviceInventoryManagement/v4/service/{id}
• TMF641 Service Ordering → /tmf-api/serviceOrderingManagement/v4/serviceOrder/{id}
"""
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 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}"
# If id is provided, append; else list first page
url = f"{url}/{_id}" if _id else f"{url}?limit=10"
action = "tmf-get"
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(action, url, {"api": api, "id": _id or "list"}, result)
return result
@staticmethod
def _suffix(base: str) -> str:
# Help users who pass /tmf-api in base
return "/tmf-api" if base.rstrip("/").endswith("/tmf-api") else ""
# --------------------------- TELECOM / GRID ADAPTERS (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
# Live SNMPv2c public for demo (customize to v3 in real env)
from pysnmp.hlapi import SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity, getCmd
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")
# Real gNMI get would go here (TLS gRPC)
result = {"host": host, "path": path, "value": {}}
sign_and_print("gnmi-get", host, {"path": path}, result); return result
class DNP3Reader:
def read_analogs(self, host: str, indices: List[int]) -> Dict[str,Any]:
if not is_live():
data = {f"AI{idx}": round(idx*1.234,3) for idx in indices}
else:
if not is_host_allowed(host): raise PermissionError("Host not allowed")
data = {f"AI{idx}": None for idx in indices} # stub
sign_and_print("dnp3-read", host, {"indices": indices}, data)
return data
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
# --------------------------- EDGE DEVICES (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 = 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
# --------------------------- POWERLINE COMM (data & linguistics) -------------
class PLCCodec:
"""
Linguistics/data codec for powerline frames (transport-agnostic).
We encode structured messages as UTF-8 JSON or CBOR (if cbor2 installed).
"""
def __init__(self):
try:
import cbor2 # type: ignore
self.cbor = True
except Exception:
self.cbor = False
def encode(self, obj: Dict[str,Any]) -> bytes:
if self.cbor:
import cbor2
return cbor2.dumps(obj)
return json.dumps(obj, ensure_ascii=False).encode("utf-8")
def decode(self, data: bytes) -> Dict[str,Any]:
# Try CBOR first
if self.cbor and data and data[0] in (0xA0, 0xA1, 0xBF, 0x80, 0x81): # rough CBOR map/array lead bytes
import cbor2
return cbor2.loads(data)
try:
return json.loads(data.decode("utf-8"))
except Exception:
return {"raw_hex": data.hex(), "note": "unknown encoding; returned raw bytes hex"}
class PLCMonitor:
"""
Powerline RX monitor (simulation): read frames from a file/stream and decode with PLCCodec.
Real PHY/MAC integration (HomePlug/FSK/G3) is out-of-scope; keep this as logical layer.
"""
def __init__(self):
self.codec = PLCCodec()
def receive(self, source: str) -> Dict[str,Any]:
if not is_live():
# Simulated frame
frame = self.codec.encode({"type":"linguistic","text":"salvē, mundi","lang":"la"})
result = {"source":source,"frames":[self.codec.decode(frame)]}
sign_and_print("plc-rx-sim", source, {}, result); return result
# In live, source could be a pipe/file. We still treat as READ-ONLY monitor.
try:
with open(source,"rb") as f:
data = f.read()
result = {"source":source,"decoded": self.codec.decode(data)}
except Exception as e:
result = {"source":source,"error": str(e)}
sign_and_print("plc-rx", source, {}, result); return result
# --------------------------- GATEWAY (FACADE) --------------------------------
class InfraGateway:
def __init__(self):
# Grid / OpenADR
self.openadr = OpenADR()
self.oadr_events = OpenADREvents()
self.iec = IEC61850Browser()
# Telecom
self.snmp = SNMPMonitor()
self.netconf = NETCONFClient()
self.gnmi = GNMIClient()
# OSS/BSS
self.tmf = TMForumAPI()
# Edge
self.modbus = ModbusReader()
self.opcua = OPCUAReader()
self.mqtt = MQTTMonitor()
# Powerline
self.plc = PLCMonitor()
# --------------------------- CLI ---------------------------------------------
def main():
gw = InfraGateway()
p = argparse.ArgumentParser(description="UCLS Infra Gateway v2 — universal, read-only by default")
sub = p.add_subparsers(dest="cmd")
# Demo
sub.add_parser("demo", help="Run simulated flows to show structure")
# OpenADR registration flows
c = sub.add_parser("openadr-create", help="OpenADR VEN CreatePartyRegistration (DRY-RUN unless LIVE)")
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", required=False)
q = sub.add_parser("openadr-query", help="OpenADR VEN QueryRegistration (DRY-RUN unless LIVE)")
q.add_argument("--vtn", required=True); q.add_argument("--ven", required=True)
q.add_argument("--token", required=False)
x = sub.add_parser("openadr-cancel", help="OpenADR VEN CancelPartyRegistration (DRY-RUN unless LIVE)")
x.add_argument("--vtn", required=True); x.add_argument("--ven", required=True)
x.add_argument("--token", required=False)
e = sub.add_parser("openadr-events", help="OpenADR events poll (read-only)")
e.add_argument("--vtn", required=True); e.add_argument("--ven", required=True); e.add_argument("--token")
# IEC-61850 helpers
ip = sub.add_parser("iec-parse", help="Parse IEC-61850 path LD/LN.DO.DA[.sub]")
ip.add_argument("--path", required=True)
ib = sub.add_parser("iec-browse", help="Read-only browse DA value at path")
ib.add_argument("--host", required=True); ib.add_argument("--path", required=True)
# TM Forum
t = sub.add_parser("tmf-get", help="TM Forum Open API GET (read-only)")
t.add_argument("--base", required=True, help="Base, e.g., https://oss.example.com/tmf-api")
t.add_argument("--api", required=True, choices=list(TMForumAPI.API_MAP.keys()))
t.add_argument("--id", required=False); t.add_argument("--token", required=False)
# Telecom
s = sub.add_parser("snmp-get", help="SNMP GET (read-only)")
s.add_argument("--host", required=True); s.add_argument("--oid", required=True)
n = sub.add_parser("netconf-get", help="NETCONF get (read-only)")
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", required=False)
g = sub.add_parser("gnmi-get", help="gNMI get (read-only, stub unless client installed)")
g.add_argument("--host", required=True); g.add_argument("--path", required=True)
# Edge
m = sub.add_parser("modbus-read", help="Modbus/TCP read holding registers (read-only)")
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", help="OPC UA read (read-only)")
o.add_argument("--endpoint", required=True); o.add_argument("--node", required=True)
mq = sub.add_parser("mqtt-sub", help="MQTT subscribe (read-only monitor)")
mq.add_argument("--broker", required=True); mq.add_argument("--topic", required=True)
mq.add_argument("--seconds", type=int, default=10)
# PLC monitor
pr = sub.add_parser("plc-rx", help="Powerline monitor: decode frames from file/pipe (read-only)")
pr.add_argument("--source", required=False, default="simulated")
args = p.parse_args()
if args.cmd in (None, "demo"):
# Simulated panorama
OpenADR.create_party_registration("https://vtn.sim/oadr","VEN_DEMO","2.0b","https://ven.sim/cb","demo")
gw.oadr_events.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)
gw.plc.receive("sim")
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":
gw.oadr_events.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
# PLC
if args.cmd == "plc-rx":
gw.plc.receive(args.source); return
if __name__ == "__main__":
main()
What’s new in this v2 gateway
- OpenADR VEN registration flows
openadr-create,openadr-query,openadr-cancel— dry-run by default; actual POSTs only whenUCLS_LIVE=1, host is allow-listed, andrequestsis available. - IEC-61850 path helpers
iec-parseto decomposeLD/LN.DO.DA[.sub], andiec-browseto read a DA value (simulated/live depending on safety flags). - TM Forum Open APIs
tmf-getperforms safe, read-only GETs for TMF622/632/633/639/641, with token support. - Edge integrations(read-only): Modbus/TCP, OPC UA, and MQTT subscriber; all respect LIVE/allowlist and degrade to simulation if libs are missing.
- Powerline communicationa transport-agnostic PLC linguistics codec (JSON/CBOR) and a receive monitor (
plc-rx) that decodes frames; real PHY/MAC is intentionally out-of-scope for safety.
Plug this into your existing UCLS ecosystem to keep everything interoperable, auditable, and non-destructive by default.
ucls_infra_gateway.py – SolveForce Communications
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.
Artificial Intelligence (AI)
Software designed to perform tasks involving prediction, classification, generation, reasoning, or decision support. Business use still requires clear data, governance, security, and human accountability.
Cybersecurity
The practices and controls used to protect identities, devices, networks, applications, and data from unauthorized access, disruption, or manipulation.
Zero Trust
A security model that does not automatically trust a user or device because of its location. Access is continuously verified and limited to what is necessary.
SASE
Secure Access Service Edge combines networking and security capabilities in a cloud-delivered architecture so users and locations can receive consistent policy wherever they connect.