https://pastein.ru/t/yrg
Загрузка данных
"""Switch Controller — управление переключением каналов по TCP.
Зависимости: pip install PyQt6
Необязательно: pip install pynput (глобальные горячие клавиши ← →)
"""
from __future__ import annotations
import asyncio
import html
import ipaddress
import json
import os
import re
import socket
import sys
import tempfile
import threading
import subprocess
import platform
import time
import webbrowser
from dataclasses import dataclass, field, replace
from datetime import datetime
from pathlib import Path
from PyQt6.QtCore import (
QEvent, QMimeData, QObject, QPoint, QPointF, QRect, QSize, Qt, QTimer, pyqtSignal,
)
from PyQt6.QtGui import (
QBrush, QColor, QDrag, QIntValidator, QKeySequence, QPainter, QPen, QPixmap, QPolygonF, QShortcut, QWheelEvent,
)
from PyQt6.QtNetwork import QAbstractSocket, QTcpSocket
from PyQt6.QtWidgets import (
QAbstractItemView, QHeaderView, QMenu, QProgressBar, QTableWidget, QTableWidgetItem,
)
from PyQt6.QtWidgets import (
QApplication, QButtonGroup, QCheckBox, QComboBox, QDialog, QDialogButtonBox,
QFrame, QGridLayout, QHBoxLayout, QLabel, QLayout, QLineEdit, QListWidget, QListWidgetItem,
QMessageBox, QPushButton, QScrollArea, QSizePolicy, QTextEdit, QVBoxLayout, QWidget, QWidgetItem
)
try:
from pynput import keyboard as pynput_keyboard
except ImportError:
pynput_keyboard = None
# ============================================================================
# КОНСТАНТЫ И НАСТРОЙКИ РЕЛИЗА
# ============================================================================
CONFIG_NAME = "switch_config_pro.json"
CONFIG_PATH = Path(os.path.expanduser("~")) / CONFIG_NAME
PASSWORD = "1337"
TEST_MODE = False
UI_SCALE = 0.80
CH_MIN, CH_MAX = 1, 60
CONNECT_TIMEOUT_MS = 3000
DEBOUNCE_TIMEOUT = 1.1 # Отсечка стрелок 1.1 сек
QWIDGETSIZE_MAX = 16777215
SCHEMES = {
"SH5expLB": 16, "SH5expVzor": 17, "SH6HB": 18, "SH6LB": 19, "Sparrow": 20,
"LNA8_0930_3060": 21, "SW8_1_01_8": 22, "VzorSH6_HX": 23, "SH5HB_C549": 24,
"SH5LMB_C549": 25, "SH5Vzor_C549": 26, "SH6M": 27, "SH7V": 28,
"SH4PCM_SH6_W2": 29, "SH4PCM_SH6_W1": 30, "SH4PCM_SH6_W3": 31,
"SH7_C_MB_VRNJ": 32, "SH7_C_LB_VRNJ": 33, "SH7_V_VRNJ": 34,
}
DEFAULT_PRESETS = {
"Дефолт": list(range(1, 61)),
"F25": [13, 19, 14, 20],
"F26": [1, 6, 12, 18],
"Test": [13, 19],
}
THEMES = {
"graphite": ("Графит", {
"BG": "#121418", "CARD": "#1a1d23", "INPUT": "#0f1115",
"BORDER": "#2a2e37", "BORDERHI": "#3b414d",
"BTN": "#232731", "BTNHI": "#2d323e",
"TEXT": "#e8eaef", "DIM": "#939aa8", "MUTED": "#5c6370",
"ACCENT": "#4f8cff", "ACCENTHI": "#6fa0ff", "ON_ACCENT": "#0b0d12",
"OK": "#3ddc97", "WARN": "#f5a524", "BAD": "#ff6b6b",
"WHITE_BTN": "#e8eaef", "WHITE_BTN_HI": "#ffffff", "WHITE_BTN_TXT": "#14161b", "WHITE_BTN_PRESS": "#c9ced8",
}),
"ash": ("Пепел", {
"BG": "#2a2b2a", "CARD": "#323332", "INPUT": "#232423",
"BORDER": "#454645", "BORDERHI": "#5a5b5a",
"BTN": "#3b3c3b", "BTNHI": "#474847",
"TEXT": "#ececec", "DIM": "#a4a5a4", "MUTED": "#757675",
"ACCENT": "#6ea8fe", "ACCENTHI": "#8dbcff", "ON_ACCENT": "#101210",
"OK": "#4cd98f", "WARN": "#f2b04a", "BAD": "#ff7676",
"WHITE_BTN": "#ececec", "WHITE_BTN_HI": "#ffffff", "WHITE_BTN_TXT": "#222322", "WHITE_BTN_PRESS": "#cdcecd",
}),
"charcoal": ("Уголь", {
"BG": "#0b0b0c", "CARD": "#141415", "INPUT": "#0a0a0b",
"BORDER": "#27272a", "BORDERHI": "#3a3a3f",
"BTN": "#1d1d20", "BTNHI": "#27272b",
"TEXT": "#ececee", "DIM": "#8e8e96", "MUTED": "#58585f",
"ACCENT": "#8b5cf6", "ACCENTHI": "#a78bfa", "ON_ACCENT": "#ffffff",
"OK": "#4ade80", "WARN": "#fb923c", "BAD": "#f87171",
"WHITE_BTN": "#ececee", "WHITE_BTN_HI": "#ffffff", "WHITE_BTN_TXT": "#0b0b0c", "WHITE_BTN_PRESS": "#c4c4c8",
}),
"light": ("Светлая", {
"BG": "#f0f0f0", "CARD": "#ffffff", "INPUT": "#f9f9f9",
"BORDER": "#cccccc", "BORDERHI": "#bbbbbb",
"BTN": "#e0e0e0", "BTNHI": "#d0d0d0",
"TEXT": "#111111", "DIM": "#666666", "MUTED": "#999999",
"ACCENT": "#0078d7", "ACCENTHI": "#005a9e", "ON_ACCENT": "#ffffff",
"OK": "#107c10", "WARN": "#d83b01", "BAD": "#a80000",
"WHITE_BTN": "#333333", "WHITE_BTN_HI": "#111111", "WHITE_BTN_TXT": "#ffffff", "WHITE_BTN_PRESS": "#000000",
}),
}
DEFAULT_THEME = "graphite"
QSS_TEMPLATE = """
QWidget { color: @TEXT; font-family: "Segoe UI", "Inter", "Helvetica Neue", sans-serif; font-size: 13px; }
QWidget#root, QDialog { background-color: @BG; }
QLabel { background: transparent; }
QLabel#cardTitle { color: @DIM; font-size: 11px; font-weight: 700; }
QLabel#dim { color: @DIM; }
QLabel#value { color: @TEXT; font-weight: 600; }
QLabel#hint { color: @WARN; }
QLabel#status { font-weight: 700; }
QFrame#card { background-color: @CARD; border: 1px solid @BORDER; border-radius: 10px; }
QFrame#chipBox { background-color: @INPUT; border: 1px solid @BORDER; border-radius: 6px; }
QLineEdit, QSpinBox, QComboBox {
background-color: @INPUT; border: 1px solid @BORDER; border-radius: 6px;
padding: 4px 8px; min-height: 18px; selection-background-color: @ACCENT; selection-color: @ON_ACCENT;
}
QLineEdit:focus, QSpinBox:focus, QComboBox:focus { border-color: @ACCENT; }
QLineEdit:disabled, QSpinBox:disabled, QComboBox:disabled { color: @MUTED; }
QLineEdit#big {
font-size: 22px; font-weight: 700; padding: 0px;
border: 1px solid @BORDER; background-color: @INPUT; border-radius: 6px;
}
QLineEdit#big:focus { border-color: @ACCENT; }
QFrame#tagField { background-color: @INPUT; border: 1px solid @BORDER; border-radius: 6px; }
QFrame#tagField[focused="true"] { border-color: @ACCENT; }
QLineEdit#tagInput, QLineEdit#tagInput:focus { background: transparent; border: none; padding: 2px 4px; min-height: 22px; }
QLabel#tag {
background-color: @BTN; border: 1px solid @BORDERHI; border-radius: 12px;
padding: 3px 11px; font-weight: 700; color: @TEXT;
}
QLabel#tag:hover { border-color: @BAD; color: @BAD; }
QFrame#dropLine { background-color: @ACCENT; border: none; }
QComboBox::drop-down { border: none; width: 22px; }
QComboBox::down-arrow { image: url("@ARROW"); width: 12px; height: 8px; margin-right: 8px; }
QComboBox QAbstractItemView {
background-color: @CARD; border: 1px solid @BORDER; outline: 0;
selection-background-color: @ACCENT; selection-color: @ON_ACCENT;
}
QPushButton {
background-color: @BTN; border: 1px solid @BORDER; border-radius: 6px;
padding: 4px 12px; font-weight: 600; min-height: 18px; font-size: 12px;
}
QPushButton:hover { background-color: @BTNHI; border-color: @BORDERHI; }
QPushButton:pressed { background-color: @CARD; }
QPushButton:disabled { color: @MUTED; background-color: @CARD; border-color: @BORDER; }
QPushButton[variant="white"] {
background-color: @WHITE_BTN; border-color: @WHITE_BTN; color: @WHITE_BTN_TXT;
}
QPushButton[variant="white"]:hover { background-color: @WHITE_BTN_HI; border-color: @WHITE_BTN_HI; }
QPushButton[variant="white"]:pressed { background-color: @WHITE_BTN_PRESS; border-color: @WHITE_BTN_PRESS; }
QPushButton[variant="white"]:disabled { background-color: @CARD; border-color: @BORDER; color: @MUTED; }
QPushButton[variant="danger"] { color: @BAD; }
QPushButton[variant="danger"]:disabled { color: @MUTED; }
QPushButton#nav { font-size: 14px; padding: 0; min-width: 30px; min-height: 30px; }
QPushButton#ghost { background: transparent; border: none; color: @DIM; padding: 2px 6px; min-height: 16px; font-size: 12px; }
QPushButton#ghost:hover { color: @TEXT; }
QPushButton#chip {
background-color: @BTN; border: 1px solid @BORDER; border-radius: 13px;
padding: 3px 12px; min-height: 18px; font-weight: 700;
}
QPushButton#chip:hover { border-color: @ACCENT; }
QPushButton#chip:checked { background-color: @WHITE_BTN; border-color: @WHITE_BTN; color: @WHITE_BTN_TXT; }
QPushButton#chip:disabled { color: @MUTED; }
QPushButton#chip:checked:disabled { background-color: @BORDERHI; border-color: @BORDERHI; color: @MUTED; }
QPushButton#chipHighlight {
background-color: @BORDERHI; border: 1px solid @BORDER; border-radius: 13px;
padding: 3px 12px; min-height: 18px; font-weight: 800; color: @TEXT;
}
QPushButton#chipHighlight:hover { border-color: @ACCENT; }
QPushButton#chipHighlight:checked { background-color: @WHITE_BTN; border-color: @WHITE_BTN; color: @WHITE_BTN_TXT; }
QPushButton#chipHighlight:disabled { color: @MUTED; }
QPushButton#chipHighlight:checked:disabled { background-color: @BORDERHI; border-color: @BORDERHI; color: @MUTED; }
QCheckBox { spacing: 8px; color: @TEXT; font-size: 12px; }
QCheckBox:disabled { color: @MUTED; }
QCheckBox::indicator { width: 16px; height: 16px; border: 1px solid @BORDERHI; border-radius: 4px; background: @INPUT; }
QCheckBox::indicator:checked { background: @ACCENT; border-color: @ACCENT; image: url("@CHECK"); }
QListWidget { background-color: @INPUT; border: 1px solid @BORDER; border-radius: 6px; padding: 4px; outline: 0; }
QListWidget::item { padding: 7px 8px; border-radius: 4px; }
QListWidget::item:selected { background-color: @WHITE_BTN; color: @WHITE_BTN_TXT; }
QListWidget::item:hover:!selected { background-color: @BTN; }
QTextEdit#log {
background-color: @INPUT; border: 1px solid @BORDER; border-radius: 8px; padding: 6px;
font-family: "Cascadia Code", "Consolas", "Menlo", monospace; font-size: 12px; color: @TEXT;
}
QScrollBar:vertical { background: transparent; width: 10px; margin: 0; }
QScrollBar::handle:vertical { background: @BORDER; border-radius: 5px; min-height: 24px; }
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { height: 0; }
QScrollArea { border: none; background: transparent; }
QScrollArea > QWidget > QWidget { background: transparent; }
QTableWidget { background-color: @INPUT; border: 1px solid @BORDER; border-radius: 6px; outline: 0; }
QTableWidget::item { padding: 3px 8px; border: none; }
QTableWidget::item:selected { background-color: @WHITE_BTN; color: @WHITE_BTN_TXT; }
QHeaderView { background: transparent; }
QHeaderView::section {
background-color: @CARD; color: @DIM; border: none; border-bottom: 1px solid @BORDER;
border-right: 1px solid @BORDER; padding: 5px 8px; font-weight: 600;
}
QHeaderView::down-arrow { image: url("@ARROW"); width: 10px; height: 8px; }
QHeaderView::up-arrow { image: url("@ARROW_UP"); width: 10px; height: 8px; }
QTableCornerButton::section { background-color: @CARD; border: none; }
QProgressBar { background-color: @INPUT; border: 1px solid @BORDER; border-radius: 4px; max-height: 8px; }
QProgressBar::chunk { background-color: @ACCENT; border-radius: 3px; }
QMenu { background-color: @CARD; border: 1px solid @BORDER; padding: 4px; }
QMenu::item { padding: 6px 20px; border-radius: 4px; }
QMenu::item:selected { background-color: @WHITE_BTN; color: @WHITE_BTN_TXT; }
QMenu::item:disabled { color: @MUTED; }
QMenu::separator { height: 1px; background: @BORDER; margin: 4px 6px; }
QToolTip { background-color: @CARD; color: @TEXT; border: 1px solid @BORDER; padding: 4px; }
"""
def make_icons(palette: dict, theme: str) -> dict:
folder = Path(tempfile.gettempdir()) / "switch_controller_icons"
folder.mkdir(exist_ok=True)
arrow = QPixmap(12, 8)
arrow.fill(Qt.GlobalColor.transparent)
p = QPainter(arrow)
p.setRenderHint(QPainter.RenderHint.Antialiasing)
p.setPen(Qt.PenStyle.NoPen)
p.setBrush(QColor(palette["DIM"]))
p.drawPolygon(QPolygonF([QPointF(1, 1), QPointF(11, 1), QPointF(6, 7)]))
p.end()
check = QPixmap(16, 16)
check.fill(Qt.GlobalColor.transparent)
p = QPainter(check)
p.setRenderHint(QPainter.RenderHint.Antialiasing)
p.setPen(QPen(QColor(palette["ON_ACCENT"]), 2, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap, Qt.PenJoinStyle.RoundJoin))
p.drawPolyline(QPolygonF([QPointF(4, 8.5), QPointF(7, 11.5), QPointF(12, 5)]))
p.end()
arrow_up = QPixmap(12, 8)
arrow_up.fill(Qt.GlobalColor.transparent)
p = QPainter(arrow_up)
p.setRenderHint(QPainter.RenderHint.Antialiasing)
p.setPen(Qt.PenStyle.NoPen)
p.setBrush(QColor(palette["DIM"]))
p.drawPolygon(QPolygonF([QPointF(1, 7), QPointF(11, 7), QPointF(6, 1)]))
p.end()
arrow_path = folder / f"arrow_{theme}.png"
arrow_up_path = folder / f"arrow_up_{theme}.png"
check_path = folder / f"check_{theme}.png"
arrow.save(str(arrow_path))
check.save(str(check_path))
arrow_up.save(str(arrow_up_path))
return {"ARROW": arrow_path.as_posix(), "ARROW_UP": arrow_up_path.as_posix(), "CHECK": check_path.as_posix()}
def build_qss(palette: dict, icons: dict | None = None) -> str:
tokens = dict(palette)
tokens.update(icons or {"ARROW": "arrow.png", "ARROW_UP": "arrow_up.png", "CHECK": "check.png"})
qss = QSS_TEMPLATE
for key in sorted(tokens, key=len, reverse=True):
qss = qss.replace(f"@{key}", tokens[key])
return qss
def set_variant(widget: QWidget, variant: str) -> None:
widget.setProperty("variant", variant)
widget.style().unpolish(widget)
widget.style().polish(widget)
# ============================================================================
# Протокол устройства
# ============================================================================
CMD_SET_CHANNEL = 0x18
CMD_SET_SCHEME = 0x68
def build_set_packet(cmd: int, value: int) -> bytes:
now = datetime.now()
return bytes([0x06, 0x00, 0x00, 0x00, 0x08, now.hour, now.minute, 0x10, 0x01, cmd, value])
def build_uptime_packet() -> bytes:
now = datetime.now()
return bytes([0x05, 0x00, 0x00, 0x00, 0x08, now.hour, now.minute, 0xA0, 0x01, 0x01])
def parse_uptime(data: bytes) -> str | None:
return None
def hex_str(data: bytes) -> str:
return " ".join(f"{b:02X}" for b in data)
# ============================================================================
# Сетевой сканер (по мотивам Advanced IP Scanner)
# ============================================================================
MAC_RE = re.compile(r"(?<![0-9A-Fa-f])(?:[0-9A-Fa-f]{1,2}[-:]){5}[0-9A-Fa-f]{1,2}(?![0-9A-Fa-f])")
IP_RE = re.compile(r"(?<![\d.])(\d{1,3}(?:\.\d{1,3}){3})(?![\d.])")
RTT_RE = re.compile(r"(?:time|время)\s*([=<])\s*([\d.,]+)\s*(?:ms|мс)", re.I)
MAX_TARGETS = 65536
BUILTIN_OUI = {
"005056": "VMware", "000C29": "VMware", "000569": "VMware", "080027": "VirtualBox",
"00155D": "Microsoft (Hyper-V)", "525400": "QEMU/KVM",
"B827EB": "Raspberry Pi", "DCA632": "Raspberry Pi", "E45F01": "Raspberry Pi", "28CDC1": "Raspberry Pi",
}
def normalize_mac(text: str) -> str | None:
m = MAC_RE.search(text)
if not m:
return None
parts = re.split(r"[-:]", m.group(0))
mac = ":".join(p.zfill(2) for p in parts).upper()
return None if mac in ("00:00:00:00:00:00", "FF:FF:FF:FF:FF:FF") else mac
def oui_lookup(mac: str) -> str:
if not mac:
return ""
key = mac.replace(":", "").replace("-", "").upper()[:6]
name = BUILTIN_OUI.get(key)
if name:
return name
try:
if int(key[:2], 16) & 0x02:
return "Случайный MAC"
except ValueError:
pass
return ""
class _RangeError(Exception):
pass
def parse_targets(text: str, limit: int = MAX_TARGETS) -> tuple[list[str], str]:
ips: list[str] = []
seen: set[str] = set()
ipv4 = r"\d{1,3}(?:\.\d{1,3}){3}"
def addr(t: str) -> ipaddress.IPv4Address:
try:
return ipaddress.IPv4Address(t)
except ValueError:
raise _RangeError(f"Не удалось разобрать адрес: {t}")
def add(lo: int, hi: int) -> None:
if hi < lo:
lo, hi = hi, lo
if hi - lo + 1 > limit:
raise _RangeError(f"Слишком большой диапазон (максимум {limit} адресов).")
for n in range(lo, hi + 1):
ip = str(ipaddress.IPv4Address(n))
if ip not in seen:
seen.add(ip)
ips.append(ip)
if len(ips) > limit:
raise _RangeError(f"Слишком много адресов (максимум {limit}).")
tokens = [t.strip() for t in re.split(r"[,;\n]+", text) if t.strip()]
if not tokens:
return [], "Введите диапазон адресов."
try:
for tok in tokens:
if "/" in tok:
try:
net = ipaddress.IPv4Network(tok, strict=False)
except ValueError:
raise _RangeError(f"Не удалось разобрать сеть: {tok}")
lo, hi = int(net.network_address), int(net.broadcast_address)
if net.prefixlen < 31:
lo, hi = lo + 1, hi - 1
add(lo, hi)
continue
m = re.fullmatch(rf"({ipv4})\s*-\s*({ipv4})", tok)
if m:
add(int(addr(m.group(1))), int(addr(m.group(2))))
continue
m = re.fullmatch(rf"({ipv4})\s*-\s*(\d{{1,3}})", tok)
if m:
a = int(addr(m.group(1)))
last = int(m.group(2))
if last > 255:
raise _RangeError(f"В «{tok}» последнее число больше 255.")
add(a, (a & ~255) | last)
continue
add(int(addr(tok)), int(addr(tok)))
except _RangeError as e:
return [], str(e)
return ips, ""
def detect_local_range() -> str:
ip = ""
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.connect(("8.8.8.8", 80))
ip = sock.getsockname()[0]
except OSError:
try:
ip = socket.gethostbyname(socket.gethostname())
except OSError:
return ""
if not ip or ip.startswith("127."):
return ""
return f"{ip.rsplit('.', 1)[0]}.1-254"
def read_arp_table() -> dict[str, str]:
table: dict[str, str] = {}
try:
system = platform.system()
if system == "Linux":
with open("/proc/net/arp", "r") as f:
next(f, None)
for line in f:
p = line.split()
if len(p) >= 4 and p[2] != "0x0":
mac = normalize_mac(p[3])
if mac:
table[p[0]] = mac
return table
kwargs = {"creationflags": getattr(subprocess, "CREATE_NO_WINDOW", 0)} if system == "Windows" else {}
out = subprocess.check_output(["arp", "-a"], text=True, errors="ignore",
stderr=subprocess.DEVNULL, timeout=5, **kwargs)
for line in out.splitlines():
m = IP_RE.search(line)
mac = normalize_mac(line)
if m and mac:
table.setdefault(m.group(1), mac)
except Exception:
pass
return table
def send_wol(mac: str) -> str:
try:
raw = bytes.fromhex(mac.replace(":", "").replace("-", ""))
packet = b"\xff" * 6 + raw * 16
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.sendto(packet, ("255.255.255.255", 9))
return ""
except (OSError, ValueError) as e:
return str(e)
@dataclass
class HostResult:
ip: str
alive: bool = False
name: str = ""
mac: str = ""
rtt: float | None = None
port_open: bool = False
note: str = ""
class NetworkScanner(QObject):
host_result = pyqtSignal(object)
progress = pyqtSignal(int, int)
finished = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self._loop = None
self._task = None
self._thread = None
self._running = False
@property
def running(self) -> bool:
return self._running
def start(self, ips: list[str], port: int) -> None:
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._run, args=(list(ips), port), daemon=True)
self._thread.start()
def stop(self) -> None:
self._running = False
loop, task = self._loop, self._task
if loop is not None and task is not None and loop.is_running():
loop.call_soon_threadsafe(task.cancel)
def wait(self, timeout: float = 3.0) -> None:
t = self._thread
if t is not None and t.is_alive() and t is not threading.current_thread():
t.join(timeout)
def _run(self, ips: list[str], port: int) -> None:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self._loop = loop
try:
self._task = loop.create_task(self._scan(ips, port))
loop.run_until_complete(self._task)
except asyncio.CancelledError:
pass
except Exception:
pass
finally:
try:
loop.run_until_complete(loop.shutdown_default_executor())
except Exception:
pass
loop.close()
self._loop = None
self._task = None
self._running = False
self.finished.emit()
async def _scan(self, ips: list[str], port: int) -> None:
total = len(ips)
done = 0
results: dict[str, HostResult] = {}
sem = asyncio.Semaphore(128)
async def probe(ip: str) -> None:
nonlocal done
async with sem:
if not self._running:
return
(alive, rtt), port_open = await asyncio.gather(self._ping(ip), self._tcp_open(ip, port))
r = HostResult(ip=ip, alive=alive or port_open, rtt=rtt, port_open=port_open,
note="" if alive else ("только порт" if port_open else ""))
results[ip] = r
done += 1
self.host_result.emit(replace(r))
self.progress.emit(done, total)
await asyncio.gather(*(probe(ip) for ip in ips))
if not self._running:
return
arp = await asyncio.to_thread(read_arp_table)
for ip, r in results.items():
mac = arp.get(ip)
if mac:
r.mac = mac
if not r.alive:
r.alive = True
r.note = "только ARP"
sem2 = asyncio.Semaphore(32)
async def finish(r: HostResult) -> None:
async with sem2:
try:
name = await asyncio.wait_for(asyncio.to_thread(self._get_hostname, r.ip), timeout=2.0)
except asyncio.TimeoutError:
name = None
r.name = name or ""
if self._running:
self.host_result.emit(replace(r))
await asyncio.gather(*(finish(r) for r in results.values() if r.alive))
@staticmethod
def _ping_command(ip: str) -> tuple[list[str], dict]:
system = platform.system()
if system == "Windows":
return ["ping", "-n", "1", "-w", "700", ip], {"creationflags": getattr(subprocess, "CREATE_NO_WINDOW", 0)}
if system == "Darwin":
return ["ping", "-c", "1", "-t", "1", ip], {}
return ["ping", "-c", "1", "-W", "1", ip], {}
@staticmethod
async def _kill(proc) -> None:
try:
if proc is not None and proc.returncode is None:
proc.kill()
await asyncio.wait_for(proc.wait(), timeout=1)
except (ProcessLookupError, asyncio.TimeoutError):
pass
async def _ping(self, ip: str) -> tuple[bool, float | None]:
cmd, kwargs = self._ping_command(ip)
proc = None
try:
proc = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, **kwargs)
out, _ = await asyncio.wait_for(proc.communicate(), timeout=3)
except asyncio.TimeoutError:
await self._kill(proc)
return False, None
except asyncio.CancelledError:
await self._kill(proc)
raise
except Exception:
return False, None
text = out.decode(errors="ignore")
if not re.search(r"ttl=", text, re.I):
return False, None
m = RTT_RE.search(text)
rtt = None
if m:
rtt = 0.5 if m.group(1) == "<" else float(m.group(2).replace(",", "."))
return True, rtt
@staticmethod
async def _tcp_open(ip: str, port: int) -> bool:
try:
_r, writer = await asyncio.wait_for(asyncio.open_connection(ip, port), timeout=0.5)
writer.close()
await writer.wait_closed()
return True
except Exception:
return False
@staticmethod
def _get_hostname(ip: str) -> str | None:
try:
return socket.gethostbyaddr(ip)[0]
except Exception:
return None
# ============================================================================
# Каналы и конфиг
# ============================================================================
def parse_channels(text: str) -> tuple[list[int], list[str]]:
ok: list[int] = []
bad: list[str] = []
for tok in re.split(r"[,;\s]+", text.strip()):
if not tok:
continue
nums: range | list[int] | None
m = re.fullmatch(r"(\d+)-(\d+)", tok)
if m:
lo, hi = int(m[1]), int(m[2])
nums = range(lo, hi + 1) if lo <= hi else None
elif tok.isascii() and tok.isdigit():
nums = [int(tok)]
else:
nums = None
if nums is None or any(not CH_MIN <= n <= CH_MAX for n in nums):
bad.append(tok)
continue
ok.extend(nums)
return ok, bad
def sanitize_channels(values) -> list[int]:
result: list[int] = []
if not isinstance(values, (list, tuple)):
return result
for v in values:
if isinstance(v, bool) or not isinstance(v, int):
continue
if CH_MIN <= v <= CH_MAX and v not in result:
result.append(v)
return result
def clean_preset_name(name: str) -> str:
return re.sub(r"\s*\(\s*\d[\d,\s]*\)\s*$", "", name).strip()
@dataclass
class Config:
ip: str = "192.168.100.44"
port: int = 40250
scheme: str = "SH7_C_MB_VRNJ"
preset: str = "Дефолт"
global_hotkeys: bool = False
theme: str = DEFAULT_THEME
presets: dict = field(default_factory=lambda: {k: list(v) for k, v in DEFAULT_PRESETS.items()})
highlight_channels: dict = field(default_factory=dict)
authorized: bool = False
@classmethod
def load(cls) -> tuple["Config", str | None]:
cfg = cls()
path = next((p for p in (CONFIG_PATH, Path.cwd() / CONFIG_NAME) if p.is_file()), None)
if path is None:
return cfg, None
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError) as e:
return cfg, f"Не удалось прочитать {path.name}: {e}. Взяты настройки по умолчанию."
if not isinstance(raw, dict):
return cfg, f"{path.name} имеет неверный формат. Взяты настройки по умолчанию."
ip = str(raw.get("ip", "")).strip()
if ip:
cfg.ip = ip
try:
port = int(raw.get("port", cfg.port))
if 1 <= port <= 65535:
cfg.port = port
except (TypeError, ValueError):
pass
if raw.get("scheme") in SCHEMES:
cfg.scheme = raw["scheme"]
cfg.preset = str(raw.get("preset", "Дефолт"))
cfg.global_hotkeys = bool(raw.get("global_hotkeys", False))
theme = raw.get("theme")
if theme in THEMES:
cfg.theme = theme
elif raw.get("dark_theme") is False:
cfg.theme = "light"
cfg.authorized = bool(raw.get("authorized", False))
if isinstance(raw.get("presets"), dict):
cleaned: dict = {}
for name, chans in raw["presets"].items():
name = clean_preset_name(str(name))
chans = sanitize_channels(chans)
if name and chans:
cleaned[name] = chans
if cleaned:
cfg.presets = cleaned
if isinstance(raw.get("highlight_channels"), dict):
cfg.highlight_channels = {
str(k): sanitize_channels(v) for k, v in raw["highlight_channels"].items()
}
return cfg, None
def save(self) -> str | None:
data = {
"ip": self.ip, "port": self.port, "scheme": self.scheme, "preset": self.preset,
"global_hotkeys": self.global_hotkeys, "theme": self.theme,
"presets": self.presets, "highlight_channels": self.highlight_channels,
"authorized": self.authorized,
}
tmp = CONFIG_PATH.with_name(CONFIG_NAME + ".tmp")
try:
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=4), encoding="utf-8")
tmp.replace(CONFIG_PATH)
except OSError as e:
return str(e)
return None
# ============================================================================
# Сеть
# ============================================================================
def _error_text(sock: QTcpSocket) -> str:
E = QAbstractSocket.SocketError
known = {
E.ConnectionRefusedError: "Соединение отклонено",
E.RemoteHostClosedError: "Соединение закрыто устройством",
E.HostNotFoundError: "Хост не найден",
E.SocketTimeoutError: "Таймаут соединения",
E.NetworkError: "Сетевая ошибка",
}
return known.get(sock.error(), sock.errorString())
class DeviceClient(QObject):
state_changed = pyqtSignal(object)
error = pyqtSignal(str)
data = pyqtSignal(bytes)
def __init__(self, parent=None):
super().__init__(parent)
self._sock = QTcpSocket(self)
self._sock.setSocketOption(QAbstractSocket.SocketOption.LowDelayOption, 1)
self._sock.stateChanged.connect(lambda state: self.state_changed.emit(state))
self._sock.errorOccurred.connect(lambda _e: self.error.emit(_error_text(self._sock)))
self._sock.readyRead.connect(lambda: self.data.emit(bytes(self._sock.readAll())))
self._timeout = QTimer(self)
self._timeout.setSingleShot(True)
self._timeout.timeout.connect(self._on_timeout)
@property
def state(self):
return self._sock.state()
@property
def is_connected(self) -> bool:
return self.state == QAbstractSocket.SocketState.ConnectedState
def open(self, host: str, port: int) -> None:
self._timeout.start(CONNECT_TIMEOUT_MS)
self._sock.connectToHost(host, port)
def close(self) -> None:
self._timeout.stop()
if self.state != QAbstractSocket.SocketState.UnconnectedState:
self._sock.abort()
def send(self, packet: bytes) -> bool:
if not self.is_connected:
return False
return self._sock.write(packet) == len(packet)
def _on_timeout(self) -> None:
if not self.is_connected:
self._sock.abort()
self.error.emit("Таймаут подключения")
class GlobalHotkeys(QObject):
pressed = pyqtSignal(str)
def __init__(self, parent=None):
super().__init__(parent)
self._listener = None
@staticmethod
def available() -> bool:
return pynput_keyboard is not None
def start(self) -> None:
if self._listener is not None or not self.available():
return
def on_press(key):
if key == pynput_keyboard.Key.right:
self.pressed.emit("next")
elif key == pynput_keyboard.Key.left:
self.pressed.emit("prev")
self._listener = pynput_keyboard.Listener(on_press=on_press)
self._listener.daemon = True
self._listener.start()
def stop(self) -> None:
if self._listener is not None:
self._listener.stop()
self._listener = None
# ============================================================================
# Виджеты
# ============================================================================
class FlowLayout(QLayout):
def __init__(self, parent=None, hspacing: int = 6, vspacing: int = 6):
self._items: list = []
self._h, self._v = hspacing, vspacing
super().__init__(parent)
self.setContentsMargins(0, 0, 0, 0)
def addItem(self, item):
self._items.append(item)
self.invalidate()
def insertWidget(self, index: int, widget: QWidget) -> None:
self.addChildWidget(widget)
self._items.insert(index, QWidgetItem(widget))
widget.show()
self.invalidate()
def reorder(self, widgets: list) -> None:
ordered = []
for w in widgets:
for it in self._items:
if it.widget() is w:
ordered.append(it)
break
used = {id(it) for it in ordered}
self._items = ordered + [it for it in self._items if id(it) not in used]
self.invalidate()
def count(self):
return len(self._items)
def itemAt(self, index):
return self._items[index] if 0 <= index < len(self._items) else None
def takeAt(self, index):
if 0 <= index < len(self._items):
item = self._items.pop(index)
self.invalidate()
return item
return None
def expandingDirections(self):
return Qt.Orientation.Horizontal & Qt.Orientation.Vertical
def hasHeightForWidth(self):
return True
def heightForWidth(self, width):
return self._do_layout(QRect(0, 0, width, 0), True)
def setGeometry(self, rect):
super().setGeometry(rect)
self._do_layout(rect, False)
def sizeHint(self):
return self.minimumSize()
def minimumSize(self):
size = QSize()
for item in self._items:
size = size.expandedTo(item.minimumSize())
m = self.contentsMargins()
return size + QSize(m.left() + m.right(), m.top() + m.bottom())
def _do_layout(self, rect: QRect, test_only: bool) -> int:
m = self.contentsMargins()
r = rect.adjusted(m.left(), m.top(), -m.right(), -m.bottom())
x, y, line_h = r.x(), r.y(), 0
for item in self._items:
hint = item.sizeHint()
next_x = x + hint.width()
if next_x > r.right() + 1 and line_h > 0:
x = r.x()
y += line_h + self._v
next_x = x + hint.width()
line_h = 0
if not test_only:
item.setGeometry(QRect(QPoint(x, y), hint))
x = next_x + self._h
line_h = max(line_h, hint.height())
return y + line_h - rect.y() + m.bottom()
class ChipStrip(QWidget):
chipClicked = pyqtSignal(int)
def __init__(self, parent=None):
super().__init__(parent)
self._flow = FlowLayout(self)
self._group = QButtonGroup(self)
self._group.setExclusive(True)
self._buttons: list[QPushButton] = []
self._highlighted: set[int] = set()
def set_channels(self, channels: list[int], highlighted: set[int] | None = None) -> None:
self._highlighted = highlighted or set()
for b in self._buttons:
self._group.removeButton(b)
b.setParent(None)
b.deleteLater()
self._buttons.clear()
for i, ch in enumerate(channels):
b = QPushButton(str(ch))
if ch in self._highlighted:
b.setObjectName("chipHighlight")
else:
b.setObjectName("chip")
b.setCheckable(True)
b.setCursor(Qt.CursorShape.PointingHandCursor)
b.clicked.connect(lambda _=False, idx=i: self.chipClicked.emit(idx))
self._group.addButton(b)
self._flow.addWidget(b)
self._buttons.append(b)
self.updateGeometry()
def row_height(self) -> int:
h = max((b.sizeHint().height() for b in self._buttons), default=0)
return h or 30
def set_active(self, index: int) -> None:
self._group.setExclusive(False)
for i, b in enumerate(self._buttons):
b.setChecked(i == index)
self._group.setExclusive(True)
class ChipScroll(QScrollArea):
MAX_ROWS = 4
def __init__(self, strip: ChipStrip, parent=None):
super().__init__(parent)
self._strip = strip
self._last_hint = -1
self.setWidget(strip)
self.setWidgetResizable(True)
self.setFrameShape(QFrame.Shape.NoFrame)
self.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred)
def minimumSizeHint(self) -> QSize:
return QSize(80, self._strip.row_height())
def sizeHint(self) -> QSize:
width = max(self.viewport().width(), 200)
row = self._strip.row_height()
cap = row * self.MAX_ROWS + 6 * (self.MAX_ROWS - 1)
content = self._strip.heightForWidth(width)
if content < 0:
content = row
return QSize(width, max(row, min(content, cap)))
def resizeEvent(self, event) -> None:
super().resizeEvent(event)
h = self.sizeHint().height()
if h != self._last_hint:
self._last_hint = h
self.updateGeometry()
class IntField(QLineEdit):
def __init__(self, lo: int, hi: int, value: int, parent=None):
super().__init__(parent)
self._lo, self._hi, self._last = lo, hi, value
self.setValidator(QIntValidator(lo, hi, self))
self.setText(str(value))
self.editingFinished.connect(lambda: self.setText(str(self.value())))
def value(self) -> int:
try:
v = int(self.text())
except ValueError:
return self._last
self._last = min(max(v, self._lo), self._hi)
return self._last
def setValue(self, v: int) -> None:
self._last = min(max(int(v), self._lo), self._hi)
self.setText(str(self._last))
def wheelEvent(self, event: QWheelEvent) -> None:
delta = event.angleDelta().y()
if delta > 0:
self.setValue(self.value() + 1)
elif delta < 0:
self.setValue(self.value() - 1)
event.accept()
TAG_MIME = "application/x-switch-tag"
class TagChip(QLabel):
removeRequested = pyqtSignal(object)
dragRequested = pyqtSignal(object)
highlightToggled = pyqtSignal(object)
def __init__(self, ch: int, highlighted: bool = False):
super().__init__(str(ch))
self.setObjectName("tag")
self.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.setToolTip("Двойной клик — убрать, перетащите — изменить порядок, ПКМ — подсветить")
self._press = None
self._highlighted = highlighted
self._update_style()
def set_highlighted(self, on: bool) -> None:
if on != self._highlighted:
self._highlighted = on
self._update_style()
def _update_style(self) -> None:
if self._highlighted:
self.setStyleSheet("background-color: #4f8cff; color: white; border: 1px solid #4f8cff;")
else:
self.setStyleSheet("")
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.RightButton:
self.highlightToggled.emit(self)
elif event.button() == Qt.MouseButton.LeftButton:
self._press = event.position().toPoint()
super().mousePressEvent(event)
def mouseMoveEvent(self, event):
if self._press is not None and event.buttons() & Qt.MouseButton.LeftButton:
if (event.position().toPoint() - self._press).manhattanLength() >= QApplication.startDragDistance():
self._press = None
self.dragRequested.emit(self)
return
super().mouseMoveEvent(event)
def mouseDoubleClickEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self._press = None
self.removeRequested.emit(self)
super().mouseDoubleClickEvent(event)
def mouseReleaseEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self._press = None
super().mouseReleaseEvent(event)
class _TagFrame(QFrame):
clicked = pyqtSignal()
dragMoved = pyqtSignal(QPoint)
dragLeft = pyqtSignal()
dropped = pyqtSignal(QPoint, int)
def __init__(self, parent=None):
super().__init__(parent)
self.setAcceptDrops(True)
def mousePressEvent(self, event):
self.clicked.emit()
super().mousePressEvent(event)
def dragEnterEvent(self, event):
if event.mimeData().hasFormat(TAG_MIME):
event.acceptProposedAction()
else:
event.ignore()
def dragMoveEvent(self, event):
if event.mimeData().hasFormat(TAG_MIME):
event.acceptProposedAction()
self.dragMoved.emit(event.position().toPoint())
def dragLeaveEvent(self, event):
self.dragLeft.emit()
def dropEvent(self, event):
if event.mimeData().hasFormat(TAG_MIME):
try:
src = int(bytes(event.mimeData().data(TAG_MIME)).decode())
except ValueError:
event.ignore()
return
event.acceptProposedAction()
self.dropped.emit(event.position().toPoint(), src)
class TagField(QWidget):
changed = pyqtSignal()
highlightChanged = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self._channels: list[int] = []
self._chips: list[TagChip] = []
self._highlighted: set[int] = set()
self.frame = _TagFrame()
self.frame.setObjectName("tagField")
self.frame.setMinimumHeight(54)
self._flow = FlowLayout(self.frame, 6, 6)
self._flow.setContentsMargins(8, 8, 8, 8)
self.input = QLineEdit()
self.input.setObjectName("tagInput")
self.input.setMinimumWidth(150)
self.input.setAcceptDrops(False)
self._flow.addWidget(self.input)
self._indicator = QFrame(self.frame)
self._indicator.setObjectName("dropLine")
self._indicator.hide()
self.frame.clicked.connect(self.input.setFocus)
self.frame.dragMoved.connect(self._on_drag_move)
self.frame.dragLeft.connect(self._indicator.hide)
self.frame.dropped.connect(self._on_drop)
self.input.returnPressed.connect(self._commit)
self.input.textEdited.connect(self._on_text)
self.input.installEventFilter(self)
self.hint = QLabel()
self.hint.setObjectName("hint")
self.hint.setWordWrap(True)
lay = QVBoxLayout(self)
lay.setContentsMargins(0, 0, 0, 0)
lay.setSpacing(6)
lay.addWidget(self.frame)
lay.addWidget(self.hint)
self._update_placeholder()
def set_channels(self, channels: list[int], highlighted: set[int] | None = None) -> None:
for chip in self._chips:
chip.setParent(None)
chip.deleteLater()
self._chips.clear()
self._channels = []
self._highlighted = highlighted or set()
for ch in channels:
self._channels.append(ch)
self._add_chip(ch)
self.input.clear()
self.hint.clear()
self._update_placeholder()
def set_highlighted(self, highlighted: set[int]) -> None:
self._highlighted = highlighted
for chip in self._chips:
try:
ch = int(chip.text())
chip.set_highlighted(ch in highlighted)
except ValueError:
pass
def channels(self) -> list[int]:
return list(self._channels)
def highlighted(self) -> set[int]:
return set(self._highlighted)
def commit_pending(self) -> None:
self._commit()
def eventFilter(self, obj, event):
if obj is self.input:
t = event.type()
if t == QEvent.Type.KeyPress and event.key() == Qt.Key.Key_Backspace \
and not self.input.text() and self._channels:
self._remove(len(self._channels) - 1)
return True
if t == QEvent.Type.FocusIn:
self._set_focused(True)
elif t == QEvent.Type.FocusOut:
self._set_focused(False)
self._commit()
return super().eventFilter(obj, event)
def _set_focused(self, on: bool) -> None:
self.frame.setProperty("focused", on)
self.frame.style().unpolish(self.frame)
self.frame.style().polish(self.frame)
def _update_placeholder(self) -> None:
self.input.setPlaceholderText("" if self._channels else "Введите каналы (13 19 14 или 1-4) и нажмите Enter")
def _on_text(self, text: str) -> None:
if text and text[-1] in " ,;":
self._commit()
def _commit(self) -> None:
text = self.input.text()
if not text.strip():
self.input.clear()
return
ok, bad = parse_channels(text)
added, dupes = [], []
for n in ok:
if n in self._channels:
dupes.append(n)
else:
self._channels.append(n)
self._add_chip(n)
added.append(n)
notes = []
if bad:
notes.append(f"Не подходит: {', '.join(bad)} (допустимо {CH_MIN}–{CH_MAX})")
if dupes:
notes.append(f"Уже есть: {', '.join(map(str, dupes))}")
self.hint.setText(". ".join(notes))
self.input.setText(" ".join(bad))
if added:
self._update_placeholder()
self.changed.emit()
def _add_chip(self, ch: int) -> None:
chip = TagChip(ch, ch in self._highlighted)
chip.removeRequested.connect(self._on_chip_remove)
chip.dragRequested.connect(self._start_drag)
chip.highlightToggled.connect(self._on_highlight_toggle)
self._flow.insertWidget(len(self._chips), chip)
self._chips.append(chip)
def _on_highlight_toggle(self, chip) -> None:
try:
ch = int(chip.text())
except ValueError:
return
if ch in self._highlighted:
self._highlighted.remove(ch)
chip.set_highlighted(False)
else:
self._highlighted.add(ch)
chip.set_highlighted(True)
self.highlightChanged.emit()
def _on_chip_remove(self, chip) -> None:
if chip in self._chips:
self._remove(self._chips.index(chip))
def _remove(self, index: int) -> None:
if 0 <= index < len(self._channels):
del self._channels[index]
chip = self._chips.pop(index)
try:
ch = int(chip.text())
if ch in self._highlighted:
self._highlighted.discard(ch)
except ValueError:
pass
chip.setParent(None)
chip.deleteLater()
self._update_placeholder()
self.changed.emit()
self.input.setFocus()
def _start_drag(self, chip) -> None:
if chip not in self._chips:
return
mime = QMimeData()
mime.setData(TAG_MIME, str(self._chips.index(chip)).encode())
drag = QDrag(self)
drag.setMimeData(mime)
pixmap = chip.grab()
drag.setPixmap(pixmap)
drag.setHotSpot(QPoint(pixmap.width() // 2, pixmap.height() // 2))
drag.exec(Qt.DropAction.MoveAction)
self._indicator.hide()
def _target_index(self, pos: QPoint) -> int:
for i, chip in enumerate(self._chips):
g = chip.geometry()
if pos.y() < g.top():
return i
if pos.y() <= g.bottom() and pos.x() < g.center().x():
return i
return len(self._chips)
def _on_drag_move(self, pos: QPoint) -> None:
if not self._chips:
return
t = self._target_index(pos)
if t < len(self._chips):
g = self._chips[t].geometry()
x = g.left() - 4
else:
g = self._chips[-1].geometry()
x = g.right() + 3
self._indicator.setGeometry(x, g.top(), 2, g.height())
self._indicator.show()
self._indicator.raise_()
def _on_drop(self, pos: QPoint, src: int) -> None:
self._indicator.hide()
t = self._target_index(pos)
if not 0 <= src < len(self._chips):
return
if t > src:
t -= 1
if t == src:
return
self._channels.insert(t, self._channels.pop(src))
self._chips.insert(t, self._chips.pop(src))
self._flow.reorder(self._chips)
self.changed.emit()
# ============================================================================
# Диалог пароля
# ============================================================================
class AuthDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("Вход")
self.setFixedSize(300, 140)
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
layout = QVBoxLayout(self)
layout.setContentsMargins(20, 20, 20, 20)
layout.setSpacing(15)
lbl = QLabel("Введите пароль для доступа:")
lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(lbl)
self.pwd_input = QLineEdit()
self.pwd_input.setEchoMode(QLineEdit.EchoMode.Password)
self.pwd_input.setPlaceholderText("Пароль")
self.pwd_input.returnPressed.connect(self.try_auth)
layout.addWidget(self.pwd_input)
btn = QPushButton("Войти")
set_variant(btn, "white")
btn.clicked.connect(self.try_auth)
layout.addWidget(btn)
self.lbl_error = QLabel()
self.lbl_error.setStyleSheet("color: #ff6b6b;")
self.lbl_error.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(self.lbl_error)
def try_auth(self):
if self.pwd_input.text() == PASSWORD:
self.accept()
else:
self.lbl_error.setText("Неверный пароль")
self.pwd_input.clear()
# ============================================================================
# Диалог сканера
# ============================================================================
class _SortItem(QTableWidgetItem):
def __init__(self, text: str, key):
super().__init__(text)
self._key = key
def __lt__(self, other) -> bool:
try:
return self._key < other._key
except (AttributeError, TypeError):
return super().__lt__(other)
class ScannerDialog(QDialog):
COLS = ("Статус", "IP-адрес", "Имя", "Производитель", "MAC-адрес", "Отклик", "Порт")
COL_IP, COL_MAC = 1, 4
def __init__(self, parent=None, base_ip: str = "", port: int = 40250, palette: dict | None = None):
super().__init__(parent)
self.setWindowTitle("Сканер сети")
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
self.resize(940, 580)
self._colors = palette or THEMES[DEFAULT_THEME][1]
self.selected_ip = ""
self._hosts: dict[str, HostResult] = {}
self._total = 0
self._dirty = False
self._started_at = 0.0
default_range = ""
try:
default_range = f"{str(ipaddress.IPv4Address(base_ip.strip())).rsplit('.', 1)[0]}.1-254"
except ValueError:
default_range = detect_local_range()
self.range_input = QLineEdit(default_range)
self.range_input.setPlaceholderText("192.168.100.1-254 или 192.168.100.0/24")
self.range_input.setToolTip("Примеры: 192.168.1.1-254 · 10.0.0.5 - 10.0.0.20 · 192.168.1.0/24 · несколько через запятую")
self.btn_mynet = QPushButton("Моя сеть")
self.btn_mynet.setToolTip("Подставить подсеть этого компьютера")
self.port_input = IntField(1, 65535, port)
self.port_input.setFixedWidth(80)
self.port_input.setToolTip("Порт устройства: узлы с открытым портом подсвечиваются в таблице")
self.btn_scan = QPushButton("Сканировать")
set_variant(self.btn_scan, "white")
self.btn_scan.setMinimumWidth(130)
top = QHBoxLayout()
top.setSpacing(8)
lbl_range = QLabel("Диапазон")
lbl_range.setObjectName("dim")
lbl_port = QLabel("Порт")
lbl_port.setObjectName("dim")
top.addWidget(lbl_range)
top.addWidget(self.range_input, 1)
top.addWidget(self.btn_mynet)
top.addWidget(lbl_port)
top.addWidget(self.port_input)
top.addWidget(self.btn_scan)
self.progress = QProgressBar()
self.progress.setTextVisible(False)
self.progress.setRange(0, 1)
self.progress.setValue(0)
self.lbl_error = QLabel()
self.lbl_error.setObjectName("hint")
self.table = QTableWidget(0, len(self.COLS))
self.table.setHorizontalHeaderLabels(list(self.COLS))
self.table.verticalHeader().setVisible(False)
self.table.verticalHeader().setDefaultSectionSize(26)
self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.table.setShowGrid(False)
self.table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
header = self.table.horizontalHeader()
header.setHighlightSections(False)
header.setSortIndicatorShown(True)
for col, mode in enumerate((
QHeaderView.ResizeMode.ResizeToContents, QHeaderView.ResizeMode.ResizeToContents,
QHeaderView.ResizeMode.Stretch, QHeaderView.ResizeMode.Stretch,
QHeaderView.ResizeMode.ResizeToContents, QHeaderView.ResizeMode.ResizeToContents,
QHeaderView.ResizeMode.ResizeToContents,
)):
header.setSectionResizeMode(col, mode)
header.setSortIndicator(self.COL_IP, Qt.SortOrder.AscendingOrder)
self.table.setSortingEnabled(True)
self.chk_dead = QCheckBox("Показывать недоступные")
self.lbl_stats = QLabel("Укажите диапазон и нажмите «Сканировать».")
self.lbl_stats.setObjectName("dim")
self.btn_close = QPushButton("Закрыть")
bottom = QHBoxLayout()
bottom.setSpacing(8)
bottom.addWidget(self.chk_dead)
bottom.addStretch()
bottom.addWidget(self.btn_close)
layout = QVBoxLayout(self)
layout.setContentsMargins(16, 16, 16, 16)
layout.setSpacing(10)
layout.addLayout(top)
layout.addWidget(self.progress)
layout.addWidget(self.lbl_error)
layout.addWidget(self.table, 1)
layout.addWidget(self.lbl_stats)
layout.addLayout(bottom)
self.scanner = NetworkScanner(self)
self.timer = QTimer(self)
self.timer.setInterval(250)
self.timer.timeout.connect(self._tick)
self.scanner.host_result.connect(self._on_host)
self.scanner.progress.connect(self._on_progress)
self.scanner.finished.connect(self._on_finished)
self.btn_scan.clicked.connect(self._toggle_scan)
self.range_input.returnPressed.connect(self._toggle_scan)
self.btn_mynet.clicked.connect(self._use_my_network)
self.btn_close.clicked.connect(self.reject)
self.chk_dead.toggled.connect(lambda _: self._render())
self.table.itemDoubleClicked.connect(lambda _item: self._pick())
self.table.customContextMenuRequested.connect(self._context_menu)
def keyPressEvent(self, event):
if event.key() in (Qt.Key.Key_Return, Qt.Key.Key_Enter):
event.accept()
return
super().keyPressEvent(event)
def done(self, result: int) -> None:
self.timer.stop()
self.scanner.stop()
self.scanner.wait()
super().done(result)
def _use_my_network(self) -> None:
text = detect_local_range()
if text:
self.range_input.setText(text)
self.lbl_error.clear()
else:
self.lbl_error.setText("Не удалось определить локальную сеть.")
def _toggle_scan(self) -> None:
if self.scanner.running:
self.scanner.stop()
self.btn_scan.setEnabled(False)
self.btn_scan.setText("Остановка…")
return
ips, error = parse_targets(self.range_input.text())
self.lbl_error.setText(error)
if not ips:
return
port = self.port_input.value()
self._hosts.clear()
self._total = len(ips)
self._started_at = time.monotonic()
self.progress.setRange(0, self._total)
self.progress.setValue(0)
labels = list(self.COLS)
labels[6] = f"Порт {port}"
self.table.setHorizontalHeaderLabels(labels)
self.btn_scan.setText("Стоп")
set_variant(self.btn_scan, "danger")
for w in (self.range_input, self.btn_mynet, self.port_input):
w.setEnabled(False)
self._render()
self.timer.start()
self.scanner.start(ips, port)
def _on_host(self, host: HostResult) -> None:
self._hosts[host.ip] = host
self._dirty = True
def _on_progress(self, done: int, total: int) -> None:
self.progress.setValue(done)
def _on_finished(self) -> None:
self.timer.stop()
self._render()
self.btn_scan.setEnabled(True)
self.btn_scan.setText("Сканировать")
set_variant(self.btn_scan, "white")
for w in (self.range_input, self.btn_mynet, self.port_input):
w.setEnabled(True)
elapsed = time.monotonic() - self._started_at
alive = sum(1 for h in self._hosts.values() if h.alive)
self.lbl_stats.setText(f"Готово за {elapsed:.1f} с · в сети: {alive} · проверено адресов: {len(self._hosts)} из {self._total}")
def _tick(self) -> None:
if self._dirty:
self._render()
def _ip_at(self, row: int) -> str:
item = self.table.item(row, self.COL_IP)
return item.data(Qt.ItemDataRole.UserRole) if item is not None else ""
def _render(self) -> None:
self._dirty = False
show_dead = self.chk_dead.isChecked()
hosts = [h for h in self._hosts.values() if h.alive or show_dead]
selected = self._ip_at(self.table.currentRow()) if self.table.currentRow() >= 0 else ""
header = self.table.horizontalHeader()
col, order = header.sortIndicatorSection(), header.sortIndicatorOrder()
self.table.setSortingEnabled(False)
self.table.setRowCount(len(hosts))
for row, h in enumerate(hosts):
self._fill_row(row, h)
self.table.setSortingEnabled(True)
self.table.sortItems(col, order)
if selected:
for row in range(self.table.rowCount()):
if self._ip_at(row) == selected:
self.table.selectRow(row)
break
if self.scanner.running:
alive = sum(1 for h in self._hosts.values() if h.alive)
self.lbl_stats.setText(f"Сканирование… в сети: {alive} · проверено: {len(self._hosts)} из {self._total}")
def _fill_row(self, row: int, h: HostResult) -> None:
c = self._colors
inf = float("inf")
vendor = oui_lookup(h.mac) if h.mac else ""
if h.rtt is None:
rtt_text, rtt_key = "—", inf
else:
rtt_text, rtt_key = ("<1 мс" if h.rtt < 1 else f"{h.rtt:g} мс"), h.rtt
status = "● В сети" if h.alive else "○ Нет ответа"
if h.note:
status += f" ({h.note})"
cells = [
(status, 0 if h.alive else 1, c["OK"] if h.alive else c["MUTED"]),
(h.ip, int(ipaddress.IPv4Address(h.ip)), c["ACCENT"] if h.port_open else None),
(h.name or "—", h.name.lower() if h.name else "\uffff", None),
(vendor or "—", vendor.lower() if vendor else "\uffff", None),
(h.mac or "—", h.mac or "\uffff", None),
(rtt_text, rtt_key, None),
("✓ открыт" if h.port_open else "—", 0 if h.port_open else 1, c["ACCENT"] if h.port_open else None),
]
for col, (text, key, color) in enumerate(cells):
item = _SortItem(text, key)
if color:
item.setForeground(QBrush(QColor(color)))
elif not h.alive:
item.setForeground(QBrush(QColor(c["MUTED"])))
if col == self.COL_IP:
item.setData(Qt.ItemDataRole.UserRole, h.ip)
self.table.setItem(row, col, item)
def _pick(self) -> None:
row = self.table.currentRow()
ip = self._ip_at(row) if row >= 0 else ""
if ip:
self.selected_ip = ip
self.accept()
def _context_menu(self, pos) -> None:
row = self.table.rowAt(pos.y())
if row < 0:
return
self.table.selectRow(row)
ip = self._ip_at(row)
mac_text = self.table.item(row, self.COL_MAC).text()
mac = "" if mac_text == "—" else mac_text
clip = QApplication.clipboard()
menu = QMenu(self)
actions = [
(menu.addAction("Использовать этот IP"), self._pick),
(menu.addAction("Копировать IP"), lambda: clip.setText(ip)),
]
if mac:
actions.append((menu.addAction("Копировать MAC"), lambda: clip.setText(mac)))
menu.addSeparator()
actions.append((menu.addAction("Открыть в браузере"), lambda: webbrowser.open(f"http://{ip}")))
if platform.system() == "Windows":
actions.append((menu.addAction("Пинг в консоли"), lambda: subprocess.Popen(
["cmd", "/c", "start", f"Ping {ip}", "cmd", "/k", "ping", "-t", ip])))
if mac:
actions.append((menu.addAction("Разбудить (Wake-on-LAN)"), lambda: self._wake(mac)))
chosen = menu.exec(self.table.viewport().mapToGlobal(pos))
for action, fn in actions:
if action is chosen:
fn()
break
def _wake(self, mac: str) -> None:
err = send_wol(mac)
self.lbl_error.setText(f"Wake-on-LAN не отправлен: {err}" if err else "")
if not err:
self.lbl_stats.setText(f"Пакет пробуждения отправлен на {mac}.")
# ============================================================================
# Редактор пресетов
# ============================================================================
class PresetEditorDialog(QDialog):
def __init__(self, presets: dict, current: str, parent=None, highlight_channels: dict | None = None):
super().__init__(parent)
self.setWindowTitle("Редактор пресетов")
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
self.resize(720, 500)
self._entries: list[list] = [[n, list(c)] for n, c in presets.items()]
self._highlight: dict = dict(highlight_channels or {})
title_l = QLabel("ПРЕСЕТЫ")
title_l.setObjectName("cardTitle")
self.list = QListWidget()
self.btn_new = QPushButton("Новый")
self.btn_dup = QPushButton("Копия")
self.btn_del = QPushButton("Удалить")
set_variant(self.btn_del, "danger")
self.btn_new.clicked.connect(self._new)
self.btn_dup.clicked.connect(self._duplicate)
self.btn_del.clicked.connect(self._delete)
btn_row = QHBoxLayout()
btn_row.setSpacing(6)
for b in (self.btn_new, self.btn_dup, self.btn_del):
btn_row.addWidget(b)
left = QVBoxLayout()
left.setSpacing(8)
left.addWidget(title_l)
left.addWidget(self.list, 1)
left.addLayout(btn_row)
left_w = QWidget()
left_w.setFixedWidth(250)
left_w.setLayout(left)
left.setContentsMargins(0, 0, 0, 0)
title_n = QLabel("НАЗВАНИЕ")
title_n.setObjectName("cardTitle")
self.name_edit = QLineEdit()
self.name_edit.setPlaceholderText("Например, F25")
title_c = QLabel("КАНАЛЫ · двойной клик — убрать, перетаскивание — порядок, ПКМ — подсветить")
title_c.setObjectName("cardTitle")
self.editor = TagField()
right = QVBoxLayout()
right.setContentsMargins(0, 0, 0, 0)
right.setSpacing(8)
right.addWidget(title_n)
right.addWidget(self.name_edit)
right.addSpacing(6)
right.addWidget(title_c)
right.addWidget(self.editor)
right.addStretch()
self.right_panel = QWidget()
self.right_panel.setLayout(right)
self.list.currentRowChanged.connect(self._on_row_changed)
self.name_edit.textEdited.connect(self._on_name_edited)
self.editor.changed.connect(self._on_channels_changed)
self.editor.highlightChanged.connect(self._on_highlight_changed)
body = QHBoxLayout()
body.setSpacing(16)
body.addWidget(left_w)
body.addWidget(self.right_panel, 1)
box = QDialogButtonBox(QDialogButtonBox.StandardButton.Save | QDialogButtonBox.StandardButton.Cancel)
save_btn = box.button(QDialogButtonBox.StandardButton.Save)
save_btn.setText("Сохранить")
box.button(QDialogButtonBox.StandardButton.Cancel).setText("Отмена")
box.accepted.connect(self.accept)
box.rejected.connect(self.reject)
root = QVBoxLayout(self)
root.setContentsMargins(16, 16, 16, 16)
root.setSpacing(14)
root.addLayout(body, 1)
root.addWidget(box)
start = next((i for i, e in enumerate(self._entries) if e[0] == current), 0 if self._entries else -1)
self._refresh_list(start)
def presets(self) -> dict:
return {n.strip(): list(c) for n, c in self._entries}
def highlight_channels(self) -> dict:
return self._highlight
def selected_name(self) -> str:
row = self.list.currentRow()
return self._entries[row][0].strip() if 0 <= row < len(self._entries) else ""
def keyPressEvent(self, event):
if event.key() in (Qt.Key.Key_Return, Qt.Key.Key_Enter):
event.accept()
return
super().keyPressEvent(event)
def accept(self) -> None:
self.editor.commit_pending()
problem = self._validate()
if problem:
row, message = problem
self.list.setCurrentRow(row)
QMessageBox.warning(self, "Проверьте пресет", message)
return
super().accept()
@staticmethod
def _item_text(entry: list) -> str:
name, chans = entry
shown = ", ".join(map(str, chans)) if chans else "пусто"
return f"{name.strip() or '(без названия)'} · {shown}"
def _refresh_list(self, select_row: int) -> None:
self.list.blockSignals(True)
self.list.clear()
for e in self._entries:
self.list.addItem(self._item_text(e))
if 0 <= select_row < len(self._entries):
self.list.setCurrentRow(select_row)
self.list.blockSignals(False)
self._on_row_changed(self.list.currentRow())
def _update_item(self, row: int) -> None:
item = self.list.item(row)
if item is not None:
item.setText(self._item_text(self._entries[row]))
def _on_row_changed(self, row: int) -> None:
has = 0 <= row < len(self._entries)
self.right_panel.setEnabled(has)
self.btn_dup.setEnabled(has)
self.btn_del.setEnabled(has)
if has:
name = self._entries[row][0].strip()
self.name_edit.setText(self._entries[row][0])
self.editor.set_channels(self._entries[row][1], set(self._highlight.get(name, [])))
else:
self.name_edit.clear()
self.editor.set_channels([])
def _on_name_edited(self, text: str) -> None:
row = self.list.currentRow()
if 0 <= row < len(self._entries):
old_name = self._entries[row][0].strip()
self._entries[row][0] = text
self._update_item(row)
new_name = text.strip()
if old_name in self._highlight and old_name != new_name:
self._highlight[new_name] = self._highlight.pop(old_name)
def _on_channels_changed(self) -> None:
row = self.list.currentRow()
if 0 <= row < len(self._entries):
self._entries[row][1] = self.editor.channels()
self._update_item(row)
def _on_highlight_changed(self) -> None:
row = self.list.currentRow()
if 0 <= row < len(self._entries):
name = self._entries[row][0].strip()
hl = self.editor.highlighted()
if hl:
self._highlight[name] = list(hl)
else:
self._highlight.pop(name, None)
def _unique_name(self, base: str) -> str:
names = {e[0].strip() for e in self._entries}
if base not in names:
return base
n = 2
while f"{base} {n}" in names:
n += 1
return f"{base} {n}"
def _new(self) -> None:
self._entries.append([self._unique_name("Новый пресет"), []])
self._refresh_list(len(self._entries) - 1)
self.name_edit.setFocus()
self.name_edit.selectAll()
def _duplicate(self) -> None:
row = self.list.currentRow()
if 0 <= row < len(self._entries):
name, chans = self._entries[row]
new_name = self._unique_name(f"{name.strip()} копия")
self._entries.append([new_name, list(chans)])
old_hl = self._highlight.get(name.strip())
if old_hl:
self._highlight[new_name] = list(old_hl)
self._refresh_list(len(self._entries) - 1)
def _delete(self) -> None:
row = self.list.currentRow()
if 0 <= row < len(self._entries):
name = self._entries[row][0].strip()
self._highlight.pop(name, None)
del self._entries[row]
self._refresh_list(min(row, len(self._entries) - 1))
def _validate(self) -> tuple[int, str] | None:
seen = set()
for i, (name, chans) in enumerate(self._entries):
n = name.strip()
if not n:
return i, "У пресета нет названия."
if n in seen:
return i, f"Название «{n}» встречается дважды."
if not chans:
return i, f"В пресете «{n}» нет ни одного канала."
seen.add(n)
return None
# ============================================================================
# Иконка замка (маленькая, залитая)
# ============================================================================
class LockIcon(QWidget):
"""Крошечный залитый замочек для нижней панели."""
clicked = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self.setFixedSize(10, 10) # Совсем крошечный
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.setToolTip("Блокировка (3 клика)")
self._clicks = 0
self._last_click = 0.0
def paintEvent(self, event):
p = QPainter(self)
p.setRenderHint(QPainter.RenderHint.Antialiasing)
p.setPen(Qt.PenStyle.NoPen)
p.setBrush(QColor("#8b8b95")) # Залитый цвет (серый)
# Рисуем простой прямоугольник (тело) и полукруг (дужка), все слито воедино
p.drawRect(2, 5, 6, 4) # Тело
p.drawEllipse(3, 2, 4, 4) # Дужка (верхняя часть)
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
now = time.monotonic()
if now - self._last_click > 1.0:
self._clicks = 0
self._last_click = now
self._clicks += 1
if self._clicks >= 3:
self._clicks = 0
self.clicked.emit()
# ============================================================================
# Главное окно
# ============================================================================
def make_card(title: str, right: QWidget | None = None) -> tuple[QFrame, QVBoxLayout]:
frame = QFrame()
frame.setObjectName("card")
frame.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Maximum)
lay = QVBoxLayout(frame)
lay.setContentsMargins(14, 12, 14, 14)
lay.setSpacing(10)
head = QHBoxLayout()
lbl = QLabel(title)
lbl.setObjectName("cardTitle")
head.addWidget(lbl)
head.addStretch()
if right is not None:
head.addWidget(right)
lay.addLayout(head)
return frame, lay
class SwitchController(QWidget):
def __init__(self):
super().__init__()
S = QAbstractSocket.SocketState
self._S = S
self.cfg, cfg_warning = Config.load()
self.presets: dict = {k: list(v) for k, v in self.cfg.presets.items()}
self.highlight_channels: dict = dict(self.cfg.highlight_channels)
self.current_preset = self.cfg.preset if self.cfg.preset in self.presets else next(iter(self.presets), "")
self.device_channel: int | None = None
self.index = -1
self._uptime_pending = False
self._prev_state = S.UnconnectedState
self._save_warned = False
self._log_visible = True
self._channels_visible = True
self._last_step_time = 0.0
self._wide_width = 1000
self.client = DeviceClient(self)
self.hotkeys = GlobalHotkeys(self)
self._build_ui()
self._wire()
self._fill_preset_combo()
self._on_state(S.UnconnectedState)
if cfg_warning:
self.log(cfg_warning, "warn")
if self.cfg.global_hotkeys and GlobalHotkeys.available():
self.chk_global.setChecked(True)
self._set_log_visible(False)
self._refresh_theme_label()
def _build_ui(self) -> None:
self.setObjectName("root")
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
self.setWindowTitle("Switch Controller" + (" — тестовый режим" if TEST_MODE else ""))
self.resize(1000, 750)
self.lbl_status = QLabel()
self.lbl_status.setObjectName("status")
card_conn, lay = make_card("ПОДКЛЮЧЕНИЕ", self.lbl_status)
self.ip_input = QLineEdit(self.cfg.ip)
self.ip_input.setPlaceholderText("IP-адрес")
lay.addWidget(self.ip_input)
row_port = QHBoxLayout()
row_port.setSpacing(8)
self.port_input = IntField(1, 65535, self.cfg.port)
self.port_input.setFixedWidth(100)
self.port_input.setToolTip("Порт")
self.btn_connect = QPushButton()
row_port.addWidget(self.port_input)
row_port.addWidget(self.btn_connect, 1)
lay.addLayout(row_port)
card_ch, lay = make_card("КАНАЛ")
nav = QHBoxLayout()
nav.setSpacing(4)
self.btn_prev = QPushButton("◀")
self.btn_prev.setObjectName("nav")
self.btn_prev.setFixedHeight(36)
self.btn_prev.setToolTip("Предыдущий канал пресета (←)")
self.btn_next = QPushButton("▶")
self.btn_next.setObjectName("nav")
self.btn_next.setFixedHeight(36)
self.btn_next.setToolTip("Следующий канал пресета (→)")
self.spin = IntField(CH_MIN, CH_MAX, 13)
self.spin.setObjectName("big")
self.spin.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.spin.setFixedHeight(36)
nav.addWidget(self.btn_prev)
nav.addWidget(self.spin, 1)
nav.addWidget(self.btn_next)
lay.addLayout(nav)
self.btn_set_channel = QPushButton("Установить канал")
set_variant(self.btn_set_channel, "white")
lay.addWidget(self.btn_set_channel)
lbl_preset = QLabel("Пресет")
lbl_preset.setObjectName("dim")
lay.addWidget(lbl_preset)
prow = QHBoxLayout()
prow.setSpacing(8)
self.preset_combo = QComboBox()
self.btn_edit = QPushButton("Изменить")
prow.addWidget(self.preset_combo, 1)
prow.addWidget(self.btn_edit)
lay.addLayout(prow)
self.strip = ChipStrip()
self.strip_scroll = ChipScroll(self.strip)
lay.addWidget(self.strip_scroll)
self.lbl_empty = QLabel("Пресетов нет — создайте их в редакторе")
self.lbl_empty.setObjectName("dim")
lay.addWidget(self.lbl_empty)
self.chk_global = QCheckBox("Глобальные клавиши ← →")
self.chk_global.setToolTip(
"Переключать каналы стрелками, даже когда окно программы не в фокусе.\n"
"Пока окно активно, стрелки работают и без этой опции (кроме текстовых полей)."
)
if not GlobalHotkeys.available():
self.chk_global.setEnabled(False)
self.chk_global.setToolTip("Для этой возможности установите pynput: pip install pynput")
lay.addWidget(self.chk_global)
card_sch, lay = make_card("СХЕМА")
up_layout = QHBoxLayout()
up_layout.setSpacing(8)
lbl_up = QLabel("Аптайм")
lbl_up.setObjectName("dim")
self.lbl_uptime = QLabel("—")
self.lbl_uptime.setObjectName("value")
self.btn_uptime = QPushButton("⟳")
self.btn_uptime.setObjectName("nav")
self.btn_uptime.setToolTip("Запросить аптайм")
up_layout.addWidget(lbl_up)
up_layout.addWidget(self.lbl_uptime, 1)
up_layout.addWidget(self.btn_uptime)
lay.addLayout(up_layout)
app_layout = QHBoxLayout()
app_layout.setSpacing(8)
lbl_applied = QLabel("Применена")
lbl_applied.setObjectName("dim")
self.lbl_scheme_applied = QLabel("—")
self.lbl_scheme_applied.setObjectName("value")
app_layout.addWidget(lbl_applied)
app_layout.addWidget(self.lbl_scheme_applied, 1)
lay.addLayout(app_layout)
srow = QHBoxLayout()
srow.setSpacing(8)
self.scheme_combo = QComboBox()
self.scheme_combo.addItems(list(SCHEMES))
self.scheme_combo.setCurrentText(self.cfg.scheme)
self.btn_set_scheme = QPushButton("Установить")
srow.addWidget(self.scheme_combo, 1)
srow.addWidget(self.btn_set_scheme)
lay.addLayout(srow)
self.left_w = QWidget()
left = QVBoxLayout(self.left_w)
left.setContentsMargins(0, 0, 0, 0)
left.setSpacing(12)
left.addWidget(card_conn)
left.addWidget(card_ch)
left.addWidget(card_sch)
self.left_w.setFixedWidth(370)
left.addStretch()
self.right_w = QWidget()
right = QVBoxLayout(self.right_w)
right.setContentsMargins(0, 0, 0, 0)
right.setSpacing(0)
self.log_view = QTextEdit()
self.log_view.setObjectName("log")
self.log_view.setReadOnly(True)
self.log_view.document().setMaximumBlockCount(2000)
right.addWidget(self.log_view, 1)
self.bottom_bar = QWidget()
self.bottom_bar.setFixedHeight(30)
bottom_layout = QHBoxLayout(self.bottom_bar)
bottom_layout.setContentsMargins(10, 0, 10, 0)
self.btn_scan = QPushButton("Сканер сети")
self.btn_scan.setObjectName("ghost")
self.btn_scan.clicked.connect(self.open_scanner)
self.btn_toggle_log = QPushButton("Лог")
self.btn_toggle_log.setObjectName("ghost")
self.btn_toggle_log.clicked.connect(self._toggle_log)
self.btn_toggle_channels = QPushButton("Скрыть каналы")
self.btn_toggle_channels.setObjectName("ghost")
self.btn_toggle_channels.clicked.connect(self._toggle_channels)
self.btn_theme = QPushButton("Тема")
self.btn_theme.setObjectName("ghost")
self.btn_theme.clicked.connect(self._toggle_theme)
self.lock_icon = LockIcon()
self.lock_icon.clicked.connect(self.lock_program)
bottom_layout.addWidget(self.btn_scan)
bottom_layout.addWidget(self.btn_toggle_log)
bottom_layout.addWidget(self.btn_toggle_channels)
bottom_layout.addWidget(self.btn_theme)
bottom_layout.addStretch()
bottom_layout.addWidget(self.lock_icon)
self.content_layout = QHBoxLayout()
self.content_layout.setSpacing(14)
self.content_layout.addWidget(self.left_w)
self.content_layout.addWidget(self.right_w, 1)
self.root_layout = QVBoxLayout(self)
self.root_layout.setContentsMargins(14, 14, 14, 14)
self.root_layout.setSpacing(10)
self.root_layout.addLayout(self.content_layout, 1)
self.root_layout.addWidget(self.bottom_bar)
def lock_program(self):
self.cfg.authorized = False
self.cfg.save()
QApplication.instance().quit()
def _toggle_log(self) -> None:
self._set_log_visible(not self._log_visible)
def _set_log_visible(self, visible: bool) -> None:
self._log_visible = visible
self.right_w.setVisible(visible)
self.btn_toggle_log.setText("<< Лог" if visible else "Лог >>")
if visible:
self.setMaximumWidth(QWIDGETSIZE_MAX)
self.resize(max(self._wide_width, self.minimumSizeHint().width()), self.height())
else:
self._wide_width = self.width()
self.layout().activate()
w = self.minimumSizeHint().width()
self.resize(w, self.height())
self.setMaximumWidth(w)
def _toggle_channels(self):
if self._channels_visible:
self.strip_scroll.setVisible(False)
self.lbl_empty.setVisible(False)
self.btn_toggle_channels.setText("Показать каналы")
self._channels_visible = False
else:
self.strip_scroll.setVisible(True)
self.lbl_empty.setVisible(not self.presets)
self.btn_toggle_channels.setText("Скрыть каналы")
self._channels_visible = True
def _pal(self) -> dict:
return THEMES.get(self.cfg.theme, THEMES[DEFAULT_THEME])[1]
def _refresh_theme_label(self) -> None:
name = THEMES.get(self.cfg.theme, THEMES[DEFAULT_THEME])[0]
self.btn_theme.setText(f"Тема: {name}")
def _toggle_theme(self) -> None:
keys = list(THEMES)
i = keys.index(self.cfg.theme) if self.cfg.theme in keys else 0
self.cfg.theme = keys[(i + 1) % len(keys)]
self._save()
self.apply_theme()
def apply_theme(self) -> None:
palette = self._pal()
QApplication.instance().setStyleSheet(build_qss(palette, make_icons(palette, self.cfg.theme)))
self._refresh_theme_label()
self._on_state(self.client.state)
def _wire(self) -> None:
self.btn_connect.clicked.connect(self.toggle_connection)
self.btn_prev.clicked.connect(lambda: self.step(-1))
self.btn_next.clicked.connect(lambda: self.step(+1))
self.btn_set_channel.clicked.connect(self.send_spin)
self.spin.returnPressed.connect(self.send_spin)
self.preset_combo.currentIndexChanged.connect(self._on_preset_selected)
self.btn_edit.clicked.connect(self.open_preset_editor)
self.strip.chipClicked.connect(self._go)
self.btn_set_scheme.clicked.connect(self.send_scheme)
self.btn_uptime.clicked.connect(self.request_uptime)
self.chk_global.toggled.connect(self._toggle_global)
self.client.state_changed.connect(self._on_state)
self.client.error.connect(lambda msg: self.log(msg, "err"))
self.client.data.connect(self._on_data)
self.hotkeys.pressed.connect(self._on_global_key)
for key, delta in (("Left", -1), ("Right", +1)):
sc = QShortcut(QKeySequence(key), self)
sc.setContext(Qt.ShortcutContext.WindowShortcut)
sc.activated.connect(lambda d=delta: self.step(d))
def open_scanner(self):
dlg = ScannerDialog(self, self.ip_input.text().strip(), self.port_input.value(), self._pal())
if dlg.exec() == QDialog.DialogCode.Accepted and dlg.selected_ip:
self.ip_input.setText(dlg.selected_ip)
self.log(f"Выбран IP {dlg.selected_ip}")
self._save()
def log(self, text: str, kind: str = "info") -> None:
palette = self._pal()
colors = {"info": palette["TEXT"], "tx": palette["ACCENT"], "rx": palette["OK"],
"warn": palette["WARN"], "err": palette["BAD"]}
ts = datetime.now().strftime("%H:%M:%S")
self.log_view.append(
f'<span style="color:{palette["MUTED"]}">[{ts}]</span> '
f'<span style="color:{colors.get(kind, colors["info"])}">{html.escape(text)}</span>'
)
bar = self.log_view.verticalScrollBar()
bar.setValue(bar.maximum())
def _save(self) -> None:
self.cfg.ip = self.ip_input.text().strip() or self.cfg.ip
self.cfg.port = self.port_input.value()
self.cfg.scheme = self.scheme_combo.currentText()
self.cfg.preset = self.current_preset
self.cfg.global_hotkeys = self.chk_global.isChecked()
self.cfg.presets = {k: list(v) for k, v in self.presets.items()}
self.cfg.highlight_channels = dict(self.highlight_channels)
err = self.cfg.save()
if err and not self._save_warned:
self._save_warned = True
self.log(f"Не удалось сохранить настройки: {err}", "warn")
def toggle_connection(self) -> None:
S = self._S
if self.client.state != S.UnconnectedState:
was_connecting = self.client.state != S.ConnectedState
self.client.close()
if was_connecting:
self.log("Подключение отменено", "warn")
return
host = self.ip_input.text().strip()
if not host:
self.log("Введите IP-адрес", "warn")
return
port = self.port_input.value()
self._save()
self.log(f"Подключение к {host}:{port}…")
self.client.open(host, port)
def _on_state(self, state) -> None:
S = self._S
prev, self._prev_state = self._prev_state, state
connected = state == S.ConnectedState
idle = state == S.UnconnectedState
palette = self._pal()
if connected:
text, color = "● Подключено", palette["OK"]
self.btn_connect.setText("Отключиться")
set_variant(self.btn_connect, "danger")
elif idle:
text, color = "● Не подключено", palette["BAD"]
self.btn_connect.setText("Подключиться")
set_variant(self.btn_connect, "")
else:
text, color = "● Подключение…", palette["WARN"]
self.btn_connect.setText("Отмена")
set_variant(self.btn_connect, "")
self.lbl_status.setText(text)
self.lbl_status.setStyleSheet(f"color: {color};")
self.ip_input.setEnabled(idle)
self.port_input.setEnabled(idle)
self.btn_scan.setEnabled(idle)
for w in (self.btn_prev, self.btn_next, self.btn_set_channel, self.strip,
self.btn_set_scheme, self.btn_uptime):
w.setEnabled(connected or TEST_MODE)
if connected and prev != S.ConnectedState:
self.log(f"Подключено к {self.ip_input.text().strip()}:{self.port_input.value()}", "rx")
QTimer.singleShot(200, self.send_scheme)
QTimer.singleShot(300, self.request_uptime)
elif idle and prev == S.ConnectedState:
self.log("Отключено", "warn")
self.lbl_scheme_applied.setText("—")
self.lbl_uptime.setText("—")
self._uptime_pending = False
self.device_channel = None
self._sync_chips()
def _on_data(self, data: bytes) -> None:
self.log(f"← {hex_str(data)}", "rx")
if self._uptime_pending:
self._uptime_pending = False
self.lbl_uptime.setText(parse_uptime(data) or "ответ в журнале")
def _send(self, packet: bytes, what: str) -> bool:
if not self.client.is_connected:
if TEST_MODE:
self.log(f"→ {what} [{hex_str(packet)}] (тест: не отправлено)", "warn")
return True
self.log("Нет подключения к устройству", "warn")
return False
if not self.client.send(packet):
self.log(f"Не удалось отправить: {what}", "err")
return False
self.log(f"→ {what} [{hex_str(packet)}]", "tx")
return True
def send_channel(self, ch: int) -> bool:
ok = self._send(build_set_packet(CMD_SET_CHANNEL, ch), f"канал {ch}")
if ok:
self.device_channel = ch
self.spin.setValue(ch)
self._sync_chips()
return ok
def send_spin(self) -> None:
self.send_channel(self.spin.value())
def send_scheme(self) -> None:
name = self.scheme_combo.currentText()
code = SCHEMES[name]
if self._send(build_set_packet(CMD_SET_SCHEME, code), f"схема {name} (код {code})"):
self.lbl_scheme_applied.setText(name)
self._save()
def request_uptime(self) -> None:
if self._send(build_uptime_packet(), "запрос аптайма"):
if self.client.is_connected:
self._uptime_pending = True
self.lbl_uptime.setText("запрошено…")
else:
self.lbl_uptime.setText("тест")
def _channels(self) -> list[int]:
return self.presets.get(self.current_preset, [])
def _fill_preset_combo(self) -> None:
self.preset_combo.blockSignals(True)
self.preset_combo.clear()
self.preset_combo.addItems(list(self.presets))
i = self.preset_combo.findText(self.current_preset)
if i >= 0:
self.preset_combo.setCurrentIndex(i)
self.preset_combo.blockSignals(False)
self._refresh_strip()
def _refresh_strip(self) -> None:
chs = self._channels()
hl = set(self.highlight_channels.get(self.current_preset, []))
self.strip.set_channels(chs, hl)
self.lbl_empty.setVisible(not self.presets and self._channels_visible)
self._sync_chips()
def _sync_chips(self) -> None:
chs = self._channels()
self.index = chs.index(self.device_channel) if self.device_channel in chs else -1
self.strip.set_active(self.index)
def _on_preset_selected(self, i: int) -> None:
if i < 0:
return
self.current_preset = self.preset_combo.itemText(i)
self._refresh_strip()
self._save()
self.log(f"Пресет: {self.current_preset}")
def step(self, delta: int) -> None:
now = time.monotonic()
if now - self._last_step_time < DEBOUNCE_TIMEOUT:
return
self._last_step_time = now
chs = self._channels()
if not chs:
return
if self.index < 0:
i = 0 if delta > 0 else len(chs) - 1
else:
i = (self.index + delta) % len(chs)
self._go(i)
def _go(self, i: int) -> None:
chs = self._channels()
if 0 <= i < len(chs):
self.send_channel(chs[i])
else:
self._sync_chips()
def open_preset_editor(self) -> None:
dlg = PresetEditorDialog(self.presets, self.current_preset, self, self.highlight_channels)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
self.presets = dlg.presets()
self.highlight_channels = dlg.highlight_channels()
chosen = dlg.selected_name()
self.current_preset = chosen if chosen in self.presets else next(iter(self.presets), "")
self._fill_preset_combo()
self._save()
self.log("Пресеты сохранены")
def _toggle_global(self, on: bool) -> None:
if on:
self.hotkeys.start()
else:
self.hotkeys.stop()
self._save()
def _on_global_key(self, which: str) -> None:
if QApplication.activeWindow() is not None:
return
self.step(-1 if which == "prev" else +1)
def closeEvent(self, event) -> None:
self._save()
self.hotkeys.stop()
self.client.close()
super().closeEvent(event)
def main() -> None:
os.environ.setdefault("QT_SCALE_FACTOR", str(UI_SCALE))
app = QApplication(sys.argv)
app.setStyle("Fusion")
cfg, _ = Config.load()
theme = cfg.theme if cfg.theme in THEMES else DEFAULT_THEME
palette = THEMES[theme][1]
app.setStyleSheet(build_qss(palette, make_icons(palette, theme)))
if not cfg.authorized:
dlg = AuthDialog()
if dlg.exec() != QDialog.DialogCode.Accepted:
sys.exit(0)
cfg.authorized = True
cfg.save()
window = SwitchController()
window.show()
sys.exit(app.exec())
if __name__ == "__main__":
main()