#!/usr/bin/env python3
#
# IPSC3 Project by Matt Perkins, VK2FLY
# IPSC3qt version 1.0.50
# Date: 2026-06-14
# Copyright (C) 2026 Matt Perkins, VK2FLY.
# Licensed under the GNU General Public License v3.0 or later.
from __future__ import annotations
import argparse
import ctypes
import json
import os
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import venv
from datetime import datetime
from pathlib import Path
from typing import Any
IPSC3QT_VERSION = "1.0.50"
# Compatibility marker for older auto-updaters that used the IPSC3 API
# version as the expected client version during the 1.0.30 transition.
# IPSC3QT_VERSION = "1.0.30"
PROCESS_NAME = "IPSC3Qt"
DEFAULT_API_URL = "https://ipsc3.vkdmr.com/api/ipsc3view"
DOWNLOADS_URL = "https://ipsc3.vkdmr.com/downloads"
DOWNLOAD_PATH = "/downloads/ipsc3qt"
DEFAULT_REFRESH_SECONDS = 2.0
DEFAULT_TIMEOUT_SECONDS = 5.0
DEFAULT_LIMIT = 30
ACTIVE_TALKGROUP_WINDOW_SECONDS = 15
MAP_REFRESH_SECONDS = 30.0
MAP_LIVE_REFRESH_SECONDS = 2.5
TABS = ("Overview", "Last Heard", "Roaming", "Map", "Repeaters", "Hotspots", "Trunks", "Talkgroups")
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
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),
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 dependency_venv_dir() -> Path:
if os.name == "nt":
base = Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local")
elif sys.platform == "darwin":
base = Path.home() / "Library" / "Application Support"
else:
base = Path(os.environ.get("XDG_DATA_HOME") or Path.home() / ".local" / "share")
return base / "ipsc3" / "ipsc3qt-venv"
def venv_python(venv_dir: Path) -> Path:
if os.name == "nt":
return venv_dir / "Scripts" / "python.exe"
return venv_dir / "bin" / "python"
def ensure_dependency_python(venv_dir: Path) -> Path:
python = venv_python(venv_dir)
if not python.exists() and venv_dir.exists():
shutil.rmtree(venv_dir)
if not python.exists():
venv_dir.parent.mkdir(parents=True, exist_ok=True)
venv.EnvBuilder(with_pip=True, clear=False).create(venv_dir)
if not python.exists():
raise RuntimeError(f"dependency Python was not created at {python}")
return python
def relaunch_with_python(python: Path, script_path: str, env: dict[str, str]) -> None:
argv = [str(python), script_path, *sys.argv[1:]]
if os.name == "nt":
subprocess.Popen(argv, env=env, cwd=os.getcwd())
raise SystemExit(0)
os.execvpe(str(python), argv, env)
def ensure_pyside6() -> None:
try:
import PySide6 # noqa: F401
return
except ImportError:
pass
if os.environ.get("IPSC3QT_BOOTSTRAPPED") == "1":
raise RuntimeError("PySide6 is not available after dependency bootstrap")
script_path = current_script_path()
if not script_path:
raise RuntimeError(
"could not determine the ipsc3qt script path. "
"Run ipsc3qt from a real file path, for example: py .\\ipsc3qt.py"
)
venv_dir = dependency_venv_dir()
python = ensure_dependency_python(venv_dir)
print(f"ipsc3qt: installing PySide6 into {venv_dir}", file=sys.stderr)
pip_command = [
str(python),
"-m",
"pip",
"install",
"--disable-pip-version-check",
"--upgrade",
"PySide6",
]
subprocess.check_call(pip_command)
env = os.environ.copy()
env["IPSC3QT_BOOTSTRAPPED"] = "1"
relaunch_with_python(python, script_path, env)
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}/ipsc3qt"
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 ipsc3qt {download_url}\n"
" chmod +x ipsc3qt\n"
" ./ipsc3qt\n"
"\n"
"On Windows, run it with Python:\n"
" py ipsc3qt\n"
"\n"
"If ipsc3qt is installed in your PATH, replace it after download:\n"
" sudo install -m 0755 ipsc3qt /usr/local/bin/ipsc3qt"
)
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"] = IPSC3QT_VERSION
query["client"] = "ipsc3qt"
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"ipsc3qt: API version missing from {api_url}.\n\n"
f"{upgrade_instructions(api_url)}",
api_url=api_url,
)
viewer_version = data.get("ipsc3qtVersion") or data.get("ipsc3viewVersion") or api_version
if viewer_version and str(viewer_version) != IPSC3QT_VERSION:
raise VersionMismatchError(
f"ipsc3qt: viewer API version mismatch. Client {IPSC3QT_VERSION} requires "
f"IPSC3qt API {IPSC3QT_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
candidate = Path(invoked)
if candidate.exists():
return str(candidate.resolve())
if not candidate.is_absolute():
cwd_candidate = Path.cwd() / invoked
if cwd_candidate.exists():
return str(cwd_candidate.resolve())
path_candidate = shutil.which(invoked)
if path_candidate:
return str(Path(path_candidate).resolve())
return None
def validate_update_script(script: bytes, expected_version: str) -> None:
if len(script) > 1_500_000:
raise SelfUpdateError("downloaded ipsc3qt is unexpectedly large")
if not script.startswith(b"#!/usr/bin/env python3"):
raise SelfUpdateError("downloaded file does not look like ipsc3qt")
version_line = f'IPSC3QT_VERSION = "{expected_version}"'.encode("utf-8")
if version_line not in script:
raise SelfUpdateError(
f"downloaded ipsc3qt 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 ipsc3qt 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 ipsc3qt {script_path}"
)
request = urllib.request.Request(
download_url_for_api(api_url),
headers={
"Accept": "text/x-python,text/plain,*/*",
"User-Agent": f"ipsc3qt/{IPSC3QT_VERSION}",
},
)
with urllib.request.urlopen(request, timeout=timeout) as response:
script = response.read(1_500_001)
validate_update_script(script, target_version)
mode = os.stat(script_path).st_mode & 0o777 or 0o755
tmp_path = ""
try:
fd, tmp_path = tempfile.mkstemp(prefix=".ipsc3qt-", 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 fetch_json(api_url: str, timeout: float) -> dict[str, Any]:
request = urllib.request.Request(
api_url,
headers={"Accept": "application/json", "User-Agent": f"ipsc3qt/{IPSC3QT_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 map_base_url(api_url: str) -> str:
parsed = urllib.parse.urlparse(api_url)
if parsed.scheme in ("http", "https") and parsed.netloc:
return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, "/dashboard/map", "", "", ""))
return "https://ipsc3.vkdmr.com/dashboard/map"
def map_data_url(api_url: str, query: str, hours: int) -> str:
parsed = urllib.parse.urlparse(api_url)
if parsed.scheme not in ("http", "https") or not parsed.netloc:
parsed = urllib.parse.urlparse(DEFAULT_API_URL)
params = {"hours": str(hours)}
if query.strip():
params["q"] = query.strip()
return urllib.parse.urlunparse(
parsed._replace(path="/dashboard/map/data", query=urllib.parse.urlencode(params), fragment="")
)
def fetch_map_json(api_url: str, query: str, hours: int, timeout: float) -> dict[str, Any]:
request = urllib.request.Request(
map_data_url(api_url, query, hours),
headers={"Accept": "application/json", "User-Agent": f"ipsc3qt/{IPSC3QT_VERSION}"},
)
with urllib.request.urlopen(request, timeout=timeout) as response:
charset = response.headers.get_content_charset() or "utf-8"
return json.loads(response.read().decode(charset))
def map_html() -> str:
return r"""
Loading map...
"""
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 roaming_motorola_info(row: dict[str, Any]) -> str:
if not roaming_motorola_ars_active(row):
return ""
parts = ["Motorola ARS active"]
if row.get("motorola_ars_seen_at_epoch"):
parts.append(f"ARS seen {fmt_epoch(row.get('motorola_ars_seen_at_epoch'))}")
if row.get("motorola_ars_device_id"):
parts.append(f"device {row.get('motorola_ars_device_id')}")
if row.get("motorola_ars_user_id"):
parts.append(f"user {row.get('motorola_ars_user_id')}")
if as_int(row.get("motorola_ars_source_mismatch"), 0) == 1:
parts.append(f"outer source {row.get('motorola_ars_outer_source_id') or row.get('radio_id')}")
if row.get("motorola_ars_ip_source") and row.get("motorola_ars_ip_destination"):
parts.append(f"ip {row.get('motorola_ars_ip_source')} > {row.get('motorola_ars_ip_destination')}")
if row.get("motorola_ars_payload_hex"):
parts.append(f"ARS payload {row.get('motorola_ars_payload_hex')}")
if row.get("motorola_bms_seen_at_epoch"):
parts.append(f"BMS seen {fmt_epoch(row.get('motorola_bms_seen_at_epoch'))}")
if row.get("motorola_bms_radio_esn"):
parts.append(f"ESN {row.get('motorola_bms_radio_esn')}")
if row.get("motorola_bms_hash"):
parts.append(f"BMS hash {row.get('motorola_bms_hash')}")
if row.get("motorola_bms_battery_serial"):
parts.append(f"battery {row.get('motorola_bms_battery_serial')}")
if row.get("motorola_bms_code") is not None:
parts.append(f"BMS code 0x{as_int(row.get('motorola_bms_code'), 0):02X}")
return " | ".join(parts)
def roaming_opengd77_info(row: dict[str, Any]) -> str:
if not roaming_opengd77_ta_gps_active(row):
return ""
parts = ["OpenGD77 in-band TA-GPS"]
if row.get("gps_received_at_epoch"):
parts.append(f"GPS seen {fmt_epoch(row.get('gps_received_at_epoch'))}")
if row.get("gps_speed_knots") is not None:
try:
parts.append(f"speed {float(row.get('gps_speed_knots')):.1f} kt")
except (TypeError, ValueError):
pass
if row.get("gps_course_degrees") is not None:
try:
parts.append(f"course {float(row.get('gps_course_degrees')):.1f} deg")
except (TypeError, ValueError):
pass
sentence = row.get("gps_normalized_sentence") or row.get("gps_raw_sentence")
if sentence:
parts.append(str(sentence))
return " | ".join(parts)
def roaming_anytone_aprs_gps_info(row: dict[str, Any]) -> str:
if not roaming_anytone_aprs_gps_active(row):
return ""
parts = ["DMR APRS GPS via 9998"]
if row.get("gps_received_at_epoch"):
parts.append(f"GPS seen {fmt_epoch(row.get('gps_received_at_epoch'))}")
if row.get("gps_timeslot"):
parts.append(f"TS{row.get('gps_timeslot')}")
if row.get("gps_speed_knots") is not None:
try:
parts.append(f"speed {float(row.get('gps_speed_knots')):.1f} kt")
except (TypeError, ValueError):
pass
if row.get("gps_course_degrees") is not None:
try:
parts.append(f"course {float(row.get('gps_course_degrees')):.1f} deg")
except (TypeError, ValueError):
pass
sentence = row.get("gps_normalized_sentence") or row.get("gps_raw_sentence")
if sentence:
parts.append(str(sentence))
return " | ".join(parts)
def gps_point_position(point: dict[str, Any]) -> str:
latitude = point.get("latitude")
longitude = point.get("longitude")
if latitude is None or longitude is None:
return "-"
try:
return f"{float(latitude):.2f},{float(longitude):.2f}"
except (TypeError, ValueError):
return "-"
def gps_point_speed(point: dict[str, Any]) -> str:
value = point.get("speedKnots")
if value is None:
return "-"
try:
return f"{float(value):.1f} kt"
except (TypeError, ValueError):
return "-"
def gps_point_course(point: dict[str, Any]) -> str:
value = point.get("courseDegrees")
if value is None:
return "-"
try:
return f"{float(value):.0f} deg"
except (TypeError, ValueError):
return "-"
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 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 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 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 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)
def repeater_slot_rows(data: dict[str, Any]) -> list[list[str]]:
rows: list[list[str]] = []
for row in data.get("repeaters") or []:
if not isinstance(row, dict):
continue
status = endpoint_status(row)
for slot in (1, 2):
rows.append(
[
endpoint_identity(row),
endpoint_name(row),
endpoint_repeater_flag(row),
status,
f"TS{slot}",
endpoint_slot_state(row, slot),
endpoint_talkgroup_label(row, slot),
endpoint_speaker_label(row, slot),
fmt_epoch(row.get("last_seen_at_epoch")),
f"{fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}",
]
)
return rows
def hotspot_slot_rows(data: dict[str, Any]) -> list[list[str]]:
rows: list[list[str]] = []
for row in data.get("hotspots") or []:
if not isinstance(row, dict):
continue
status = endpoint_status(row)
slots = (simplex_display_slot(row),) if row.get("simplex") else (1, 2)
for slot in slots:
rows.append(
[
endpoint_identity(row),
"Simplex" if row.get("simplex") else "Duplex",
status,
"-" if row.get("simplex") else f"TS{slot}",
endpoint_slot_state(row, slot),
endpoint_talkgroup_label(row, slot),
endpoint_speaker_label(row, slot),
endpoint_signal_label(row, slot),
f"{fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}",
]
)
return rows
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 repeater_slot_rows(data):
print(" " + " ".join(row))
if section in ("all", "hotspots"):
print("Hotspots")
for row in hotspot_slot_rows(data):
print(" " + " ".join(row))
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="IPSC3 Qt 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 without starting the Qt GUI",
)
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 run_gui(args: argparse.Namespace) -> int:
ensure_pyside6()
from PySide6.QtCore import QObject, QTimer, Qt, QUrl, Signal, Slot
from PySide6.QtGui import QColor, QPalette
from PySide6.QtWidgets import (
QApplication,
QAbstractItemView,
QCheckBox,
QComboBox,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QMainWindow,
QMenu,
QMessageBox,
QProgressBar,
QPushButton,
QTabWidget,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
try:
from PySide6.QtWebEngineWidgets import QWebEngineView
except Exception:
QWebEngineView = None # type: ignore[assignment]
try:
from PySide6.QtWebChannel import QWebChannel
except Exception:
QWebChannel = None # type: ignore[assignment]
class FetchBridge(QObject):
data_ready = Signal(dict)
fetch_error = Signal(str)
version_error = Signal(str, str, str)
map_data_ready = Signal(dict)
map_fetch_error = Signal(str)
class MapSearchBridge(QObject):
search_requested = Signal(str)
@Slot(str)
def search(self, query: str) -> None:
self.search_requested.emit(str(query or "").strip())
class MainWindow(QMainWindow):
def __init__(self) -> None:
super().__init__()
self.bridge = FetchBridge()
self.bridge.data_ready.connect(self.on_data_ready)
self.bridge.fetch_error.connect(self.on_fetch_error)
self.bridge.version_error.connect(self.on_version_error)
self.bridge.map_data_ready.connect(self.on_map_data_ready)
self.bridge.map_fetch_error.connect(self.on_map_fetch_error)
self.fetch_in_progress = False
self.map_fetch_in_progress = False
self.update_in_progress = False
self.data: dict[str, Any] = {}
self.lastlog_seen: set[tuple[Any, ...]] = set()
self.lastlog_initialized = False
self.lastlog_error = ""
self.tables: dict[str, QTableWidget] = {}
self.summaries: dict[str, QLabel] = {}
self.map_web_view: Any = None
self.map_table: QTableWidget | None = None
self.map_query_input: QLineEdit | None = None
self.map_hours_combo: QComboBox | None = None
self.map_live_checkbox: QCheckBox | None = None
self.map_bridge: MapSearchBridge | None = None
self.map_channel: Any = None
self.last_map_data: dict[str, Any] | None = None
self.map_live_enabled = False
self.refresh_interval_ms = int(args.refresh * 1000)
self.next_refresh_at = time.monotonic() + args.refresh
self.setWindowTitle(f"IPSC3Qt {IPSC3QT_VERSION}")
self.resize(1280, 760)
self.build_ui()
self.init_lastlog()
self.timer = QTimer(self)
self.timer.timeout.connect(self.refresh_data)
self.timer.start(self.refresh_interval_ms)
self.progress_timer = QTimer(self)
self.progress_timer.timeout.connect(self.update_refresh_progress)
self.progress_timer.start(100)
self.map_timer = QTimer(self)
self.map_timer.timeout.connect(self.refresh_map_data)
self.map_timer.start(int(MAP_REFRESH_SECONDS * 1000))
self.refresh_data()
self.refresh_map_data()
def build_ui(self) -> None:
root = QWidget()
layout = QVBoxLayout(root)
layout.setContentsMargins(14, 14, 14, 14)
layout.setSpacing(10)
top = QHBoxLayout()
self.title_label = QLabel(f"IPSC3Qt {IPSC3QT_VERSION}")
self.title_label.setObjectName("Title")
self.meta_label = QLabel("")
self.meta_label.setTextInteractionFlags(Qt.TextSelectableByMouse)
self.meta_label.setVisible(False)
self.updated_label = QLabel("Waiting for data")
self.retry_button = QPushButton("Retry")
self.retry_button.clicked.connect(lambda: self.refresh_data(force=True))
self.retry_button.setVisible(False)
top.addWidget(self.title_label)
top.addStretch(1)
top.addWidget(self.meta_label)
top.addWidget(self.updated_label)
top.addWidget(self.retry_button)
layout.addLayout(top)
self.tabs = QTabWidget()
for tab in TABS:
if tab == "Map":
self.build_map_tab()
continue
page = QWidget()
page_layout = QVBoxLayout(page)
page_layout.setContentsMargins(8, 8, 8, 8)
page_layout.setSpacing(6)
summary = QLabel("")
summary.setTextInteractionFlags(Qt.TextSelectableByMouse)
table = QTableWidget()
table.setAlternatingRowColors(True)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
table.setSelectionBehavior(QAbstractItemView.SelectRows)
table.setSelectionMode(QAbstractItemView.SingleSelection)
table.verticalHeader().setVisible(False)
table.horizontalHeader().setStretchLastSection(True)
table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
page_layout.addWidget(summary)
page_layout.addWidget(table, 1)
self.summaries[tab] = summary
self.tables[tab] = table
self.tabs.addTab(page, tab)
layout.addWidget(self.tabs, 1)
bottom = QHBoxLayout()
bottom.setSpacing(10)
self.status_version_label = QLabel(f"IPSC3Qt {IPSC3QT_VERSION}")
self.status_version_label.setTextInteractionFlags(Qt.TextSelectableByMouse)
self.refresh_progress = QProgressBar()
self.refresh_progress.setRange(0, self.refresh_interval_ms)
self.refresh_progress.setValue(self.refresh_interval_ms)
self.refresh_progress.setTextVisible(False)
self.refresh_progress.setFixedWidth(self.status_version_label.sizeHint().width())
bottom.addWidget(self.status_version_label)
bottom.addWidget(self.refresh_progress)
bottom.addStretch(1)
layout.addLayout(bottom)
self.setCentralWidget(root)
self.setStyleSheet(
"""
QWidget { font: 13px "Menlo", "Consolas", "DejaVu Sans Mono", monospace; }
QMainWindow, QWidget { background: #071116; color: #eaf7f5; }
QLabel#Title { font-size: 20px; font-weight: 700; color: #7af2db; }
QTabWidget::pane { border: 1px solid rgba(122, 242, 219, 0.20); border-radius: 8px; }
QTabBar::tab { background: #10242c; padding: 8px 12px; border-top-left-radius: 6px; border-top-right-radius: 6px; }
QTabBar::tab:selected { background: #1c4b56; color: #ffffff; }
QTableWidget { gridline-color: #24444d; alternate-background-color: #0b1b22; background: #08151a; }
QLineEdit, QComboBox { background: #08151a; color: #eaf7f5; border: 1px solid #24444d; border-radius: 6px; padding: 6px; }
QHeaderView::section { background: #1c4b56; color: #ffffff; padding: 6px; border: 0; }
QTabWidget QLabel { color: #b8d8d4; }
QProgressBar { background: #08151a; border: 1px solid #24444d; border-radius: 5px; max-height: 10px; min-height: 10px; }
QProgressBar::chunk { background: #7af2db; border-radius: 4px; }
QPushButton { background: #1c4b56; color: #ffffff; border: 1px solid #4fb8aa; border-radius: 6px; padding: 7px 12px; }
QPushButton:hover { background: #266673; }
"""
)
def build_map_tab(self) -> None:
page = QWidget()
page_layout = QVBoxLayout(page)
page_layout.setContentsMargins(8, 8, 8, 8)
page_layout.setSpacing(8)
controls = QHBoxLayout()
controls.setSpacing(8)
controls.addWidget(QLabel("Call/ID"))
self.map_query_input = QLineEdit()
self.map_query_input.setPlaceholderText("VK2FLY or 5052540")
self.map_query_input.returnPressed.connect(lambda: self.refresh_map_data(force=True))
controls.addWidget(self.map_query_input, 1)
controls.addWidget(QLabel("Window"))
self.map_hours_combo = QComboBox()
for label, hours in (
("1 hour", 1),
("8 hours", 8),
("24 hours", 24),
("3 days", 72),
("7 days", 168),
):
self.map_hours_combo.addItem(label, hours)
self.map_hours_combo.setCurrentIndex(2)
self.map_hours_combo.currentIndexChanged.connect(lambda _index: self.refresh_map_data(force=True))
controls.addWidget(self.map_hours_combo)
search_button = QPushButton("Search Track")
search_button.clicked.connect(lambda: self.refresh_map_data(force=True))
controls.addWidget(search_button)
latest_button = QPushButton("Latest")
latest_button.clicked.connect(self.reset_map_search)
controls.addWidget(latest_button)
self.map_live_checkbox = QCheckBox("Live Map")
self.map_live_checkbox.toggled.connect(self.on_map_live_toggled)
controls.addWidget(self.map_live_checkbox)
page_layout.addLayout(controls)
summary = QLabel("Map waiting for GPS data")
summary.setTextInteractionFlags(Qt.TextSelectableByMouse)
page_layout.addWidget(summary)
self.summaries["Map"] = summary
if QWebEngineView is not None:
self.map_web_view = QWebEngineView()
self.map_web_view.setContextMenuPolicy(Qt.CustomContextMenu)
self.map_web_view.customContextMenuRequested.connect(self.show_map_context_menu)
if QWebChannel is not None:
self.map_bridge = MapSearchBridge()
self.map_bridge.search_requested.connect(self.search_map_callsign)
self.map_channel = QWebChannel(self.map_web_view.page())
self.map_channel.registerObject("ipsc3qt", self.map_bridge)
self.map_web_view.page().setWebChannel(self.map_channel)
self.map_web_view.setHtml(map_html(), QUrl(map_base_url(args.api_url)))
self.map_web_view.loadFinished.connect(lambda _ok: self.render_map_web_data())
page_layout.addWidget(self.map_web_view, 1)
else:
summary.setText("Qt WebEngine is not available. Showing GPS data table fallback.")
table = QTableWidget()
table.setAlternatingRowColors(True)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
table.setSelectionBehavior(QAbstractItemView.SelectRows)
table.setSelectionMode(QAbstractItemView.SingleSelection)
table.verticalHeader().setVisible(False)
table.horizontalHeader().setStretchLastSection(True)
table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
self.map_table = table
page_layout.addWidget(table, 1)
self.tabs.addTab(page, "Map")
def show_map_context_menu(self, position: Any) -> None:
if self.map_web_view is None:
return
menu = QMenu(self)
refresh_action = menu.addAction("Refresh")
selected_action = menu.exec(self.map_web_view.mapToGlobal(position))
if selected_action == refresh_action:
self.refresh_map_data(force=True)
def update_refresh_progress(self) -> None:
remaining_ms = int(max(0.0, self.next_refresh_at - time.monotonic()) * 1000)
self.refresh_progress.setValue(min(self.refresh_interval_ms, remaining_ms))
def init_lastlog(self) -> None:
if not args.lastlog:
return
try:
with open(args.lastlog, "a", encoding="utf-8") as handle:
started = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
handle.write(f"# ipsc3qt 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 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(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 refresh_data(self, force: bool = False) -> None:
if self.update_in_progress:
return
if self.fetch_in_progress and not force:
return
self.fetch_in_progress = True
self.next_refresh_at = time.monotonic() + args.refresh
self.update_refresh_progress()
if force:
self.timer.start(self.refresh_interval_ms)
def worker() -> None:
try:
data = fetch_json(args.api_url, args.timeout)
except VersionMismatchError as exc:
self.bridge.version_error.emit(str(exc), exc.server_version or "", exc.api_url or args.api_url)
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, json.JSONDecodeError, OSError) as exc:
self.bridge.fetch_error.emit(str(exc))
else:
self.bridge.data_ready.emit(data)
threading.Thread(target=worker, daemon=True).start()
def selected_map_hours(self) -> int:
if not self.map_hours_combo:
return 24
value = self.map_hours_combo.currentData()
return as_int(value, 24)
def selected_map_query(self) -> str:
if not self.map_query_input:
return ""
return self.map_query_input.text().strip()
def reset_map_search(self) -> None:
if self.map_query_input:
self.map_query_input.setText("")
self.refresh_map_data(force=True)
def search_map_callsign(self, query: str) -> None:
query = str(query or "").strip()
if not query:
return
if self.map_query_input:
self.map_query_input.setText(query)
self.refresh_map_data(force=True)
def on_map_live_toggled(self, checked: bool) -> None:
self.map_live_enabled = bool(checked)
interval = MAP_LIVE_REFRESH_SECONDS if self.map_live_enabled else MAP_REFRESH_SECONDS
self.map_timer.start(int(interval * 1000))
self.render_map_live_mode()
self.refresh_map_data(force=True)
def render_map_live_mode(self) -> None:
if self.map_web_view is None:
return
enabled = "true" if self.map_live_enabled else "false"
self.map_web_view.page().runJavaScript(
f"if (window.setLiveMap) {{ window.setLiveMap({enabled}); }}"
)
def refresh_map_data(self, force: bool = False) -> None:
if self.update_in_progress:
return
if self.map_fetch_in_progress and not force:
return
self.map_fetch_in_progress = True
query = self.selected_map_query()
hours = self.selected_map_hours()
if "Map" in self.summaries:
label = f"Loading GPS map for {query or 'latest positions'} over {hours} hours..."
self.summaries["Map"].setText(label)
def worker() -> None:
try:
data = fetch_map_json(args.api_url, query, hours, args.timeout)
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, json.JSONDecodeError, OSError) as exc:
self.bridge.map_fetch_error.emit(str(exc))
else:
self.bridge.map_data_ready.emit(data)
threading.Thread(target=worker, daemon=True).start()
def on_map_data_ready(self, data: dict[str, Any]) -> None:
self.map_fetch_in_progress = False
latest = data.get("latest") if isinstance(data.get("latest"), list) else []
tracks = data.get("tracks") if isinstance(data.get("tracks"), list) else []
track_points = 0
for track in tracks:
if isinstance(track, dict) and isinstance(track.get("points"), list):
track_points += len(track.get("points") or [])
ids = data.get("resolvedRadioIds") if isinstance(data.get("resolvedRadioIds"), list) else []
parts = [f"Latest {len(latest)}", f"Track points {track_points}", f"Window {data.get('hours', '-')}h"]
live_activity = data.get("liveActivity") if isinstance(data.get("liveActivity"), list) else []
if self.map_live_enabled:
parts.append(f"Live RX {len(live_activity)}")
if ids:
parts.append("IDs " + ", ".join(str(item) for item in ids))
if data.get("truncated"):
parts.append("track capped at 2000 points")
if "Map" in self.summaries:
self.summaries["Map"].setText(" | ".join(parts))
if self.map_web_view is not None:
self.last_map_data = data
self.render_map_web_data()
elif self.map_table is not None:
self.populate_map_table(data)
def render_map_web_data(self) -> None:
if self.map_web_view is None or self.last_map_data is None:
return
payload = json.dumps(self.last_map_data, separators=(",", ":"))
enabled = "true" if self.map_live_enabled else "false"
self.map_web_view.page().runJavaScript(
f"if (window.setLiveMap) {{ window.setLiveMap({enabled}); }} "
f"if (window.renderGpsMap) {{ window.renderGpsMap({payload}); }}"
)
def on_map_fetch_error(self, message: str) -> None:
self.map_fetch_in_progress = False
if "Map" in self.summaries:
self.summaries["Map"].setText(f"Map API error: {message}")
def populate_map_table(self, data: dict[str, Any]) -> None:
if self.map_table is None:
return
rows: list[list[Any]] = []
latest = data.get("latest") if isinstance(data.get("latest"), list) else []
for point in latest:
if not isinstance(point, dict):
continue
rows.append(
[
"Latest",
point.get("callsign") or point.get("radioId") or "-",
point.get("radioId") or "-",
gps_point_position(point),
gps_point_speed(point),
gps_point_course(point),
fmt_epoch(point.get("receivedAtEpoch")),
]
)
tracks = data.get("tracks") if isinstance(data.get("tracks"), list) else []
for track in tracks:
if not isinstance(track, dict):
continue
points = track.get("points") if isinstance(track.get("points"), list) else []
if not points:
continue
last_point = points[-1] if isinstance(points[-1], dict) else {}
rows.append(
[
f"Track {len(points)} pts",
track.get("callsign") or track.get("radioId") or "-",
track.get("radioId") or "-",
gps_point_position(last_point),
gps_point_speed(last_point),
gps_point_course(last_point),
fmt_epoch(last_point.get("receivedAtEpoch")),
]
)
table = self.map_table
table.setSortingEnabled(False)
table.clear()
table.setColumnCount(7)
table.setHorizontalHeaderLabels(["Type", "Call", "Radio ID", "Position", "Speed", "Course", "Seen Local"])
table.setRowCount(len(rows))
for row_index, row in enumerate(rows):
for col_index, value in enumerate(row):
item = QTableWidgetItem(as_text(value))
item.setTextAlignment(Qt.AlignLeft | Qt.AlignVCenter)
table.setItem(row_index, col_index, item)
table.resizeColumnsToContents()
table.setSortingEnabled(True)
def on_data_ready(self, data: dict[str, Any]) -> None:
self.fetch_in_progress = False
self.retry_button.setVisible(False)
self.meta_label.setVisible(False)
self.meta_label.setText("")
self.data = data
self.update_lastlog(data.get("lastHeard"))
generated = fmt_epoch(data.get("generatedAtEpoch"))
if generated == "-":
generated = "not loaded"
self.updated_label.setText(f"Updated {generated} {local_tz_label()}")
status_parts = [f"IPSC3Qt {IPSC3QT_VERSION}", f"API {data.get('version', 'unknown')}"]
if args.lastlog:
status_parts.append(f"lastlog error: {self.lastlog_error}" if self.lastlog_error else f"lastlog {args.lastlog}")
self.status_version_label.setToolTip(" | ".join(status_parts))
self.populate_all()
def on_fetch_error(self, message: str) -> None:
self.fetch_in_progress = False
self.retry_button.setVisible(True)
self.meta_label.setText(f"API {args.api_url}")
self.meta_label.setVisible(True)
self.status_version_label.setToolTip(f"API error: {message}")
def on_version_error(self, message: str, server_version: str, api_url: str) -> None:
self.fetch_in_progress = False
if self.update_in_progress:
return
self.update_in_progress = True
self.timer.stop()
self.retry_button.setVisible(False)
self.meta_label.setText(f"API {api_url}")
self.meta_label.setVisible(True)
if not server_version:
QMessageBox.critical(self, "IPSC3Qt Version Error", message)
self.status_version_label.setToolTip("Version error. Automatic refresh paused.")
return
result = QMessageBox.question(
self,
"IPSC3Qt Version Mismatch",
f"{message}\n\nDownload and replace ipsc3qt with {server_version} now?",
)
if result != QMessageBox.Yes:
self.status_version_label.setToolTip("Update required. Automatic refresh paused.")
return
try:
path = self_update(api_url, server_version, args.timeout)
except Exception as exc:
self.update_in_progress = False
self.retry_button.setVisible(True)
QMessageBox.critical(
self,
"IPSC3Qt Update Failed",
f"Automatic update failed: {exc}\n\n{upgrade_instructions(api_url)}",
)
self.status_version_label.setToolTip("Update failed. Use Retry after fixing the install path or permissions.")
return
QMessageBox.information(
self,
"IPSC3Qt Updated",
f"ipsc3qt updated to {server_version}:\n{path}\n\nThe app will restart now.",
)
self.restart_after_update(path)
def restart_after_update(self, script_path: str) -> None:
self.status_version_label.setToolTip("Restarting with updated ipsc3qt...")
QApplication.processEvents()
try:
restart_python = sys.executable
if not restart_python or not Path(restart_python).exists():
restart_python = shutil.which("python") or shutil.which("py") or ""
if not restart_python:
raise OSError("could not find a Python interpreter for restart")
argv = [restart_python, script_path, *sys.argv[1:]]
if os.name == "nt":
subprocess.Popen(argv, cwd=os.getcwd(), env=os.environ.copy())
QApplication.quit()
return
os.execv(restart_python, argv)
except OSError as exc:
QMessageBox.critical(
self,
"IPSC3Qt Restart Failed",
f"The update completed but restart failed: {exc}\n\nStart ipsc3qt manually.",
)
QApplication.quit()
def set_table(self, name: str, headers: list[str], rows: list[list[Any]]) -> None:
table = self.tables[name]
table.setSortingEnabled(False)
table.clear()
table.setColumnCount(len(headers))
table.setHorizontalHeaderLabels(headers)
table.setRowCount(len(rows))
for row_index, row in enumerate(rows):
for col_index, value in enumerate(row):
item = QTableWidgetItem(as_text(value))
item.setTextAlignment(Qt.AlignLeft | Qt.AlignVCenter)
text = as_text(value)
if text in ("OK", "Talking", "TX", "RX", "LK", "Listening"):
item.setForeground(QColor("#7af2db"))
elif text in ("Down", "DOWN"):
item.setForeground(QColor("#ff6b6b"))
elif text in ("Scan", "DISABLED", "NS"):
item.setForeground(QColor("#ffd166"))
table.setItem(row_index, col_index, item)
table.resizeColumnsToContents()
table.setSortingEnabled(True)
def populate_all(self) -> None:
self.populate_overview()
self.populate_last_heard()
self.populate_roaming()
self.populate_repeaters()
self.populate_hotspots()
self.populate_trunks()
self.populate_talkgroups()
def populate_overview(self) -> None:
overview = self.data.get("overview") or {}
self.summaries["Overview"].setText(
f"Connections {overview.get('connectedPeers', 0)} / {overview.get('peers', 0)} | "
f"Talkgroups {overview.get('talkgroups', 0)} | "
f"Active {overview.get('activeTalkgroups', 0)}"
)
rows = []
for row in overview.get("boardRows") or []:
if not isinstance(row, dict):
continue
age = as_int(row.get("age_seconds"), 999)
rows.append(
[
row.get("talkgroup"),
row.get("name"),
row.get("active_timeslot") or row.get("configured_slot") or "-",
"UP" if age <= ACTIVE_TALKGROUP_WINDOW_SECONDS else "IDLE",
speaker_label(row),
row.get("peer_callsign") or row.get("peer_kind") or "-",
row.get("stream_id"),
]
)
self.set_table("Overview", ["TG", "Name", "TS", "State", "Speaker", "Peer", "Stream"], rows)
def populate_last_heard(self) -> None:
rows = []
for row in self.data.get("lastHeard") or []:
if not isinstance(row, dict):
continue
rows.append(
[
row.get("radio_id"),
row.get("radio_callsign") or "-",
row.get("radio_name") or "-",
row.get("peer_callsign") or row.get("peer_kind") or row.get("protocol"),
destination_label(row),
row.get("call_type") or "-",
fmt_transfer(row),
fmt_epoch(row.get("last_seen_at_epoch")),
]
)
self.summaries["Last Heard"].setText(f"Most recent {len(rows)} radios")
self.set_table(
"Last Heard",
["Radio ID", "Call", "Name", "Seen Via", "Destination", "Type", "Dur/Data", "Last Seen Local"],
rows,
)
def populate_roaming(self) -> None:
roaming = self.data.get("roaming") or {}
raw_rows = roaming.get("rows") or [] if isinstance(roaming, dict) else []
dwell_hours = as_int(roaming.get("dwellHours"), 0) if isinstance(roaming, dict) else 0
rows = []
for row in raw_rows:
if not isinstance(row, dict):
continue
rows.append(
[
roaming_radio_flag(row),
roaming_radio_id_label(row),
row.get("radio_callsign") or "-",
truncate_label(row.get("radio_name"), 18),
roaming_location_type(row),
roaming_location_name(row),
roaming_gps_position(row),
fmt_time(row.get("last_seen_at_epoch")),
]
)
self.summaries["Roaming"].setText(
f"Roaming {len(rows)} radios inside {dwell_hours} hour dwell window. "
"If a radio is not shown here the caller will receive a busy signal."
)
self.set_table("Roaming", ["Radio", "Radio ID", "Call", "Name", "Location", "Location Name", "Radio-GPS Position", "Last Seen"], rows)
table = self.tables["Roaming"]
for row_index, row in enumerate(raw_rows):
if not isinstance(row, dict):
continue
item = table.item(row_index, 0)
if item is None:
continue
info = roaming_motorola_info(row)
if info:
item.setToolTip(info)
item.setForeground(QColor("#7af2db"))
continue
info = roaming_opengd77_info(row)
if info:
item.setToolTip(info)
item.setForeground(QColor("#8fcaff"))
continue
info = roaming_anytone_aprs_gps_info(row)
if info:
item.setToolTip(info)
item.setForeground(QColor("#8fcaff"))
def populate_repeaters(self) -> None:
count = len(self.data.get("repeaters") or [])
self.summaries["Repeaters"].setText(f"Repeaters {count}")
self.set_table(
"Repeaters",
["Call", "Name", "Flag", "Status", "Slot", "State", "Talkgroup", "Speaker", "Last Seen", "RX/TX"],
repeater_slot_rows(self.data),
)
def populate_hotspots(self) -> None:
count = len(self.data.get("hotspots") or [])
self.summaries["Hotspots"].setText(f"Hotspots {count}")
self.set_table(
"Hotspots",
["Call", "Kind", "Status", "Slot", "State", "Talkgroup", "Speaker", "BER/RSSI", "RX/TX"],
hotspot_slot_rows(self.data),
)
def populate_trunks(self) -> None:
rows = []
for row in self.data.get("trunks") or []:
if not isinstance(row, dict):
continue
state = "DISABLED" if not as_int(row.get("enabled"), 0) else ("OK" if as_int(row.get("connected"), 0) else "DOWN")
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 "-"
rows.append(
[
row.get("callsign") or row.get("display_name"),
row.get("kind"),
state,
endpoint,
f"{fmt_packets(row.get('rx_packets'))}/{fmt_packets(row.get('tx_packets'))}",
fmt_epoch(row.get("last_seen_at_epoch")),
activity,
]
)
self.summaries["Trunks"].setText(f"Trunks {len(rows)}")
self.set_table("Trunks", ["Name", "Kind", "State", "Endpoint", "RX/TX", "Last Seen Local", "Activity"], rows)
def populate_talkgroups(self) -> None:
rows = []
for row in self.data.get("talkgroups") or []:
if not isinstance(row, dict):
continue
rows.append(
[
row.get("talkgroup"),
row.get("name") or "-",
row.get("default_slot") or "-",
row.get("routing_mode") or "-",
row.get("trunk_summary") or "-",
"Yes" if row.get("listen_only") else "No",
talkgroup_timeout_label(row),
talkgroup_active_label(row),
]
)
self.summaries["Talkgroups"].setText(f"Talkgroups {len(rows)}")
self.set_table("Talkgroups", ["TG", "Name", "TS", "Routing", "Trunks", "Listen", "Timeout", "Active"], rows)
app = QApplication(sys.argv)
app.setApplicationName(PROCESS_NAME)
app.setApplicationVersion(IPSC3QT_VERSION)
palette = app.palette()
palette.setColor(QPalette.Window, QColor("#071116"))
palette.setColor(QPalette.WindowText, QColor("#eaf7f5"))
palette.setColor(QPalette.Base, QColor("#08151a"))
palette.setColor(QPalette.AlternateBase, QColor("#0b1b22"))
palette.setColor(QPalette.Text, QColor("#eaf7f5"))
palette.setColor(QPalette.Button, QColor("#1c4b56"))
palette.setColor(QPalette.ButtonText, QColor("#ffffff"))
app.setPalette(palette)
window = MainWindow()
window.show()
return app.exec()
def main() -> int:
set_process_name()
args = parse_args()
if args.dump:
try:
dump(fetch_json(args.api_url, args.timeout), args.dump)
return 0
except VersionMismatchError as exc:
print(str(exc), file=sys.stderr)
return 2
except Exception as exc:
print(f"ipsc3qt: {exc}", file=sys.stderr)
return 1
try:
return run_gui(args)
except subprocess.CalledProcessError as exc:
print(f"ipsc3qt: dependency install failed: {exc}", file=sys.stderr)
return 1
except Exception as exc:
print(f"ipsc3qt: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())