paulg/venus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
36a07dacb951e143a1465b536b51e90056a4ef66
venus/dbus-no-foreign-land/nfl_tracking.py
Paul G 36a07dacb9 Extract shared signal-based D-Bus readers into lib/signal_reader.py 
- Added lib/signal_reader.py with SignalGpsReader, SignalMeteoReader, and
  SignalDepthReader that use PropertiesChanged signal subscriptions instead
  of polling via GetValue(), reducing D-Bus overhead at steady state.
- Each reader discovers its service dynamically, seeds its cache with a
  one-shot GetValue, then relies on signals for all subsequent updates.
- Refactored dbus-tides, dbus-windy-station, dbus-no-foreign-land,
  dbus-lightning, and dbus-meteoblue-forecast to import from the shared
  library, removing ~600 lines of duplicated _unwrap() helpers and
  per-service GPS/meteo/depth reader classes.
- Updated install.sh for all five services to deploy signal_reader.py
  to /data/lib/ on the target device.
- Updated build-package.sh for all five services to bundle
  signal_reader.py into the .tar.gz package.
- Updated README.md with the new lib/ entry in the project table and
  documented the shared D-Bus readers pattern.
- Bumped version numbers in affected services (e.g. nfl_tracking 2.0.1).

Made-with: Cursor
2026-03-27 01:03:16 +00:00

