seo/control
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

터치 패널 핑 스트림 처리

Browse Source
This commit is contained in:
seo committed 2026-07-31 12:56:31 +09:00
1 parent 1ac3cf0355
commit be6e9e117b
2 files changed
+46 -22
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

touch_panel/panel.py
+45 -21
View File
@@ -104,7 +104,6 @@ GRAPH_POINTS = GRAPH_SECONDS * 1000 // SAMPLE_INTERVAL_MS
SYNOLOGY_RESTART_INTERVAL_MS = 1000
DISK_TEMP_INTERVAL_MS = SAMPLE_INTERVAL_MS
RPI_THROTTLE_INTERVAL_MS = SAMPLE_INTERVAL_MS
SYNOLOGY_PING_INTERVAL_MS = SAMPLE_INTERVAL_MS
SYNOLOGY_DISK_TEMP_INTERVAL_MS = SAMPLE_INTERVAL_MS
SYNOLOGY_HEALTH_INTERVAL_MS = SAMPLE_INTERVAL_MS
@@ -326,6 +325,7 @@ class StatusTile(QFrame):
self.write_steady = None
self.error_steady = None
self.error_blink = False
self.error_blink_seconds = self.ERROR_BLINK_SECONDS
self.title = QLabel(title)
self.title.setObjectName("statusTitle")
self.detail = QLabel("-")
@@ -363,12 +363,14 @@ class StatusTile(QFrame):
write_steady=None,
error_steady=None,
error_blink=False,
error_blink_seconds=None,
):
now = time.monotonic()
self.read_steady = read_steady
self.write_steady = write_steady
self.error_steady = error_steady
self.error_blink = error_blink
self.error_blink_seconds = error_blink_seconds or self.ERROR_BLINK_SECONDS
if read_activity:
self.last_read_at = now
if write_activity:
@@ -385,7 +387,7 @@ class StatusTile(QFrame):
read_on = self.read_steady if self.read_steady is not None else now - self.last_read_at <= self.ACTIVITY_HOLD_SECONDS
write_on = self.write_steady if self.write_steady is not None else now - self.last_write_at <= self.ACTIVITY_HOLD_SECONDS
if self.error_blink:
error_on = int(now / self.ERROR_BLINK_SECONDS) % 2 == 0
error_on = int(now / self.error_blink_seconds) % 2 == 0
else:
error_on = self.error_steady if self.error_steady is not None else now - self.last_error_at <= self.ERROR_HOLD_SECONDS
self.read_dot.set_steady(read_on)
@@ -650,10 +652,13 @@ class TouchPanel(QMainWindow):
self.synology_health_timer.start(SYNOLOGY_HEALTH_INTERVAL_MS)
self.synology_ping_proc = QProcess(self)
self.synology_ping_buffer = ""
self.synology_ping_proc.readyReadStandardOutput.connect(self.synology_ping_ready)
self.synology_ping_proc.readyReadStandardError.connect(self.synology_ping_error)
self.synology_ping_proc.finished.connect(self.synology_ping_finished)
self.synology_ping_timer = QTimer(self)
self.synology_ping_timer.timeout.connect(self.ping_synology)
self.synology_ping_timer.start(SYNOLOGY_PING_INTERVAL_MS)
self.synology_ping_timer.start(1000)
self.rpi_throttle_proc = QProcess(self)
self.rpi_throttle_proc.finished.connect(self.rpi_throttle_finished)
@@ -883,26 +888,40 @@ class TouchPanel(QMainWindow):
def ping_synology(self):
if self.synology_ping_proc.state() != QProcess.NotRunning:
return
self.synology_ping_proc.start("/usr/bin/ping", ["-n", "-c", "1", "-W", "1", SYNOLOGY_PING_HOST])
self.synology_ping_buffer = ""
self.synology_ping_proc.start("/usr/bin/ping", ["-n", "-i", "0.05", SYNOLOGY_PING_HOST])
def synology_ping_ready(self):
output = bytes(self.synology_ping_proc.readAllStandardOutput()).decode("utf-8", "ignore")
self.synology_ping_buffer += output
while "\n" in self.synology_ping_buffer:
line, self.synology_ping_buffer = self.synology_ping_buffer.split("\n", 1)
latency_ms = self.parse_ping_latency(line)
if latency_ms is not None:
self.render_synology_ping(latency_ms)
def synology_ping_error(self):
output = bytes(self.synology_ping_proc.readAllStandardError()).decode("utf-8", "ignore").strip()
if output:
self.rows["synology_ping"].set_state("fail", read_steady=False, write_steady=False, error_steady=False, error_blink=True)
def synology_ping_finished(self, exit_code, exit_status):
del exit_status
output = (
bytes(self.synology_ping_proc.readAllStandardOutput()).decode("utf-8", "ignore")
+ bytes(self.synology_ping_proc.readAllStandardError()).decode("utf-8", "ignore")
)
latency_ms = None
for token in output.replace("\n", " ").split():
del exit_code, exit_status
self.rows["synology_ping"].set_state("fail", read_steady=False, write_steady=False, error_steady=False, error_blink=True)
QTimer.singleShot(1000, self.ping_synology)
@staticmethod
def parse_ping_latency(line):
for token in line.split():
if not token.startswith("time="):
continue
try:
latency_ms = float(token.split("=", 1)[1])
return float(token.split("=", 1)[1])
except ValueError:
latency_ms = None
break
if exit_code != 0 or latency_ms is None:
self.rows["synology_ping"].set_state("fail", read_steady=False, write_steady=False, error_steady=False, error_blink=True)
return
return None
return None
def render_synology_ping(self, latency_ms):
detail = f"{latency_ms:.1f}ms"
if latency_ms <= 10:
self.rows["synology_ping"].set_state(detail, read_steady=True, write_steady=False, error_steady=False)
@@ -929,12 +948,17 @@ class TouchPanel(QMainWindow):
except ValueError:
self.rows["rpi_throttle"].set_state("N/A", read_steady=False, write_steady=False, error_steady=True)
return
current = flags & 0xF
undervoltage_now = bool(flags & 0x1)
throttled_now = bool(flags & 0x4)
history = flags & 0xF0000
if current:
self.rows["rpi_throttle"].set_state(f"0x{flags:x}", read_steady=False, write_steady=False, error_steady=True)
if undervoltage_now and throttled_now:
self.rows["rpi_throttle"].set_state("UV+TH", read_steady=False, write_steady=False, error_steady=False, error_blink=True, error_blink_seconds=0.1)
elif throttled_now:
self.rows["rpi_throttle"].set_state("THROT", read_steady=False, write_steady=False, error_steady=False, error_blink=True, error_blink_seconds=0.5)
elif undervoltage_now:
self.rows["rpi_throttle"].set_state("UV", read_steady=False, write_steady=False, error_steady=True)
elif history:
self.rows["rpi_throttle"].set_state(f"0x{flags:x}", read_steady=False, write_steady=True, error_steady=False)
self.rows["rpi_throttle"].set_state("HIST", read_steady=False, write_steady=True, error_steady=False)
else:
self.rows["rpi_throttle"].set_state("OK", read_steady=True, write_steady=False, error_steady=False)