#!/usr/bin/env python3 # # IPSC3 Project by Matt Perkins, VK2FLY # IPSC3view version 1.0.32 # Date: 2026-06-14 # Copyright (C) 2026 Matt Perkins, VK2FLY. # Licensed under the GNU General Public License v3.0 or later. import argparse import ctypes import importlib import json import os import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from datetime import datetime from typing import Any IPSC3VIEW_VERSION = "1.0.32" # Compatibility marker for older auto-updaters that used the IPSC3 API # version as the expected client version during the 1.0.30 transition. # IPSC3VIEW_VERSION = "1.0.30" PROCESS_NAME = "IPSC3View" DEFAULT_API_URL = "https://ipsc3.vkdmr.com/api/ipsc3view" DOWNLOADS_URL = "https://ipsc3.vkdmr.com/downloads" DOWNLOAD_PATH = "/downloads/ipsc3view" DEFAULT_REFRESH_SECONDS = 2.0 DEFAULT_TIMEOUT_SECONDS = 5.0 DEFAULT_LIMIT = 30 ACTIVE_TALKGROUP_WINDOW_SECONDS = 15 TABS = ("Overview", "Last Heard", "Roaming", "Repeaters", "Hotspots", "Trunks", "Talkgroups") SCROLLABLE_TABS = {"Roaming", "Talkgroups", "Repeaters", "Hotspots"} curses: Any | None = None class VersionMismatchError(RuntimeError): def __init__( self, message: str, api_url: str | None = None, server_version: str | None = None, ): super().__init__(message) self.api_url = api_url self.server_version = server_version class SelfUpdateError(RuntimeError): pass class CursesDependencyError(RuntimeError): pass def set_process_name(name: str = PROCESS_NAME) -> None: if os.name != "posix": return try: libc = ctypes.CDLL(None) libc.prctl( ctypes.c_int(15), # PR_SET_NAME ctypes.c_char_p(name.encode("utf-8")[:15]), ctypes.c_ulong(0), ctypes.c_ulong(0), ctypes.c_ulong(0), ) except Exception: pass def set_terminal_title(title: str = PROCESS_NAME) -> None: if not sys.stdout.isatty(): return try: sys.stdout.write(f"\033]0;{title}\007") sys.stdout.flush() except Exception: pass def ensure_curses() -> Any: global curses if curses is not None: return curses try: curses = importlib.import_module("curses") return curses except ImportError as exc: if os.name != "nt": raise CursesDependencyError( "ipsc3view live dashboard requires Python curses support. " "Install the curses package for this Python environment." ) from exc if not sys.stdin.isatty(): raise CursesDependencyError( "ipsc3view live dashboard on Windows requires windows-curses.\n" f"Install it with: {sys.executable} -m pip install windows-curses" ) from exc print("ipsc3view live dashboard on Windows requires windows-curses.", file=sys.stderr) try: answer = input("Install windows-curses now? [y/N] ") except EOFError as input_exc: raise CursesDependencyError( f"Install it with: {sys.executable} -m pip install windows-curses" ) from input_exc if answer.strip().lower() not in ("y", "yes"): raise CursesDependencyError( f"Install it with: {sys.executable} -m pip install windows-curses" ) from exc command = [sys.executable, "-m", "pip", "install", "windows-curses"] result = subprocess.run(command) if result.returncode != 0: raise CursesDependencyError( "windows-curses install failed. " f"Run manually: {sys.executable} -m pip install windows-curses" ) from exc importlib.invalidate_caches() try: curses = importlib.import_module("curses") return curses except ImportError as second_exc: raise CursesDependencyError( "windows-curses installed, but curses still cannot be imported. " "Restart the terminal or check the active Python environment." ) from second_exc def download_url_for_api(api_url: str | None) -> str: if api_url: parsed = urllib.parse.urlparse(api_url) if parsed.scheme in ("http", "https") and parsed.netloc: return urllib.parse.urlunparse( (parsed.scheme, parsed.netloc, DOWNLOAD_PATH, "", "", "") ) return f"{DOWNLOADS_URL}/ipsc3view" def upgrade_instructions(api_url: str | None = None) -> str: download_url = download_url_for_api(api_url) return ( "Upgrade instructions:\n" f" curl -L -o ipsc3view {download_url}\n" " chmod +x ipsc3view\n" " ./ipsc3view\n" "\n" "If ipsc3view is installed in your PATH, replace it after download:\n" " sudo install -m 0755 ipsc3view /usr/local/bin/ipsc3view" ) def local_tz_label() -> str: return datetime.now().astimezone().tzname() or "local" def api_url_with_limit(api_url: str, limit: int) -> str: parsed = urllib.parse.urlparse(api_url) query = dict(urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)) query["limit"] = str(limit) query["clientVersion"] = IPSC3VIEW_VERSION return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query))) def validate_api_version(data: dict[str, Any], api_url: str) -> None: api_version = data.get("version") if not api_version: raise VersionMismatchError( f"ipsc3view: API version missing from {api_url}.\n\n" f"{upgrade_instructions(api_url)}", api_url=api_url, ) viewer_version = data.get("ipsc3viewVersion") if viewer_version and str(viewer_version) != IPSC3VIEW_VERSION: raise VersionMismatchError( f"ipsc3view: viewer API version mismatch. Client {IPSC3VIEW_VERSION} requires " f"IPSC3view API {IPSC3VIEW_VERSION}, but {api_url} reports {viewer_version}.\n\n" f"{upgrade_instructions(api_url)}", api_url=api_url, server_version=str(viewer_version), ) def current_script_path() -> str | None: invoked = sys.argv[0] if not invoked: return None has_path = os.path.sep in invoked or (os.path.altsep is not None and os.path.altsep in invoked) candidate = invoked if has_path else shutil.which(invoked) if not candidate: return None return os.path.realpath(candidate) def validate_update_script(script: bytes, expected_version: str) -> None: if len(script) > 1_000_000: raise SelfUpdateError("downloaded ipsc3view is unexpectedly large") if not script.startswith(b"#!/usr/bin/env python3"): raise SelfUpdateError("downloaded file does not look like ipsc3view") version_line = f'IPSC3VIEW_VERSION = "{expected_version}"'.encode("utf-8") if version_line not in script: raise SelfUpdateError( f"downloaded ipsc3view does not contain expected version {expected_version}" ) def self_update(api_url: str, target_version: str, timeout: float) -> str: script_path = current_script_path() if not script_path or not os.path.isfile(script_path): raise SelfUpdateError("could not determine the current ipsc3view script path") script_dir = os.path.dirname(script_path) or "." if not os.access(script_dir, os.W_OK): raise SelfUpdateError( f"current install directory is not writable: {script_dir}\n\n" "Download manually, then install with:\n" f" sudo install -m 0755 ipsc3view {script_path}" ) download_url = download_url_for_api(api_url) request = urllib.request.Request( download_url, headers={ "Accept": "text/x-python,text/plain,*/*", "User-Agent": f"ipsc3view/{IPSC3VIEW_VERSION}", }, ) with urllib.request.urlopen(request, timeout=timeout) as response: script = response.read(1_000_001) validate_update_script(script, target_version) mode = os.stat(script_path).st_mode & 0o777 if mode == 0: mode = 0o755 tmp_path = "" try: fd, tmp_path = tempfile.mkstemp(prefix=".ipsc3view-", dir=script_dir) with os.fdopen(fd, "wb") as tmp: tmp.write(script) tmp.flush() os.fsync(tmp.fileno()) os.chmod(tmp_path, mode) os.replace(tmp_path, script_path) except OSError as exc: raise SelfUpdateError(f"failed to replace {script_path}: {exc}") from exc finally: if tmp_path and os.path.exists(tmp_path): try: os.unlink(tmp_path) except OSError: pass return script_path def handle_version_mismatch(exc: VersionMismatchError, args: argparse.Namespace) -> int: print(str(exc), file=sys.stderr) if not exc.server_version: return 2 if not sys.stdin.isatty(): return 2 try: answer = input( f"\nDownload and replace ipsc3view with {exc.server_version} now? [y/N] " ) except EOFError: return 2 if answer.strip().lower() not in ("y", "yes"): return 2 api_url = exc.api_url or args.api_url try: path = self_update(api_url, exc.server_version, args.timeout) except (SelfUpdateError, urllib.error.URLError, TimeoutError, OSError) as update_exc: print(f"ipsc3view: automatic update failed: {update_exc}", file=sys.stderr) print(f"\n{upgrade_instructions(api_url)}", file=sys.stderr) return 2 print(f"ipsc3view updated to {exc.server_version}: {path}", file=sys.stderr) print("Run ipsc3view again to use the new version.", file=sys.stderr) return 0 def fetch_json(api_url: str, timeout: float) -> dict[str, Any]: request = urllib.request.Request( api_url, headers={"Accept": "application/json", "User-Agent": f"ipsc3view/{IPSC3VIEW_VERSION}"}, ) with urllib.request.urlopen(request, timeout=timeout) as response: charset = response.headers.get_content_charset() or "utf-8" data = json.loads(response.read().decode(charset)) validate_api_version(data, api_url) return data def as_text(value: Any, fallback: str = "-") -> str: if value is None or value == "": return fallback return str(value) def as_int(value: Any, fallback: int = 0) -> int: try: if value is None or value == "": return fallback return int(value) except (TypeError, ValueError): return fallback def fmt_epoch(value: Any) -> str: epoch = as_int(value, 0) if epoch <= 0: return "-" return datetime.fromtimestamp(epoch).strftime("%d/%m %H:%M:%S") def fmt_time(value: Any) -> str: epoch = as_int(value, 0) if epoch <= 0: return "-" return datetime.fromtimestamp(epoch).strftime("%H:%M:%S") def fmt_duration(value: Any) -> str: seconds = as_int(value, -1) if seconds < 0: return "-" if seconds < 3: return "Kerchunk" minutes, sec = divmod(seconds, 60) hours, minutes = divmod(minutes, 60) if hours: return f"{hours}:{minutes:02d}:{sec:02d}" return f"{minutes}:{sec:02d}" def fmt_packets(value: Any) -> str: count = as_int(value, 0) if count >= 1_000_000_000: return f"{count / 1_000_000_000:.1f}G".replace(".0G", "G") if count >= 1_000_000: return f"{count / 1_000_000:.1f}M".replace(".0M", "M") if count >= 1_000: return f"{count / 1_000:.1f}K".replace(".0K", "K") return str(count) def fmt_bytes(value: Any) -> str: count = as_int(value, -1) if count < 0: return "" if count >= 1_000_000_000: return f"{count / 1_000_000_000:.1f}G".replace(".0G", "G") if count >= 1_000_000: return f"{count / 1_000_000:.1f}M".replace(".0M", "M") if count >= 1_000: return f"{count / 1_000:.1f}K".replace(".0K", "K") return f"{count}B" def fmt_transfer(row: dict[str, Any]) -> str: if row.get("call_type") != "private_data": return fmt_duration(row.get("duration_seconds")) rx_bytes = fmt_bytes(row.get("private_data_rx_bytes")) tx_bytes = fmt_bytes(row.get("private_data_tx_bytes")) parts = [] if rx_bytes: parts.append(f"RX {rx_bytes}") if tx_bytes and as_int(row.get("private_data_tx_bytes"), 0) > 0: parts.append(f"TX {tx_bytes}") if parts: return "/".join(parts) packets = as_int(row.get("private_data_packet_count"), -1) if packets >= 0: return f"{packets} pkt" return "Data" def speaker_label(row: dict[str, Any]) -> str: callsign = as_text(row.get("source_callsign"), "") name = as_text(row.get("source_name"), "") source_id = as_text(row.get("source_id"), "") if callsign and name: return f"{callsign} ({name})" if callsign: return callsign return source_id or "-" def destination_label(row: dict[str, Any]) -> str: talkgroup = row.get("destination_talkgroup") user_id = row.get("destination_user_id") timeslot = row.get("timeslot") if talkgroup: label = f"TG {talkgroup}" if timeslot: label += f" TS{timeslot}" name = as_text(row.get("destination_talkgroup_name"), "") return f"{label} {name}" if name else label if user_id: label = f"Private {user_id}" if timeslot: label += f" TS{timeslot}" return label return "-" def activity_label(activity: dict[str, Any]) -> str: slot = as_text(activity.get("timeslot"), "?") if activity.get("destination_user_id"): dest = f"Private {activity.get('destination_user_id')}" elif activity.get("destination_talkgroup"): dest = f"TG {activity.get('destination_talkgroup')}" else: dest = "-" return f"TS{slot} {dest} {speaker_label(activity)}".strip() def talkgroup_active_label(row: dict[str, Any]) -> str: callsign = as_text(row.get("source_callsign"), "") name = as_text(row.get("source_name"), "") source_id = as_text(row.get("active_source_id"), "") if callsign and name: return f"{callsign} ({name})" if callsign: return callsign return source_id or "-" def talkgroup_timeout_label(row: dict[str, Any]) -> str: if row.get("routing_mode") != "dynamic": return "-" timeout = as_int(row.get("camp_timeout_seconds"), 0) return f"{timeout}s" if timeout > 0 else "-" def roaming_location_type(row: dict[str, Any]) -> str: return as_text(row.get("location_type") or row.get("protocol") or row.get("peer_kind") or "-") def roaming_location_name(row: dict[str, Any]) -> str: return as_text( row.get("location_name") or row.get("peer_callsign") or row.get("peer_display_name") or row.get("peer_site_name") or "-" ) def roaming_motorola_lrrp_waiting(row: dict[str, Any]) -> bool: if as_text(row.get("motorola_lrrp_status"), "").lower() != "waiting": return False lrrp_seen = as_int(row.get("motorola_lrrp_seen_at_epoch"), 0) gps_seen = as_int(row.get("gps_received_at_epoch"), 0) return lrrp_seen > 0 and (gps_seen == 0 or lrrp_seen >= gps_seen) def roaming_gps_position(row: dict[str, Any]) -> str: if roaming_motorola_lrrp_waiting(row): return "Waiting" latitude = row.get("gps_latitude") longitude = row.get("gps_longitude") if latitude is None or longitude is None: return "-" try: return f"{float(latitude):.2f},{float(longitude):.2f}" except (TypeError, ValueError): return "-" def roaming_motorola_ars_active(row: dict[str, Any]) -> bool: return as_int(row.get("motorola_ars_active"), 0) == 1 def roaming_opengd77_ta_gps_active(row: dict[str, Any]) -> bool: return ( row.get("gps_latitude") is not None and row.get("gps_longitude") is not None and as_text(row.get("gps_talker"), "").lower() == "opengd77" and as_text(row.get("gps_sentence_type"), "").upper() == "TA-GPS" ) def roaming_anytone_aprs_gps_active(row: dict[str, Any]) -> bool: return ( row.get("gps_latitude") is not None and row.get("gps_longitude") is not None and not roaming_opengd77_ta_gps_active(row) and as_int(row.get("gps_destination_id"), 0) == 9998 ) def roaming_radio_flag(row: dict[str, Any]) -> str: if roaming_motorola_ars_active(row): return "ARS" if roaming_opengd77_ta_gps_active(row): return "OGD77" if roaming_anytone_aprs_gps_active(row): return "ANYT" return "-" def roaming_radio_id_label(row: dict[str, Any]) -> str: return as_text(row.get("radio_id") or "-") def truncate_label(value: Any, max_len: int) -> str: text = as_text(value, "-").strip() if not text: return "-" if len(text) <= max_len: return text if max_len <= 3: return text[:max_len] return text[: max_len - 3] + "..." def endpoint_identity(row: dict[str, Any]) -> str: if row.get("kind") == "mmdvm_hotspot": return hotspot_identity(row) return as_text(row.get("display_name") or row.get("callsign") or row.get("radio_id")) def hotspot_identity(row: dict[str, Any]) -> str: callsign = as_text(row.get("callsign"), "") display_name = as_text(row.get("display_name"), "") radio_id = as_text(row.get("radio_id"), "") if callsign: if display_name.startswith(f"{callsign}-"): return display_name suffix = hotspot_radio_suffix(radio_id) if suffix and not callsign.endswith(f"-{suffix}"): return f"{callsign}-{suffix}" return callsign return display_name or radio_id def hotspot_radio_suffix(radio_id: str) -> str: radio_id = radio_id.strip() if len(radio_id) <= 7 or not radio_id.isdigit(): return "" return radio_id[7:] def endpoint_name(row: dict[str, Any]) -> str: return as_text(row.get("peer_display_name") or row.get("site_name") or row.get("display_name")) def endpoint_repeater_flag(row: dict[str, Any]) -> str: return "NS" if as_int(row.get("non_standard_repeater"), 0) else "-" def endpoint_activity(row: dict[str, Any], slot: int) -> dict[str, Any] | None: activity = row.get("slot_activity") or {} if not isinstance(activity, dict): return None value = activity.get(str(slot)) or activity.get(slot) return value if isinstance(value, dict) else None def endpoint_camps(row: dict[str, Any], slot: int | None = None) -> list[dict[str, Any]]: camps = row.get("camps") or [] if not isinstance(camps, list): return [] normalized = [camp for camp in camps if isinstance(camp, dict)] if slot is None: return normalized return [camp for camp in normalized if as_int(camp.get("timeslot"), 0) == slot] def endpoint_has_activity(row: dict[str, Any], slot: int) -> bool: activity = endpoint_activity(row, slot) return bool(activity and (activity.get("talkgroup") is not None or activity.get("is_direct"))) def simplex_display_slot(row: dict[str, Any]) -> int: if endpoint_has_activity(row, 2) or endpoint_camps(row, 2): return 2 return 1 def endpoint_status(row: dict[str, Any]) -> str: if endpoint_has_activity(row, 1) or endpoint_has_activity(row, 2): return "Talking" if as_int(row.get("connected"), 0) and as_int(row.get("sandbag_active"), 0): return "Scan" if as_int(row.get("connected"), 0) and endpoint_camps(row): return "Listening" if as_int(row.get("connected"), 0): return "OK" return "Down" def endpoint_slot_state(row: dict[str, Any], slot: int) -> str: activity = endpoint_activity(row, slot) if activity and (activity.get("talkgroup") is not None or activity.get("is_direct")): return "TX" if activity.get("direction") == "tx" else "RX" if as_int(row.get("connected"), 0) and endpoint_camps(row, slot): return "LK" if as_int(row.get("connected"), 0): return "--" return "Down" def endpoint_talkgroup_label(row: dict[str, Any], slot: int) -> str: activity = endpoint_activity(row, slot) if activity and (activity.get("talkgroup") is not None or activity.get("is_direct")): if activity.get("is_direct"): dest = "Private" elif as_int(row.get("sandbag_active"), 0) and as_int(activity.get("talkgroup"), 0) == 777: dest = "Scan" else: dest = f"TG {activity.get('talkgroup')}" return f"{dest} {fmt_elapsed(activity.get('age_seconds'))}" camps = endpoint_camps(row, slot) if camps: camp = camps[0] return f"TG {camp.get('talkgroup')} {as_int(camp.get('remaining_seconds'), 0)}s" return "-" def endpoint_speaker_label(row: dict[str, Any], slot: int) -> str: activity = endpoint_activity(row, slot) if not activity: return "-" callsign = as_text(activity.get("source_callsign"), "") name = as_text(activity.get("source_name"), "") source_id = as_text(activity.get("source_id"), "") if callsign and name: return f"{callsign} ({name})" if callsign: return callsign return source_id or "-" def endpoint_signal_label(row: dict[str, Any], slot: int) -> str: activity = endpoint_activity(row, slot) if not activity: return "-" parts = [] if activity.get("last_ber") is not None: parts.append(str(activity.get("last_ber"))) if activity.get("last_rssi") is not None: rssi = as_int(activity.get("last_rssi"), 0) parts.append(f"{'-' if rssi > 0 else ''}{rssi}dBm") return " / ".join(parts) if parts else "-" def fmt_elapsed(value: Any) -> str: seconds = as_int(value, 0) minutes, sec = divmod(max(0, seconds), 60) hours, minutes = divmod(minutes, 60) if hours: return f"{hours}:{minutes:02d}:{sec:02d}" return f"{minutes}:{sec:02d}" def clip(value: Any, width: int) -> str: text = as_text(value) if width <= 0: return "" if len(text) <= width: return text if width <= 1: return text[:width] return text[: width - 1] + "…" def tsv(value: Any) -> str: return as_text(value, "").replace("\t", " ").replace("\r", " ").replace("\n", " ") def last_heard_key(row: dict[str, Any]) -> tuple[Any, ...]: return ( row.get("radio_id"), row.get("last_seen_at_epoch"), row.get("peer_callsign") or row.get("peer_kind") or row.get("protocol"), row.get("destination_talkgroup"), row.get("destination_user_id"), row.get("timeslot"), row.get("call_type"), ) def last_heard_log_line(row: dict[str, Any]) -> str: seen_via = row.get("peer_callsign") or row.get("peer_kind") or row.get("protocol") or "-" fields = [ fmt_epoch(row.get("last_seen_at_epoch")), row.get("radio_id"), row.get("radio_callsign") or "-", row.get("radio_name") or "-", seen_via, destination_label(row), row.get("call_type") or "-", fmt_transfer(row), ] return "\t".join(tsv(field) for field in fields) class Screen: def __init__(self, stdscr: Any, args: argparse.Namespace) -> None: self.stdscr = stdscr self.args = args self.tab_index = 0 self.scroll_offsets = {tab: 0 for tab in TABS} self.data: dict[str, Any] = {} self.error = "" self.last_fetch = 0.0 self.lastlog_seen: set[tuple[Any, ...]] = set() self.lastlog_initialized = False self.lastlog_error = "" self.init_lastlog() def init_lastlog(self) -> None: if not self.args.lastlog: return try: with open(self.args.lastlog, "a", encoding="utf-8") as handle: started = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") handle.write(f"# ipsc3view last-heard log started {started}\n") handle.write("# time\tradio_id\tcallsign\tname\tseen_via\tdestination\tcall_type\tduration_or_data\n") except OSError as exc: self.lastlog_error = str(exc) def update_lastlog(self, rows: Any) -> None: if not self.args.lastlog or self.lastlog_error: return if not isinstance(rows, list): return valid_rows = [row for row in rows if isinstance(row, dict)] if not self.lastlog_initialized: self.lastlog_seen.update(last_heard_key(row) for row in valid_rows) self.lastlog_initialized = True return new_rows = [] for row in reversed(valid_rows): key = last_heard_key(row) if key in self.lastlog_seen: continue self.lastlog_seen.add(key) new_rows.append(row) if not new_rows: return try: with open(self.args.lastlog, "a", encoding="utf-8") as handle: for row in new_rows: handle.write(last_heard_log_line(row) + "\n") except OSError as exc: self.lastlog_error = str(exc) def run(self) -> None: curses.curs_set(0) self.stdscr.nodelay(True) self.stdscr.timeout(100) if curses.has_colors(): curses.start_color() curses.use_default_colors() curses.init_pair(1, curses.COLOR_CYAN, -1) curses.init_pair(2, curses.COLOR_GREEN, -1) curses.init_pair(3, curses.COLOR_YELLOW, -1) curses.init_pair(4, curses.COLOR_RED, -1) curses.init_pair(5, curses.COLOR_BLUE, -1) curses.init_pair(6, curses.COLOR_WHITE, -1) self.refresh_data(force=True) while True: key = self.stdscr.getch() if key in (ord("q"), ord("Q"), 27): return if key in (curses.KEY_RIGHT, ord("\t")): self.tab_index = (self.tab_index + 1) % len(TABS) elif key == curses.KEY_LEFT: self.tab_index = (self.tab_index - 1) % len(TABS) elif ord("1") <= key <= ord("0") + len(TABS): self.tab_index = key - ord("1") elif key in (curses.KEY_UP, ord("k"), ord("K")): self.scroll_active_tab(-1) elif key in (curses.KEY_DOWN, ord("j"), ord("J")): self.scroll_active_tab(1) elif key == curses.KEY_PPAGE: self.scroll_active_tab(-self.page_step()) elif key == curses.KEY_NPAGE: self.scroll_active_tab(self.page_step()) elif key == curses.KEY_HOME: self.set_active_scroll(0) elif key == curses.KEY_END: self.set_active_scroll(self.max_scroll_offset(self.active_tab())) elif key in (ord("r"), ord("R")): self.refresh_data(force=True) self.refresh_data() self.draw() def active_tab(self) -> str: return TABS[self.tab_index] def table_capacity(self, start_y: int, height: int) -> int: return max(1, height - start_y - 4) def page_step(self) -> int: height, _ = self.stdscr.getmaxyx() return max(1, self.table_capacity(5, height) - 1) def tab_item_count(self, tab: str) -> int: if tab == "Roaming": roaming = self.data.get("roaming") or {} return len(roaming.get("rows") or []) if isinstance(roaming, dict) else 0 if tab == "Talkgroups": return len(self.data.get("talkgroups") or []) if tab == "Repeaters": return len(self.repeater_slot_rows()) if tab == "Hotspots": return len(self.hotspot_slot_rows()) return 0 def max_scroll_offset(self, tab: str) -> int: if tab not in SCROLLABLE_TABS: return 0 height, _ = self.stdscr.getmaxyx() return max(0, self.tab_item_count(tab) - self.table_capacity(5, height)) def set_active_scroll(self, offset: int) -> None: tab = self.active_tab() if tab not in SCROLLABLE_TABS: return self.scroll_offsets[tab] = max(0, min(offset, self.max_scroll_offset(tab))) def scroll_active_tab(self, delta: int) -> None: tab = self.active_tab() self.set_active_scroll(self.scroll_offsets.get(tab, 0) + delta) def scroll_window(self, tab: str, total: int, capacity: int) -> tuple[int, int]: max_offset = max(0, total - capacity) offset = max(0, min(self.scroll_offsets.get(tab, 0), max_offset)) self.scroll_offsets[tab] = offset return offset, min(total, offset + capacity) def add_scroll_hint(self, y: int, total: int, start: int, end: int) -> None: if total <= 0: return hint = f"showing {start + 1}-{end} of {total} ↑/↓ scroll PgUp/PgDn page Home/End" self.add(y, 0, hint, curses.A_DIM if hasattr(curses, "A_DIM") else 0) def refresh_data(self, force: bool = False) -> None: now = time.monotonic() if not force and now - self.last_fetch < self.args.refresh: return self.last_fetch = now try: self.data = fetch_json(self.args.api_url, self.args.timeout) self.update_lastlog(self.data.get("lastHeard")) self.error = "" except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, json.JSONDecodeError) as exc: self.error = str(exc) def lastlog_status(self) -> str: if not self.args.lastlog: return "" if self.lastlog_error: return f" lastlog error: {self.lastlog_error}" return f" lastlog {self.args.lastlog}" def add(self, y: int, x: int, text: Any, attr: int = 0) -> None: height, width = self.stdscr.getmaxyx() if y < 0 or y >= height or x >= width: return safe = clip(text, width - x - 1) try: self.stdscr.addstr(y, x, safe, attr) except curses.error: pass def hline(self, y: int, x: int = 0, length: int | None = None, attr: int = 0) -> None: height, width = self.stdscr.getmaxyx() if y < 0 or y >= height or x >= width: return line_length = max(0, min(length if length is not None else width - x, width - x)) if line_length <= 0: return try: self.stdscr.hline(y, x, curses.ACS_HLINE, line_length, attr) except curses.error: self.add(y, x, "-" * line_length, attr) def draw_border(self, attr: int = 0) -> None: height, width = self.stdscr.getmaxyx() if height < 4 or width < 4: return try: self.stdscr.border( curses.ACS_VLINE, curses.ACS_VLINE, curses.ACS_HLINE, curses.ACS_HLINE, curses.ACS_ULCORNER, curses.ACS_URCORNER, curses.ACS_LLCORNER, curses.ACS_LRCORNER, ) if attr: self.stdscr.chgat(0, 0, width, attr) self.stdscr.chgat(height - 1, 0, width, attr) except curses.error: pass def title_attr(self) -> int: return curses.color_pair(1) | curses.A_BOLD if curses.has_colors() else curses.A_BOLD def muted_attr(self) -> int: return curses.color_pair(6) | curses.A_DIM if curses.has_colors() else curses.A_DIM def accent_attr(self) -> int: return curses.color_pair(5) if curses.has_colors() else curses.A_NORMAL def draw(self) -> None: self.stdscr.erase() height, width = self.stdscr.getmaxyx() if height < 8 or width < 50: self.add(0, 0, "Terminal too small for ipsc3view") self.stdscr.refresh() return version = IPSC3VIEW_VERSION api_version = as_text(self.data.get("version"), "unknown") title_attr = self.title_attr() top_attr = title_attr | curses.A_REVERSE line_attr = self.accent_attr() muted_attr = self.muted_attr() self.hline(2, 0, width, line_attr) self.hline(4, 0, width, line_attr) self.hline(height - 1, 0, width, line_attr) self.add(0, 0, f" IPSC3 VIEW {version} ", top_attr) generated = fmt_epoch(self.data.get("generatedAtEpoch")) if generated == "-": generated = "not loaded" self.add(0, 22 + len(version), f"API {api_version}", title_attr) self.add(0, 34 + len(version), self.args.api_url, muted_attr) self.add( 1, 0, f"Updated {generated} {local_tz_label()} refresh {self.args.refresh:.1f}s q quit r refresh 1-{len(TABS)} tabs arrows/PgUp/PgDn scroll{self.lastlog_status()}", muted_attr, ) tab_x = 0 for index, tab in enumerate(TABS): if index == self.tab_index: attr = title_attr | curses.A_REVERSE label = f" {index + 1}:{tab} " else: attr = muted_attr label = f" {index + 1}:{tab} " self.add(3, tab_x, label, attr) tab_x += len(tab) + 5 if self.error: attr = curses.color_pair(4) | curses.A_BOLD if curses.has_colors() else curses.A_BOLD self.add(5, 0, f"API error: {self.error}", attr) self.stdscr.refresh() return if TABS[self.tab_index] == "Overview": self.draw_overview(5, height) elif TABS[self.tab_index] == "Last Heard": self.draw_last_heard(5, height) elif TABS[self.tab_index] == "Roaming": self.draw_roaming(5, height, width) elif TABS[self.tab_index] == "Trunks": self.draw_trunks(5, height, width) elif TABS[self.tab_index] == "Talkgroups": self.draw_talkgroups(5, height, width) elif TABS[self.tab_index] == "Repeaters": self.draw_repeaters(5, height, width) else: self.draw_hotspots(5, height, width) self.stdscr.refresh() def header(self, y: int, columns: list[tuple[str, int]]) -> int: x = 0 attr = self.title_attr() | curses.A_REVERSE _, width = self.stdscr.getmaxyx() self.add(y, 0, " " * max(1, width - 1), attr) for label, width in columns: self.add(y, x, clip(label, width), attr) x += width + 1 return y + 1 def row(self, y: int, columns: list[tuple[Any, int, int | None]]) -> int: x = 0 for value, width, attr in columns: self.add(y, x, clip(value, width), attr or 0) x += width + 1 return y + 1 def repeater_slot_rows(self) -> list[tuple[dict[str, Any], int]]: return [ (row, slot) for row in self.data.get("repeaters") or [] if isinstance(row, dict) for slot in (1, 2) ] def hotspot_slot_rows(self) -> list[tuple[dict[str, Any], int]]: slot_rows: list[tuple[dict[str, Any], int]] = [] for row in self.data.get("hotspots") or []: if not isinstance(row, dict): continue for slot in ((simplex_display_slot(row),) if row.get("simplex") else (1, 2)): slot_rows.append((row, slot)) return slot_rows def draw_overview(self, start_y: int, height: int) -> None: overview = self.data.get("overview") or {} counts = ( f"Connections {overview.get('connectedPeers', 0)} / {overview.get('peers', 0)} " f"Talkgroups {overview.get('talkgroups', 0)} " f"Active {overview.get('activeTalkgroups', 0)}" ) self.add(start_y, 0, counts, curses.A_BOLD) columns = [("TG", 7), ("Name", 24), ("TS", 3), ("State", 6), ("Speaker", 24), ("Peer", 14), ("Stream", 16)] y = self.header(start_y + 2, columns) for row in overview.get("boardRows") or []: if y >= height - 1: break age = as_int(row.get("age_seconds"), 999) is_up = age <= ACTIVE_TALKGROUP_WINDOW_SECONDS attr = curses.color_pair(2) | curses.A_BOLD if curses.has_colors() and is_up else curses.A_NORMAL y = self.row( y, [ (row.get("talkgroup"), 7, None), (row.get("name"), 24, None), (row.get("active_timeslot") or row.get("configured_slot") or "-", 3, None), ("UP" if is_up else "IDLE", 6, attr), (speaker_label(row), 24, None), (row.get("peer_callsign") or row.get("peer_kind") or "-", 14, None), (row.get("stream_id"), 16, None), ], ) if y == start_y + 3: self.add(y, 0, "No active talkgroups") def draw_last_heard(self, start_y: int, height: int) -> None: rows = self.data.get("lastHeard") or [] self.add(start_y, 0, f"Most recent {len(rows)} radios", curses.A_BOLD) columns = [("Radio ID", 10), ("Call", 9), ("Name", 20), ("Seen Via", 12), ("Destination", 22), ("Type", 13), ("Dur/Data", 12), ("Last Seen Local", 15)] y = self.header(start_y + 2, columns) for row in rows: if y >= height - 1: break y = self.row( y, [ (row.get("radio_id"), 10, None), (row.get("radio_callsign") or "-", 9, None), (row.get("radio_name") or "-", 20, None), (row.get("peer_callsign") or row.get("peer_kind") or row.get("protocol"), 12, None), (destination_label(row), 22, None), (row.get("call_type") or "-", 13, None), (fmt_transfer(row), 12, None), (fmt_epoch(row.get("last_seen_at_epoch")), 15, None), ], ) if not rows: self.add(y, 0, "No last-heard radios returned by API") def draw_roaming(self, start_y: int, height: int, width: int) -> None: roaming = self.data.get("roaming") or {} rows = roaming.get("rows") or [] if isinstance(roaming, dict) else [] dwell_hours = as_int(roaming.get("dwellHours"), 0) if isinstance(roaming, dict) else 0 self.add(start_y, 0, f"Roaming {len(rows)} radios inside {dwell_hours} hour dwell window", curses.A_BOLD) self.add(start_y + 1, 0, "If a radio is not shown here the caller will receive a busy signal.", self.muted_attr()) capacity = self.table_capacity(start_y, height) start, end = self.scroll_window("Roaming", len(rows), capacity) self.add_scroll_hint(start_y + 2, len(rows), start, end) columns = [ ("Radio", 5), ("Radio ID", 9), ("Call", 9), ("Name", 18), ("Location", 10), ("Location Name", 18), ("Radio-GPS Position", 18), ("Last Seen", 8), ] y = self.header(start_y + 3, columns) for row in rows[start:end]: if y >= height - 1: break if not isinstance(row, dict): continue y = self.row( y, [ (roaming_radio_flag(row), 5, None), (roaming_radio_id_label(row), 9, None), (row.get("radio_callsign") or "-", 9, None), (truncate_label(row.get("radio_name"), 18), 18, None), (roaming_location_type(row), 10, None), (roaming_location_name(row), 18, None), (roaming_gps_position(row), 18, None), (fmt_time(row.get("last_seen_at_epoch")), 8, None), ], ) if not rows: self.add(y, 0, "No radios have been seen inside the current roaming dwell window") def draw_trunks(self, start_y: int, height: int, width: int) -> None: rows = self.data.get("trunks") or [] self.add(start_y, 0, f"Trunks {len(rows)}", curses.A_BOLD) columns = [("Name", 12), ("Kind", 12), ("State", 8), ("Endpoint", 22), ("RX/TX", 13), ("Last Seen Local", 15), ("Activity", 28)] y = self.header(start_y + 2, columns) for row in rows: if y >= height - 1: break state = "DISABLED" if not as_int(row.get("enabled"), 0) else ("OK" if as_int(row.get("connected"), 0) else "DOWN") state_attr = 0 if curses.has_colors(): state_attr = curses.color_pair(2 if state == "OK" else 4 if state == "DOWN" else 3) | curses.A_BOLD endpoint = "-" if row.get("current_host"): endpoint = f"{row.get('current_host')}:{row.get('current_port') or '-'}" activity = ", ".join(activity_label(item) for item in row.get("activity") or []) or "-" y = self.row( y, [ (row.get("callsign") or row.get("display_name"), 12, None), (row.get("kind"), 12, None), (state, 8, state_attr), (endpoint, 22, None), (f"{fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}", 13, None), (fmt_epoch(row.get("last_seen_at_epoch")), 15, None), (activity, 28, None), ], ) if not rows: self.add(y, 0, "No trunks returned by API") def draw_talkgroups(self, start_y: int, height: int, width: int) -> None: rows = self.data.get("talkgroups") or [] self.add(start_y, 0, f"Talkgroups {len(rows)}", curses.A_BOLD) capacity = self.table_capacity(start_y, height) start, end = self.scroll_window("Talkgroups", len(rows), capacity) self.add_scroll_hint(start_y + 1, len(rows), start, end) columns = [ ("TG", 7), ("Name", 24), ("TS", 3), ("Routing", 8), ("Trunks", 18), ("Listen", 7), ("Timeout", 8), ("Active", 24), ] y = self.header(start_y + 2, columns) for row in rows[start:end]: if y >= height - 1: break active = talkgroup_active_label(row) active_attr = curses.color_pair(2) | curses.A_BOLD if curses.has_colors() and active != "-" else 0 listen = "Yes" if row.get("listen_only") else "No" listen_attr = curses.color_pair(3) | curses.A_BOLD if curses.has_colors() and listen == "Yes" else 0 y = self.row( y, [ (row.get("talkgroup"), 7, None), (row.get("name") or "-", 24, None), (row.get("default_slot") or "-", 3, None), (row.get("routing_mode") or "-", 8, None), (row.get("trunk_summary") or "-", 18, None), (listen, 7, listen_attr), (talkgroup_timeout_label(row), 8, None), (active, 24, active_attr), ], ) if not rows: self.add(y, 0, "No talkgroups returned by API") def endpoint_state_attr(self, state: str) -> int: if not curses.has_colors(): return curses.A_BOLD if state in ("TX", "RX", "LK", "Talking", "Listening") else 0 if state in ("TX", "RX", "LK", "Talking", "Listening"): return curses.color_pair(2) | curses.A_BOLD if state in ("Down", "Disconnected"): return curses.color_pair(4) | curses.A_BOLD if state == "Scan": return curses.color_pair(3) | curses.A_BOLD return curses.A_NORMAL def draw_repeaters(self, start_y: int, height: int, width: int) -> None: rows = self.data.get("repeaters") or [] slot_rows = self.repeater_slot_rows() self.add(start_y, 0, f"Repeaters {len(rows)}", curses.A_BOLD) capacity = self.table_capacity(start_y, height) start, end = self.scroll_window("Repeaters", len(slot_rows), capacity) self.add_scroll_hint(start_y + 1, len(slot_rows), start, end) columns = [ ("Call", 11), ("Name", 15), ("Flag", 4), ("Status", 8), ("Slot", 4), ("State", 5), ("Talkgroup", 14), ("Speaker", 20), ("Last Seen", 15), ("RX/TX", 11), ] y = self.header(start_y + 2, columns) for row, slot in slot_rows[start:end]: if y >= height - 1: break status = endpoint_status(row) state = endpoint_slot_state(row, slot) flag = endpoint_repeater_flag(row) flag_attr = curses.color_pair(3) | curses.A_BOLD if curses.has_colors() and flag != "-" else 0 y = self.row( y, [ (endpoint_identity(row), 11, None), (endpoint_name(row), 15, None), (flag, 4, flag_attr), (status, 8, self.endpoint_state_attr(status)), (f"TS{slot}", 4, None), (state, 5, self.endpoint_state_attr(state)), (endpoint_talkgroup_label(row, slot), 14, None), (endpoint_speaker_label(row, slot), 20, None), (fmt_epoch(row.get("last_seen_at_epoch")), 15, None), (f"{fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}", 11, None), ], ) if not rows: self.add(y, 0, "No repeaters returned by API") def draw_hotspots(self, start_y: int, height: int, width: int) -> None: rows = self.data.get("hotspots") or [] slot_rows = self.hotspot_slot_rows() self.add(start_y, 0, f"Hotspots {len(rows)}", curses.A_BOLD) capacity = self.table_capacity(start_y, height) start, end = self.scroll_window("Hotspots", len(slot_rows), capacity) self.add_scroll_hint(start_y + 1, len(slot_rows), start, end) columns = [ ("Call", 12), ("Kind", 7), ("Status", 9), ("Slot", 4), ("State", 5), ("Talkgroup", 16), ("Speaker", 24), ("BER/RSSI", 12), ("RX/TX", 12), ] y = self.header(start_y + 2, columns) for row, slot in slot_rows[start:end]: if y >= height - 1: break status = endpoint_status(row) state = endpoint_slot_state(row, slot) y = self.row( y, [ (endpoint_identity(row), 12, None), ("Simplex" if row.get("simplex") else "Duplex", 7, None), (status, 9, self.endpoint_state_attr(status)), ("-" if row.get("simplex") else f"TS{slot}", 4, None), (state, 5, self.endpoint_state_attr(state)), (endpoint_talkgroup_label(row, slot), 16, None), (endpoint_speaker_label(row, slot), 24, None), (endpoint_signal_label(row, slot), 12, None), (f"{fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}", 12, None), ], ) if not rows: self.add(y, 0, "No hotspots returned by API") def dump(data: dict[str, Any], section: str) -> None: if section in ("all", "overview"): overview = data.get("overview") or {} print("Overview") print(f" Connections: {overview.get('connectedPeers', 0)} / {overview.get('peers', 0)}") print(f" Talkgroups: {overview.get('talkgroups', 0)}") print(f" Active talkgroups: {overview.get('activeTalkgroups', 0)}") for row in overview.get("boardRows") or []: slot = row.get("active_timeslot") or row.get("configured_slot") or "-" print(f" TG {row.get('talkgroup')} TS{slot} {row.get('name')} {speaker_label(row)}") if section in ("all", "radios"): print("Last Heard") for row in data.get("lastHeard") or []: print( f" {roaming_radio_id_label(row)} {row.get('radio_callsign') or '-'} " f"{destination_label(row)} via {row.get('peer_callsign') or row.get('peer_kind') or row.get('protocol')} " f"{fmt_transfer(row)} {fmt_epoch(row.get('last_seen_at_epoch'))}" ) if section in ("all", "roaming"): roaming = data.get("roaming") or {} rows = roaming.get("rows") or [] if isinstance(roaming, dict) else [] dwell_hours = as_int(roaming.get("dwellHours"), 0) if isinstance(roaming, dict) else 0 print(f"Roaming ({dwell_hours}h dwell)") for row in rows: if not isinstance(row, dict): continue print( f" {roaming_radio_flag(row):<5} {roaming_radio_id_label(row):<9} {row.get('radio_callsign') or '-'} " f"{truncate_label(row.get('radio_name'), 18)} via {roaming_location_type(row)} " f"{roaming_location_name(row)} gps {roaming_gps_position(row)} " f"{fmt_time(row.get('last_seen_at_epoch'))}" ) if section in ("all", "trunks"): print("Trunks") for row in data.get("trunks") or []: state = "DISABLED" if not as_int(row.get("enabled"), 0) else ("OK" if as_int(row.get("connected"), 0) else "DOWN") print(f" {row.get('callsign')} {state} rx/tx {fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}") if section in ("all", "talkgroups"): print("Talkgroups") for row in data.get("talkgroups") or []: slot = row.get("default_slot") or "-" listen = "listen-only" if row.get("listen_only") else "tx/rx" print( f" TG {row.get('talkgroup')} TS{slot} {row.get('name') or '-'} " f"{row.get('routing_mode') or '-'} trunks={row.get('trunk_summary') or '-'} " f"{listen} active={talkgroup_active_label(row)}" ) if section in ("all", "repeaters"): print("Repeaters") for row in data.get("repeaters") or []: status = endpoint_status(row) for slot in (1, 2): print( f" {endpoint_identity(row)} {endpoint_repeater_flag(row)} {status} TS{slot} " f"{endpoint_slot_state(row, slot)} {endpoint_talkgroup_label(row, slot)} " f"{endpoint_speaker_label(row, slot)} " f"rx/tx {fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}" ) if section in ("all", "hotspots"): print("Hotspots") for row in data.get("hotspots") or []: status = endpoint_status(row) slots = (simplex_display_slot(row),) if row.get("simplex") else (1, 2) for slot in slots: slot_label = "-" if row.get("simplex") else f"TS{slot}" print( f" {endpoint_identity(row)} {'simplex' if row.get('simplex') else 'duplex'} " f"{status} {slot_label} {endpoint_slot_state(row, slot)} " f"{endpoint_talkgroup_label(row, slot)} {endpoint_speaker_label(row, slot)} " f"ber/rssi {endpoint_signal_label(row, slot)} " f"rx/tx {fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}" ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="IPSC3 terminal dashboard over the public HTTP API.") parser.add_argument("--api-url", default=DEFAULT_API_URL, help=f"API URL, default {DEFAULT_API_URL}") parser.add_argument("--refresh", type=float, default=DEFAULT_REFRESH_SECONDS, help="Refresh interval in seconds") parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT_SECONDS, help="HTTP timeout in seconds") parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="Last-heard row limit") parser.add_argument("--lastlog", help="Append newly observed last-heard rows to this file while the dashboard runs") parser.add_argument( "--dump", choices=("all", "overview", "radios", "roaming", "trunks", "talkgroups", "repeaters", "hotspots"), help="Print once and exit", ) args = parser.parse_args() args.refresh = max(0.5, args.refresh) args.timeout = max(1.0, args.timeout) args.limit = max(1, min(args.limit, 100)) args.api_url = api_url_with_limit(args.api_url, args.limit) return args def main() -> int: set_process_name() set_terminal_title() args = parse_args() if args.dump: try: dump(fetch_json(args.api_url, args.timeout), args.dump) return 0 except VersionMismatchError as exc: return handle_version_mismatch(exc, args) except Exception as exc: print(f"ipsc3view: {exc}", file=sys.stderr) return 1 try: curses_module = ensure_curses() curses_module.wrapper(lambda stdscr: Screen(stdscr, args).run()) except VersionMismatchError as exc: return handle_version_mismatch(exc, args) except CursesDependencyError as exc: print(f"ipsc3view: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())