712 lines
28 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
NFL (No Foreign Land) Tracking for Venus OS
Sends boat position/track to noforeignland.com using D-Bus GPS data.
Publishes its own D-Bus service so status is visible via MQTT:
N/<vrm-id>/nfltracking/0/...
Features:
- Reads GPS position from Venus OS D-Bus
- Sends track data to the NFL API
- Settings adjustable via D-Bus/MQTT and Venus GUI
- 24h keepalive when boat is stationary
- Runs as a daemon with automatic restart
Author: Adapted from dbus-generator-ramp style
License: MIT
"""
import calendar
import json
import logging
import math
import os
import signal
import sys
import time
from pathlib import Path
# Add velib_python to path (Venus OS standard location)
sys.path.insert(1, os.path.join(os.path.dirname(__file__), 'ext', 'velib_python'))
sys.path.insert(1, '/opt/victronenergy/velib_python')
sys.path.insert(1, os.path.join(os.path.dirname(__file__), '..', 'lib'))
sys.path.insert(1, '/data/lib')
try:
from gi.repository import GLib
except ImportError:
print("ERROR: GLib not available. This script must run on Venus OS.")
sys.exit(1)
try:
import dbus
from dbus.mainloop.glib import DBusGMainLoop
from vedbus import VeDbusService
from settingsdevice import SettingsDevice
except ImportError as e:
print(f"ERROR: Required module not available: {e}")
print("This script must run on Venus OS.")
sys.exit(1)
from config import (
SERVICE_NAME, NFL_API_URL, NFL_PLUGIN_API_KEY,
TRACKING_CONFIG, CONFIG_DIR, TRACK_FILE, LOGGING_CONFIG,
)
from signal_reader import SignalGpsReader
import threading
# Version
VERSION = '2.0.1'
def validate_position(lat, lon):
"""Validate lat/lon for WGS84."""
try:
lat_f = float(lat)
lon_f = float(lon)
if not (-90 < lat_f < 90 and -180 < lon_f < 180):
return False
if abs(lat_f) < 0.01 and abs(lon_f) < 0.01:
return False # Likely GPS init
return True
except (TypeError, ValueError):
return False
def equirectangular_distance(lat1, lon1, lat2, lon2):
"""Distance in meters (approximate)."""
R = 6371000 # Earth radius meters
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dlam = math.radians(lon2 - lon1)
x = dlam * math.cos((phi1 + phi2) / 2)
y = dphi
return math.sqrt(x * x + y * y) * R
def send_track_to_api(track, boat_api_key, log):
"""POST track to NFL API. Returns True on success."""
if not track:
return False
if not boat_api_key or not boat_api_key.strip():
log.error("No boat API key configured")
return False
import urllib.request
import urllib.error
import urllib.parse
last_ts = track[-1][0]
params_dict = {
"timestamp": str(last_ts),
"track": json.dumps(track),
"boatApiKey": boat_api_key.strip(),
}
params = urllib.parse.urlencode(params_dict).encode()
headers = {
"X-NFL-API-Key": NFL_PLUGIN_API_KEY,
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "dbus-no-foreign-land/2.0 (Venus OS)",
}
req = urllib.request.Request(
NFL_API_URL,
data=params,
headers=headers,
method="POST",
)
# Debug: log full request (mask boatApiKey)
if log.isEnabledFor(logging.DEBUG):
debug_params = dict(params_dict)
key = boat_api_key.strip()
debug_params["boatApiKey"] = key[:4] + "***" + key[-4:] if len(key) > 8 else "***"
log.debug("API request URL: %s", NFL_API_URL)
log.debug("API request headers: %s", headers)
log.debug("API request body: %s", urllib.parse.urlencode(debug_params))
try:
with urllib.request.urlopen(req, timeout=30) as resp:
body = json.loads(resp.read().decode())
if body.get("status") == "ok":
log.info("Track sent successfully (%d points)", len(track))
return True
log.warning("API error: %s", body.get("message", "Unknown"))
return False
except urllib.error.HTTPError as e:
err_body = ""
try:
err_body = e.read().decode()
except Exception as read_err:
log.warning("Could not read error body: %s", read_err)
err_msg = e.reason
if err_body:
try:
err_json = json.loads(err_body)
err_msg = err_json.get("message", err_body[:300])
except json.JSONDecodeError:
err_msg = err_body[:300] if err_body else e.reason
log.error("HTTP %d: %s - %s", e.code, e.reason, err_msg)
# On 403, always log what we sent (masked) to help debug
if e.code == 403:
key = boat_api_key.strip()
masked_key = key[:4] + "***" + key[-4:] if len(key) > 8 else "***"
log.error("Request was: URL=%s | headers=%s | body: timestamp=%s track_len=%d boatApiKey=%s",
NFL_API_URL, headers, params_dict["timestamp"], len(track), masked_key)
return False
except Exception as e:
log.error("Send failed: %s", e)
return False
def load_track_from_file(path, max_points=None):
"""Load track points from JSONL file. Returns list of [timestamp_ms, lat, lon].
If max_points is set, returns only the most recent points (keeps tail).
"""
max_points = max_points or TRACKING_CONFIG.get('max_track_points', 10000)
track = []
p = Path(path)
if not p.exists():
return track
try:
with open(p) as f:
for line in f:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
st = time.strptime(obj["t"], "%Y-%m-%dT%H:%M:%SZ")
ts = int(calendar.timegm(st) * 1000)
lat, lon = float(obj["lat"]), float(obj["lon"])
if validate_position(lat, lon):
track.append([ts, lat, lon])
except (KeyError, ValueError, TypeError):
pass
if len(track) > max_points:
track = track[-max_points:]
except OSError:
pass
return track
class NflTrackingController:
"""
Main controller that coordinates:
- D-Bus monitoring of GPS position
- Track logging and API upload
- Publishing status to D-Bus (visible via MQTT)
"""
def __init__(self):
self._setup_logging()
self.logger = logging.getLogger('NflTracking')
self.logger.info(f"Initializing NFL Tracking v{VERSION}")
# State
self.track = []
self._send_in_progress = False
self.last_pos = None
self.last_log_time = 0
self.last_send_time = time.time()
self.last_send_time_success = 0 # Unix timestamp of last successful send
self.last_point_time = time.time()
# D-Bus connection
self.bus = dbus.SystemBus()
# Create our D-Bus service for publishing status
self._create_dbus_service()
# Set up settings (stored in Venus localsettings)
self._setup_settings()
# GPS reader
self.gps = SignalGpsReader(self.bus)
# Load existing track (capped to prevent OOM)
self.track = load_track_from_file(TRACK_FILE)
self._max_track_points = TRACKING_CONFIG.get('max_track_points', 10000)
# Ensure config is set (from _load_settings or defaults)
if not hasattr(self, 'boat_api_key'):
self.boat_api_key = ''
self.boat_name = ''
self.enabled = True
self.track_interval = TRACKING_CONFIG['track_interval']
self.min_distance = TRACKING_CONFIG['min_distance']
self.send_interval = TRACKING_CONFIG['send_interval']
self.ping_24h = TRACKING_CONFIG['ping_24h']
poll_interval_ms = min(30000, self.track_interval * 1000)
GLib.timeout_add(poll_interval_ms, self._main_loop)
self.logger.info(
f"Initialized. Poll interval: {poll_interval_ms}ms, "
f"track_interval={self.track_interval}s, send_interval={self.send_interval}min"
)
def _setup_logging(self):
"""Configure logging based on config."""
level = getattr(logging, LOGGING_CONFIG['level'], logging.INFO)
if LOGGING_CONFIG['include_timestamp']:
fmt = '%(asctime)s %(levelname)s %(name)s: %(message)s'
else:
fmt = '%(levelname)s %(name)s: %(message)s'
logging.basicConfig(level=level, format=fmt, stream=sys.stdout)
def _create_dbus_service(self):
"""Create our own D-Bus service for publishing status."""
self.logger.info(f"Creating D-Bus service: {SERVICE_NAME}")
max_retries = 5
retry_delay = 1.0
for attempt in range(max_retries):
try:
# register=False: add all paths first, then call register()
# See https://github.com/victronenergy/venus/wiki/dbus-api
self.dbus_service = VeDbusService(SERVICE_NAME, self.bus, register=False)
break
except dbus.exceptions.NameExistsException:
if attempt < max_retries - 1:
self.logger.warning(
f"D-Bus name exists, retrying in {retry_delay}s "
f"(attempt {attempt + 1}/{max_retries})"
)
time.sleep(retry_delay)
retry_delay *= 2
else:
self.logger.error("Failed to acquire D-Bus name after retries")
raise
# Management paths (required for Venus)
self.dbus_service.add_path('/Mgmt/ProcessName', 'dbus-no-foreign-land')
self.dbus_service.add_path('/Mgmt/ProcessVersion', VERSION)
self.dbus_service.add_path('/Mgmt/Connection', 'local')
# Device info
self.dbus_service.add_path('/DeviceInstance', 0)
self.dbus_service.add_path('/ProductId', 0xFFFF)
self.dbus_service.add_path('/ProductName', 'NFL Tracking')
self.dbus_service.add_path('/FirmwareVersion', VERSION)
self.dbus_service.add_path('/Connected', 1)
# Status paths (read-only)
def _status_text(p, v):
if v is not None and v in (0, 1, 2):
return ['Idle', 'Tracking', 'Sending'][v]
return 'Unknown'
self.dbus_service.add_path('/Status', 0, gettextcallback=_status_text)
self.dbus_service.add_path('/TrackPoints', 0)
self.dbus_service.add_path('/Stationary', 0)
self.dbus_service.add_path('/LastSendTimeAgo', '')
def _coord_text(p, v):
if v is not None and isinstance(v, (int, float)):
return f"{float(v):.6f}"
return "--"
self.dbus_service.add_path('/LastLatitude', 0.0, gettextcallback=_coord_text)
self.dbus_service.add_path('/LastLongitude', 0.0, gettextcallback=_coord_text)
self.dbus_service.add_path('/LastUpdate', 0)
self.dbus_service.add_path('/GpsConnected', 0)
self.dbus_service.add_path('/LastSendTime', 0,
gettextcallback=self._last_send_time_text)
# Action: send now (writable, 0->1 triggers send)
# Under /Settings/ so Venus GUI/MQTT exposes it for the switch
self.dbus_service.add_path('/Settings/SendNow', 0,
writeable=True,
onchangecallback=self._on_send_now)
# Writable settings (can be changed via D-Bus/MQTT)
self.dbus_service.add_path('/Settings/BoatApiKey', '',
writeable=True,
onchangecallback=self._on_setting_changed)
self.dbus_service.add_path('/Settings/BoatName', '',
writeable=True,
onchangecallback=self._on_setting_changed)
self.dbus_service.add_path('/Settings/Enabled', 1,
writeable=True,
onchangecallback=self._on_setting_changed)
def _int_text(p, v, suffix):
return f"{v} {suffix}" if v is not None else "--"
self.dbus_service.add_path('/Settings/TrackInterval', TRACKING_CONFIG['track_interval'],
writeable=True,
onchangecallback=self._on_setting_changed,
gettextcallback=lambda p, v: _int_text(p, v, "s"))
self.dbus_service.add_path('/Settings/MinDistance', TRACKING_CONFIG['min_distance'],
writeable=True,
onchangecallback=self._on_setting_changed,
gettextcallback=lambda p, v: _int_text(p, v, "m"))
self.dbus_service.add_path('/Settings/SendInterval', TRACKING_CONFIG['send_interval'],
writeable=True,
onchangecallback=self._on_setting_changed,
gettextcallback=lambda p, v: _int_text(p, v, "min"))
self.dbus_service.add_path('/Settings/Ping24h', 1 if TRACKING_CONFIG['ping_24h'] else 0,
writeable=True,
onchangecallback=self._on_setting_changed)
# Register after all paths added (required by Venus D-Bus API)
self.dbus_service.register()
self.logger.info("D-Bus service created")
def _update_last_send_time_ago(self):
"""Update /LastSendTimeAgo with Xh Ym format."""
if self.last_send_time_success <= 0:
self.dbus_service['/LastSendTimeAgo'] = 'Never'
else:
diff_sec = time.time() - self.last_send_time_success
total_m = int(diff_sec / 60)
h, m = total_m // 60, total_m % 60
self.dbus_service['/LastSendTimeAgo'] = f"{h}h {m}m"
def _last_send_time_text(self, path, value):
"""Format last send time for display."""
try:
if value is None or (isinstance(value, (int, float)) and value <= 0):
return "Never"
val = float(value)
except (TypeError, ValueError):
return "Never"
now = time.time()
diff = now - val
if diff < 60:
return "Just now"
if diff < 3600:
return f"{int(diff / 60)} min ago"
if diff < 86400:
return f"{int(diff / 3600)} h ago"
return time.strftime("%Y-%m-%d %H:%M", time.localtime(val))
def _on_send_now(self, path, value):
"""Handle 'Send now' action - trigger immediate send."""
if not value:
return True
self.logger.info("Send now requested via D-Bus/MQTT")
try:
self._do_send_now()
finally:
# Always reset switch; defer slightly so GUI/MQTT can process
def _reset():
self.dbus_service['/Settings/SendNow'] = 0
return False # one-shot
GLib.timeout_add(300, _reset)
return True
def _send_track_async(self, to_send, on_done):
"""Send track in background thread; on_done(ok) runs on main thread via GLib.idle_add."""
def _worker():
ok = send_track_to_api(to_send, self.boat_api_key, self.logger)
GLib.idle_add(lambda: on_done(ok))
t = threading.Thread(target=_worker, daemon=True)
t.start()
def _do_send_now(self):
"""Perform immediate track send to API (non-blocking)."""
if self._send_in_progress:
self.logger.warning("Send already in progress, skipping")
return
to_send = load_track_from_file(TRACK_FILE) if Path(TRACK_FILE).exists() else self.track
if not to_send:
self.logger.warning("No track points to send")
return
self._send_in_progress = True
self.dbus_service['/Status'] = 2 # Sending
def _on_done(ok):
self._send_in_progress = False
if ok:
self.last_send_time = time.time()
self.last_send_time_success = int(time.time())
self.dbus_service['/LastSendTime'] = self.last_send_time_success
self.track.clear()
try:
if Path(TRACK_FILE).exists():
Path(TRACK_FILE).unlink()
except OSError:
pass
self.dbus_service['/Status'] = 1 # Back to Tracking
self._send_track_async(to_send, _on_done)
def _on_setting_changed(self, path, value):
"""Handle setting changes from D-Bus/MQTT."""
self.logger.info(f"Setting changed: {path} = {value}")
if path == '/Settings/BoatApiKey':
self.boat_api_key = str(value) if value is not None else ''
self._save_setting('BoatApiKey', self.boat_api_key)
elif path == '/Settings/BoatName':
self.boat_name = str(value) if value is not None else ''
self._save_setting('BoatName', self.boat_name)
elif path == '/Settings/Enabled':
self.enabled = bool(value)
self._save_setting('Enabled', 1 if self.enabled else 0)
elif path == '/Settings/TrackInterval':
val = int(value) if value is not None else TRACKING_CONFIG['track_interval']
self.track_interval = max(10, min(3600, val))
self._save_setting('TrackInterval', self.track_interval)
elif path == '/Settings/MinDistance':
val = int(value) if value is not None else TRACKING_CONFIG['min_distance']
self.min_distance = max(1, min(10000, val))
self._save_setting('MinDistance', self.min_distance)
elif path == '/Settings/SendInterval':
val = int(value) if value is not None else TRACKING_CONFIG['send_interval']
self.send_interval = max(1, min(1440, val))
self._save_setting('SendInterval', self.send_interval)
elif path == '/Settings/Ping24h':
self.ping_24h = bool(value)
self._save_setting('Ping24h', 1 if self.ping_24h else 0)
return True
def _save_setting(self, name, value):
"""Save a setting to localsettings."""
if self.settings:
try:
self.settings[name] = value
except Exception as e:
self.logger.warning(f"Failed to save setting {name}: {e}")
def _setup_settings(self):
"""Set up persistent settings via Venus localsettings."""
self.settings = None
try:
settings_path = '/Settings/NflTracking'
settings_def = {
'BoatApiKey': [settings_path + '/BoatApiKey', '', 0, 0],
'BoatName': [settings_path + '/BoatName', '', 0, 0],
'Enabled': [settings_path + '/Enabled', 1, 0, 1],
'TrackInterval': [settings_path + '/TrackInterval', TRACKING_CONFIG['track_interval'], 10, 3600],
'MinDistance': [settings_path + '/MinDistance', TRACKING_CONFIG['min_distance'], 1, 10000],
'SendInterval': [settings_path + '/SendInterval', TRACKING_CONFIG['send_interval'], 1, 1440],
'Ping24h': [settings_path + '/Ping24h', 1 if TRACKING_CONFIG['ping_24h'] else 0, 0, 1],
}
self.settings = SettingsDevice(
self.bus,
settings_def,
self._on_persistent_setting_changed
)
if self.settings:
self._load_settings()
self.logger.info("Persistent settings initialized")
except Exception as e:
self.logger.warning(f"Could not initialize persistent settings: {e}")
self.logger.warning("Settings will not persist across restarts")
def _load_settings(self):
"""Load settings from Venus localsettings."""
if not self.settings:
return
try:
boat_api_key = str(self.settings['BoatApiKey']) if self.settings['BoatApiKey'] else ''
boat_name = str(self.settings['BoatName']) if self.settings['BoatName'] else ''
enabled = bool(self.settings['Enabled'])
track_interval = int(self.settings['TrackInterval'])
min_distance = int(self.settings['MinDistance'])
send_interval = int(self.settings['SendInterval'])
ping_24h = bool(self.settings['Ping24h'])
self.track_interval = max(10, min(3600, track_interval))
self.min_distance = max(1, min(10000, min_distance))
self.send_interval = max(1, min(1440, send_interval))
self.ping_24h = ping_24h
self.boat_api_key = boat_api_key
self.boat_name = boat_name
self.enabled = enabled
self.dbus_service['/Settings/BoatApiKey'] = boat_api_key
self.dbus_service['/Settings/BoatName'] = boat_name
self.dbus_service['/Settings/Enabled'] = 1 if enabled else 0
self.dbus_service['/Settings/TrackInterval'] = self.track_interval
self.dbus_service['/Settings/MinDistance'] = self.min_distance
self.dbus_service['/Settings/SendInterval'] = self.send_interval
self.dbus_service['/Settings/Ping24h'] = 1 if ping_24h else 0
self.logger.info(
f"Loaded settings: boat_api_key={'*' * 8 if boat_api_key else '(empty)'}, "
f"track_interval={self.track_interval}s, send_interval={self.send_interval}min"
)
except Exception as e:
self.logger.warning(f"Error loading settings: {e}")
def _on_persistent_setting_changed(self, setting, old_value, new_value):
"""Called when a persistent setting changes externally."""
self.logger.info(f"Persistent setting changed: {setting} = {new_value}")
self._load_settings()
def _main_loop(self) -> bool:
"""Main control loop - called periodically by GLib."""
try:
if not self.enabled:
self.dbus_service['/Status'] = 0 # Idle
self.dbus_service['/Stationary'] = 0
self._update_last_send_time_ago()
return True
self.dbus_service['/Status'] = 1 # Tracking
# Read GPS position
pos = self.gps.get_position()
self.dbus_service['/GpsConnected'] = 1 if pos else 0
if pos:
lat, lon = pos
if validate_position(lat, lon):
now_ms = int(time.time() * 1000)
now = time.time()
self.dbus_service['/LastLatitude'] = lat
self.dbus_service['/LastLongitude'] = lon
self.dbus_service['/LastUpdate'] = now_ms
force_save = (
self.ping_24h and self.last_point_time and
(now - self.last_point_time) >= 24 * 3600
)
if force_save or not self.last_pos:
do_log = True
elif now - self.last_log_time >= self.track_interval:
dist = equirectangular_distance(
self.last_pos[0], self.last_pos[1], lat, lon
)
do_log = dist >= self.min_distance
else:
do_log = False
# Stationary = has last_pos, hasn't moved min_distance
if self.last_pos and not do_log and not force_save:
self.dbus_service['/Stationary'] = 1
else:
self.dbus_service['/Stationary'] = 0
if do_log:
self.track.append([now_ms, lat, lon])
trimmed = False
if len(self.track) > self._max_track_points:
self.track = self.track[-self._max_track_points:]
# Trim file to match (prevents unbounded disk growth)
try:
Path(CONFIG_DIR).mkdir(parents=True, exist_ok=True)
with open(TRACK_FILE, 'w') as f:
for pt in self.track:
ts_ms, lat_pt, lon_pt = pt
t_str = time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts_ms / 1000)
)
f.write(json.dumps({"t": t_str, "lat": lat_pt, "lon": lon_pt}) + "\n")
trimmed = True
except OSError:
pass
self.last_pos = (lat, lon)
self.last_log_time = now
self.last_point_time = now
if not trimmed:
try:
Path(CONFIG_DIR).mkdir(parents=True, exist_ok=True)
with open(TRACK_FILE, "a") as f:
f.write(json.dumps({
"t": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
"lat": lat, "lon": lon
}) + "\n")
except OSError:
pass
self.logger.debug("Logged point: %.6f, %.6f", lat, lon)
# Maybe send (non-blocking; skip if send already in progress)
now = time.time()
if not self._send_in_progress and now - self.last_send_time >= self.send_interval * 60:
to_send = load_track_from_file(TRACK_FILE) if Path(TRACK_FILE).exists() else self.track
if to_send:
self._send_in_progress = True
self.dbus_service['/Status'] = 2 # Sending
def _on_done(ok):
self._send_in_progress = False
if ok:
self.last_send_time_success = int(time.time())
self.dbus_service['/LastSendTime'] = self.last_send_time_success
self.track.clear()
try:
if Path(TRACK_FILE).exists():
Path(TRACK_FILE).unlink()
except OSError:
pass
self.last_send_time = time.time()
self.dbus_service['/Status'] = 1 # Back to Tracking
self._send_track_async(to_send, _on_done)
else:
self.last_send_time = now
self.dbus_service['/TrackPoints'] = len(self.track)
self._update_last_send_time_ago()
# Update stationary when no GPS (not stationary, unknown)
if not pos:
self.dbus_service['/Stationary'] = 0
except dbus.exceptions.DBusException as e:
self.logger.warning("D-Bus error: %s", e)
except Exception as e:
self.logger.exception("Unexpected error: %s", e)
return True
def main():
"""Main entry point."""
DBusGMainLoop(set_as_default=True)
print("=" * 60)
print(f"NFL Tracking v{VERSION}")
print("=" * 60)
mainloop = None
def signal_handler(signum, frame):
"""Handle shutdown signals gracefully."""
try:
sig_name = signal.Signals(signum).name
except ValueError:
sig_name = str(signum)
logging.info(f"Received {sig_name}, shutting down...")
if mainloop is not None:
mainloop.quit()
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
try:
controller = NflTrackingController()
mainloop = GLib.MainLoop()
mainloop.run()
except KeyboardInterrupt:
print("\nShutdown requested")
except Exception as e:
logging.error(f"Fatal error: {e}", exc_info=True)
sys.exit(1)
finally:
logging.info("Service stopped")
if __name__ == '__main__':
main()
Reference in New Issue View Git Blame Copy Permalink