qsethuk/can_sniffer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix install script for working shutdown command, fix psql errors exception

Browse Source
This commit is contained in:
Alexander Poletaev
2026-01-27 18:32:24 +03:00
parent 28f8845536
commit 328fb5500e
7 changed files with 196 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
can_sniffer/deploy/install.sh
View File

@@ -150,9 +150,25 @@ chmod -R 755 "$INSTALL_DIR"
chmod 700 "$DATA_DIR"
chmod 755 "$LOGS_DIR"
# Добавляем пользователя в группы для доступа к CAN и UART
# Добавляем пользователя в группы для доступа к CAN, UART и I2C
usermod -aG dialout "$SERVICE_USER" 2>/dev/null || true
usermod -aG plugdev "$SERVICE_USER" 2>/dev/null || true
usermod -aG i2c "$SERVICE_USER" 2>/dev/null || true
usermod -aG gpio "$SERVICE_USER" 2>/dev/null || true
# === 6.1. Настройка sudoers для shutdown/reboot ===
log_info "Configuring sudoers for power management..."
SUDOERS_FILE="/etc/sudoers.d/can-sniffer"
cat > "$SUDOERS_FILE" << EOF
# Allow can_sniffer service user to run power commands without password
$SERVICE_USER ALL=(ALL) NOPASSWD: /sbin/shutdown
$SERVICE_USER ALL=(ALL) NOPASSWD: /sbin/reboot
$SERVICE_USER ALL=(ALL) NOPASSWD: /usr/sbin/shutdown
$SERVICE_USER ALL=(ALL) NOPASSWD: /usr/sbin/reboot
EOF
chmod 440 "$SUDOERS_FILE"
log_info "Sudoers configured for user: $SERVICE_USER"
# === 7. Установка systemd сервисов ===
log_info "Installing systemd services..."

12
can_sniffer/src/flipper/page_manager.py
View File

@@ -224,6 +224,18 @@ class PageManager:
# Just return None, content will be sent anyway
return None
if command.cmd_type == CommandType.SCROLL_UP:
page = self.get_current_page()
if page and hasattr(page, 'scroll_up'):
page.scroll_up()
return None
if command.cmd_type == CommandType.SCROLL_DOWN:
page = self.get_current_page()
if page and hasattr(page, 'scroll_down'):
page.scroll_down()
return None
return None
def get_last_result(self) -> Optional[str]:

98
can_sniffer/src/flipper/pages/system_info.py
View File

