#!/usr/bin/env python3 """Minimal NMOS controller: IS-04 read-out and IS-05 switching. For a stand that has no NMOS controller of its own. Needs Python 3 and nothing else; `avahi-browse` is used for mDNS discovery. Works peer-to-peer, without a registry: nodes are found over mDNS, switching is a PATCH of the receiver's staged endpoint. nmosctl.py discover nmosctl.py list nmosctl.py sdp nmosctl.py enable|disable nmosctl.py connect nmosctl.py disconnect A typical sequence for feeding a sender into an ST 2110 input: enable the sender, read its transport file, paste that text into the receiver whose addressing is set to "Transport file (SDP)". nmosctl.py enable 10.0.0.90:8090 nmosctl.py sdp 10.0.0.90:8090 Two traps worth knowing. Until a sender is enabled, `/transportfile` answers 404: the group address only exists once the sender is activated. And that address usually changes on every activation, so read it anew each time instead of keeping it in your notes. """ import json import subprocess import sys import urllib.error import urllib.request NODE_API = "/x-nmos/node/v1.3" CONN_API = "/x-nmos/connection/v1.1/single" TIMEOUT = 10 def request(url, method="GET", body=None, content_type="application/json"): data = None headers = {} if body is not None: data = body if isinstance(body, bytes) else json.dumps(body).encode() headers["Content-Type"] = content_type req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: raw = resp.read() except urllib.error.HTTPError as err: raw = err.read() sys.stderr.write(f"{method} {url} -> {err.code}\n{raw.decode(errors='replace')}\n") raise SystemExit(1) text = raw.decode(errors="replace") try: return json.loads(text) except json.JSONDecodeError: return text def node_url(host, path): return f"http://{host}{path}" def discover(): """Nodes in the network: only those announcing _nmos-node._tcp.""" out = subprocess.run( ["avahi-browse", "-rpt", "_nmos-node._tcp"], capture_output=True, text=True, timeout=30, ).stdout seen = set() for line in out.splitlines(): parts = line.split(";") # Resolved record: =;iface;IPv4;name;type;domain;hostname;addr;port;txt if len(parts) > 8 and parts[0] == "=" and parts[2] == "IPv4": entry = (parts[7], parts[8], parts[3]) if entry[:2] not in seen: seen.add(entry[:2]) print(f"{entry[0]}:{entry[1]}\t{entry[2]}") if not seen: print("no nodes found (check that avahi-daemon is running)") def list_node(host): self_info = request(node_url(host, f"{NODE_API}/self")) clocks = ", ".join( f"{c.get('name')}={c.get('ref_type')}" + (f" gm={c.get('gmid')} locked={c.get('locked')}" if c.get("ref_type") == "ptp" else "") for c in self_info.get("clocks", []) ) print(f"node: {self_info.get('label')} [{self_info.get('id')}]") print(f" {self_info.get('description')}") print(f" clocks: {clocks or '-'}") devices = {d["id"]: d for d in request(node_url(host, f"{NODE_API}/devices"))} flows = {f["id"]: f for f in request(node_url(host, f"{NODE_API}/flows"))} print("\nsenders:") for s in request(node_url(host, f"{NODE_API}/senders")): dev = devices.get(s.get("device_id"), {}).get("label", "?") flow = flows.get(s.get("flow_id"), {}) active = request(node_url(host, f"{CONN_API}/senders/{s['id']}/active")) params = active.get("transport_params", [{}]) dst = ", ".join( f"{p.get('destination_ip')}:{p.get('destination_port')}" for p in params ) state = "ON " if active.get("master_enable") else "off" print(f" [{state}] {s['id']} {dev}/{s['label']} {flow.get('media_type','?')} -> {dst}") print("\nreceivers:") for r in request(node_url(host, f"{NODE_API}/receivers")): dev = devices.get(r.get("device_id"), {}).get("label", "?") active = request(node_url(host, f"{CONN_API}/receivers/{r['id']}/active")) params = active.get("transport_params", [{}]) src = ", ".join( f"{p.get('multicast_ip')}:{p.get('destination_port')}" for p in params ) state = "ON " if active.get("master_enable") else "off" caps = ",".join(r.get("caps", {}).get("media_types", [])) print(f" [{state}] {r['id']} {dev}/{r['label']} {caps} <- {src}") def sdp(host, sender_id): print(request(node_url(host, f"{CONN_API}/senders/{sender_id}/transportfile"))) def set_sender(host, sender_id, enable): patch = { "master_enable": enable, "activation": {"mode": "activate_immediate"}, } result = request( node_url(host, f"{CONN_API}/senders/{sender_id}/staged"), "PATCH", patch ) print(json.dumps(result, indent=2)) def connect(src_host, sender_id, dst_host, receiver_id): """The classic splice: the sender's SDP goes into the receiver's staged.""" set_sender(src_host, sender_id, True) transport_file = request( node_url(src_host, f"{CONN_API}/senders/{sender_id}/transportfile") ) if not isinstance(transport_file, str): sys.stderr.write(f"the sender returned no SDP: {transport_file}\n") raise SystemExit(1) patch = { "sender_id": sender_id, "master_enable": True, "activation": {"mode": "activate_immediate"}, "transport_file": {"type": "application/sdp", "data": transport_file}, } result = request( node_url(dst_host, f"{CONN_API}/receivers/{receiver_id}/staged"), "PATCH", patch ) print(json.dumps(result, indent=2)) def disconnect(host, receiver_id): patch = { "sender_id": None, "master_enable": False, "activation": {"mode": "activate_immediate"}, } result = request( node_url(host, f"{CONN_API}/receivers/{receiver_id}/staged"), "PATCH", patch ) print(json.dumps(result, indent=2)) COMMANDS = { "discover": (0, lambda a: discover()), "list": (1, lambda a: list_node(a[0])), "sdp": (2, lambda a: sdp(a[0], a[1])), "enable": (2, lambda a: set_sender(a[0], a[1], True)), "disable": (2, lambda a: set_sender(a[0], a[1], False)), "connect": (4, lambda a: connect(a[0], a[1], a[2], a[3])), "disconnect": (2, lambda a: disconnect(a[0], a[1])), } def main(): if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS: print(__doc__) raise SystemExit(2) argc, handler = COMMANDS[sys.argv[1]] args = sys.argv[2:] if len(args) != argc: print(__doc__) raise SystemExit(2) handler(args) if __name__ == "__main__": main()