@@ -1,24 +1,45 @@
"""
System Information Page.
System Information Page with scrolling support.
Displays Raspberry Pi system metrics on Flipper Zero.
"""
from flipper.pages.base import InfoPage
import socket
from flipper.pages.base import BasePage
from flipper.protocol import PageContent, PageType
from flipper.providers.system_provider import SystemProvider
class SystemInfoPage(InfoPage):
def get_ip_address() -> str:
"""Get the primary IP address."""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0.1)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except Exception:
return "0.0.0.0"
class SystemInfoPage(BasePage):
"""
Page displaying Raspberry Pi system metrics.
Page displaying Raspberry Pi system metrics with scrolling.
Shows:
- IP address
- CPU temperature
- Power consumption
- Fan RPM
- Input voltage
Supports Up/Down scrolling when content exceeds display.
"""
MAX_VISIBLE_LINES = 4
def __init__(self):
super().__init__(
name="system_info",
@@ -26,30 +47,87 @@ class SystemInfoPage(InfoPage):
icon="cpu"
)
self._provider = SystemProvider()
self._scroll_offset = 0
self._all_lines: list[str] = []
def get_lines(self) -> list[str]:
"""Get system info lines for display."""
# Force refresh to get fresh data
def _build_all_lines(self) -> list[str]:
"""Build complete list of info lines."""
self._provider.refresh()
data = self._provider.get_data()
lines = [
f"IP: {get_ip_address()}",
f"CPU: {data.cpu_temp:.1f}C {self._get_temp_indicator(data.cpu_temp)}",
f"Power: {data.power_watts:.2f}W",
]
# Fan RPM (may not be available on all systems)
# Fan RPM
if data.fan_rpm > 0:
lines.append(f"Fan: {data.fan_rpm} RPM")
else:
lines.append("Fan: N/A")
# Input voltage
if data.input_voltage > 0:
lines.append(f"Input: {data.input_voltage:.2f}V")
# CPU details
if data.cpu_volts > 0:
lines.append(f"CPU V: {data.cpu_volts:.2f}V")
if data.cpu_amps > 0:
lines.append(f"CPU A: {data.cpu_amps:.2f}A")
return lines
def get_content(self) -> PageContent:
"""Get page content with scroll support."""
self._all_lines = self._build_all_lines()
# Calculate visible window
max_offset = max(0, len(self._all_lines) - self.MAX_VISIBLE_LINES)
self._scroll_offset = min(self._scroll_offset, max_offset)
visible_lines = self._all_lines[
self._scroll_offset:self._scroll_offset + self.MAX_VISIBLE_LINES
]
# Add scroll indicators
if self._scroll_offset > 0:
visible_lines[0] = "^ " + visible_lines[0]
if self._scroll_offset < max_offset:
visible_lines[-1] = "v " + visible_lines[-1]
return PageContent(
page_type=PageType.INFO,
title=self.title,
lines=visible_lines,
icon=self.icon
)
def handle_select(self, index: int) -> None:
"""Handle Up/Down for scrolling (index 0=up, 1=down)."""
max_offset = max(0, len(self._all_lines) - self.MAX_VISIBLE_LINES)
if index == 0 and self._scroll_offset > 0:
self._scroll_offset -= 1
elif index == 1 and self._scroll_offset < max_offset:
self._scroll_offset += 1
return None
def scroll_up(self) -> None:
"""Scroll content up."""
if self._scroll_offset > 0:
self._scroll_offset -= 1
def scroll_down(self) -> None:
"""Scroll content down."""
max_offset = max(0, len(self._all_lines) - self.MAX_VISIBLE_LINES)
if self._scroll_offset < max_offset:
self._scroll_offset += 1
def on_enter(self) -> None:
"""Reset scroll when entering page."""
self._scroll_offset = 0
def _get_temp_indicator(self, temp: float) -> str:
"""
Get temperature status indicator.

9
can_sniffer/src/flipper/protocol.py
View File

@@ -43,6 +43,8 @@ class CommandType(Enum):
CONFIRM = "CONFIRM"
CANCEL = "CANCEL"
REFRESH = "REFRESH"
SCROLL_UP = "SCROLL:up"
SCROLL_DOWN = "SCROLL:down"
@dataclass
@@ -191,4 +193,11 @@ class Protocol:
if line == "CMD:REFRESH":
return Command(CommandType.REFRESH)
# CMD:SCROLL:up / CMD:SCROLL:down
if line == "CMD:SCROLL:up":
return Command(CommandType.SCROLL_UP)
if line == "CMD:SCROLL:down":
return Command(CommandType.SCROLL_DOWN)
return None

66
can_sniffer/src/flipper/providers/system_provider.py
View File

@@ -6,6 +6,7 @@ and system control actions (shutdown, reboot).
"""
import os
import shutil
import subprocess
from typing import Optional
from dataclasses import dataclass
@@ -16,6 +17,34 @@ from logger import get_logger
logger = get_logger(__name__)
# vcgencmd path - try common locations
_VCGENCMD_PATHS = [
"/usr/bin/vcgencmd",
"/opt/vc/bin/vcgencmd",
]
def _find_vcgencmd() -> str:
"""Find vcgencmd binary."""
# Check common paths first
for path in _VCGENCMD_PATHS:
if Path(path).exists():
logger.debug(f"Found vcgencmd at: {path}")
return path
# Check if in PATH
which_result = shutil.which("vcgencmd")
if which_result:
logger.debug(f"Found vcgencmd in PATH: {which_result}")
return which_result
logger.warning("vcgencmd not found, system metrics may not work")
return "vcgencmd" # Fallback
# Find vcgencmd at module load time
VCGENCMD = _find_vcgencmd()
@dataclass
class SystemData:
@@ -46,14 +75,16 @@ class SystemProvider(BaseProvider):
Read a hardware metric using vcgencmd.
Args:
args: Command arguments
args: Command arguments (first element will be replaced with VCGENCMD path)
strip_chars: Characters to strip from value
Returns:
Float value or None on error
"""
try:
output = subprocess.check_output(args, stderr=subprocess.DEVNULL).decode("utf-8")
# Replace first argument with resolved path
cmd = [VCGENCMD] + args[1:] if args else [VCGENCMD]
output = subprocess.check_output(cmd, stderr=subprocess.DEVNULL).decode("utf-8")
value_str = output.split("=")[1].strip().rstrip(strip_chars)
return float(value_str)
except Exception as e:
@@ -63,7 +94,7 @@ class SystemProvider(BaseProvider):
def _read_cpu_temp(self) -> float:
"""Read CPU temperature in Celsius."""
# Try vcgencmd first
result = self._read_metric(["vcgencmd", "measure_temp"], "'C")
result = self._read_metric([VCGENCMD, "measure_temp"], "'C")
if result is not None:
return result
@@ -81,17 +112,17 @@ class SystemProvider(BaseProvider):
def _read_cpu_volts(self) -> float:
"""Read CPU core voltage."""
result = self._read_metric(["vcgencmd", "pmic_read_adc", "VDD_CORE_V"], "V")
result = self._read_metric([VCGENCMD, "pmic_read_adc", "VDD_CORE_V"], "V")
return result if result is not None else 0.0
def _read_cpu_amps(self) -> float:
"""Read CPU core current."""
result = self._read_metric(["vcgencmd", "pmic_read_adc", "VDD_CORE_A"], "A")
result = self._read_metric([VCGENCMD, "pmic_read_adc", "VDD_CORE_A"], "A")
return result if result is not None else 0.0
def _read_input_voltage(self) -> float:
"""Read external 5V input voltage."""
result = self._read_metric(["vcgencmd", "pmic_read_adc", "EXT5V_V"], "V")
result = self._read_metric([VCGENCMD, "pmic_read_adc", "EXT5V_V"], "V")
return result if result is not None else 0.0
def _read_power_watts(self) -> float:
@@ -101,7 +132,10 @@ class SystemProvider(BaseProvider):
Reads all PMIC ADC values and calculates wattage.
"""
try:
output = subprocess.check_output(["vcgencmd", "pmic_read_adc"]).decode("utf-8")
output = subprocess.check_output(
[VCGENCMD, "pmic_read_adc"],
stderr=subprocess.DEVNULL
).decode("utf-8")
lines = output.strip().split("\n")
amperages = {}
@@ -139,7 +173,8 @@ class SystemProvider(BaseProvider):
return wattage
except Exception:
except Exception as e:
logger.debug(f"Failed to read power watts: {e}")
return 0.0
def _read_fan_rpm(self) -> int:
@@ -229,14 +264,16 @@ class SystemProvider(BaseProvider):
"""
try:
if delay_minutes > 0:
cmd = f"sudo shutdown -P +{delay_minutes}"
cmd = ["sudo", "shutdown", "-P", f"+{delay_minutes}"]
else:
cmd = "sudo shutdown -P now"
cmd = ["sudo", "shutdown", "-P", "now"]
subprocess.Popen(cmd, shell=True)
logger.info(f"Executing shutdown: {' '.join(cmd)}")
subprocess.Popen(cmd)
return True
except Exception as e:
logger.error(f"Shutdown failed: {e}")
self._last_error = str(e)
return False
@@ -248,10 +285,13 @@ class SystemProvider(BaseProvider):
True if command executed
"""
try:
subprocess.Popen("sudo reboot", shell=True)
cmd = ["sudo", "reboot"]
logger.info(f"Executing reboot: {' '.join(cmd)}")
subprocess.Popen(cmd)
return True
except Exception as e:
logger.error(f"Reboot failed: {e}")
self._last_error = str(e)
return False
@@ -263,7 +303,7 @@ class SystemProvider(BaseProvider):
True if command executed
"""
try:
subprocess.call("sudo shutdown -c", shell=True)
subprocess.call(["sudo", "shutdown", "-c"])
return True
except Exception as e:

8
can_sniffer/src/postgresql_handler/postgresql_client.py
View File

@@ -510,8 +510,8 @@ class PostgreSQLClient:
"is_extended": bool(is_extended) if is_extended is not None else (can_id > 0x7FF)
})
# Отправляем в PostgreSQL напрямую (не через очередь)
sent_count = self._send_messages_batch(messages)
# Отправляем в PostgreSQL с retry механизмом
sent_count = self._send_messages_batch_with_retry(messages)
if sent_count > 0:
# Помечаем успешно отправленные как обработанные
@@ -522,6 +522,10 @@ class PostgreSQLClient:
f"Synced {sent_count} messages from SQLite, marked {marked} as processed"
)
return sent_count
else:
self.logger.warning(
f"Failed to sync {len(messages)} messages from SQLite to PostgreSQL"
)
return 0

16
flip_monitor/can_monitor.c
View File

@@ -483,13 +483,13 @@ static void draw_footer(Canvas* canvas, PageType page_type, uint8_t total_pages,
const char* hint = "";
switch(page_type) {
case PageTypeInfo:
hint = "";
hint = "Up/Dn=Scroll";
break;
case PageTypeMenu:
hint = "OK=Select";
hint = "Up/Dn OK=Select";
break;
case PageTypeConfirm:
hint = "OK=Yes Back=No";
hint = "Up/Dn OK Back=No";
break;
}
uint16_t hint_width = canvas_string_width(canvas, hint);
@@ -693,7 +693,10 @@ static void input_callback(InputEvent* event, void* ctx) {
if(app->conn_state == StateConnected && app->page.data_valid) {
switch(event->key) {
case InputKeyUp:
if(app->page.page_type == PageTypeMenu && app->page.selected_index > 0) {
if(app->page.page_type == PageTypeInfo) {
// Scroll up on info pages
uart_send(app, "CMD:SCROLL:up\n");
} else if(app->page.page_type == PageTypeMenu && app->page.selected_index > 0) {
app->page.selected_index--;
view_port_update(app->view_port);
} else if(app->page.page_type == PageTypeConfirm) {
@@ -703,7 +706,10 @@ static void input_callback(InputEvent* event, void* ctx) {
break;
case InputKeyDown:
if(app->page.page_type == PageTypeMenu &&
if(app->page.page_type == PageTypeInfo) {
// Scroll down on info pages
uart_send(app, "CMD:SCROLL:down\n");
} else if(app->page.page_type == PageTypeMenu &&
app->page.selected_index < app->page.action_count - 1) {
app->page.selected_index++;
view_port_update(app->view_port);