Paul G
36a07dacb9
- 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
698 lines
26 KiB
Python
698 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
dbus-tides -- Local Tide Prediction for Venus OS
|
|
|
|
Monitors vessel depth, detects tidal patterns from observations,
|
|
predicts future tides using a pure-Python harmonic engine, and
|
|
publishes merged data on D-Bus for the venus-html5-app dashboard.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
import threading
|
|
from collections import deque
|
|
|
|
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("ERROR: Required module not available: %s" % e)
|
|
print("This script must run on Venus OS.")
|
|
sys.exit(1)
|
|
|
|
from config import (
|
|
SERVICE_NAME, VERSION,
|
|
DEPTH_SAMPLE_INTERVAL, DEPTH_HISTORY_HOURS, SPEED_THRESHOLD_MS,
|
|
PREDICTION_HOURS,
|
|
GPS_SAMPLE_INTERVAL, GPS_HISTORY_SIZE, GPS_HISTORY_SAMPLE_INTERVAL,
|
|
STATIONARY_RADIUS_METERS, MODEL_RERUN_RADIUS_METERS,
|
|
MODEL_RERUN_INTERVAL,
|
|
LOGGING_CONFIG,
|
|
)
|
|
from signal_reader import SignalGpsReader, SignalDepthReader
|
|
from depth_recorder import DepthRecorder
|
|
from tide_detector import TideDetector
|
|
from tide_predictor import TidePredictor
|
|
from tide_merger import TideMerger
|
|
from tide_adapter import TideAdapter
|
|
|
|
|
|
def haversine_distance(lat1, lon1, lat2, lon2):
|
|
"""Great-circle distance in meters."""
|
|
R = 6371000
|
|
phi1, phi2 = math.radians(lat1), math.radians(lat2)
|
|
dphi = math.radians(lat2 - lat1)
|
|
dlam = math.radians(lon2 - lon1)
|
|
a = (math.sin(dphi / 2) ** 2 +
|
|
math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2)
|
|
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
|
|
|
|
|
class MovementDetector:
|
|
"""Determines whether the vessel has moved beyond a threshold."""
|
|
|
|
def __init__(self, history_size, threshold_meters):
|
|
self.history_size = history_size
|
|
self.threshold_meters = threshold_meters
|
|
self.positions = deque(maxlen=history_size)
|
|
self.last_sample_time = 0
|
|
|
|
def add_position(self, lat, lon, now=None):
|
|
if now is None:
|
|
now = time.time()
|
|
self.positions.append((now, lat, lon))
|
|
self.last_sample_time = now
|
|
|
|
def is_moving(self, current_lat, current_lon):
|
|
if not self.positions:
|
|
return False
|
|
for _, hist_lat, hist_lon in self.positions:
|
|
dist = haversine_distance(hist_lat, hist_lon,
|
|
current_lat, current_lon)
|
|
if dist >= self.threshold_meters:
|
|
return True
|
|
return False
|
|
|
|
|
|
class TidesController:
|
|
"""Coordinates all tide components and D-Bus publishing."""
|
|
|
|
def __init__(self):
|
|
self._setup_logging()
|
|
self.logger = logging.getLogger('Tides')
|
|
self.logger.info("Initializing dbus-tides v%s", VERSION)
|
|
|
|
self.bus = dbus.SystemBus()
|
|
self._create_dbus_service()
|
|
self._setup_settings()
|
|
|
|
self.gps = SignalGpsReader(self.bus)
|
|
self.depth_reader = SignalDepthReader(self.bus)
|
|
self.depth_recorder = DepthRecorder()
|
|
self.tide_detector = TideDetector()
|
|
self.tide_predictor = TidePredictor()
|
|
self.tide_merger = TideMerger()
|
|
self.tide_adapter = TideAdapter()
|
|
|
|
self.movement_stationary = MovementDetector(
|
|
GPS_HISTORY_SIZE, STATIONARY_RADIUS_METERS)
|
|
self.movement_model = MovementDetector(
|
|
GPS_HISTORY_SIZE, MODEL_RERUN_RADIUS_METERS)
|
|
|
|
self.current_lat = None
|
|
self.current_lon = None
|
|
self.model_lat = None
|
|
self.model_lon = None
|
|
self.last_depth_sample = 0
|
|
self.last_gps_check = 0
|
|
self.last_gps_history_sample = 0
|
|
self.last_prediction_run = 0
|
|
self.last_event_refresh = 0
|
|
self._predict_in_progress = False
|
|
|
|
GLib.timeout_add(1000, self._main_loop)
|
|
self.logger.info("Initialized -- sampling depth every %ds",
|
|
DEPTH_SAMPLE_INTERVAL)
|
|
|
|
def _setup_logging(self):
|
|
level = getattr(logging, LOGGING_CONFIG['level'], logging.INFO)
|
|
fmt = ('%(asctime)s %(levelname)s %(name)s: %(message)s'
|
|
if LOGGING_CONFIG['include_timestamp']
|
|
else '%(levelname)s %(name)s: %(message)s')
|
|
logging.basicConfig(level=level, format=fmt, stream=sys.stdout)
|
|
|
|
def _create_dbus_service(self):
|
|
self.logger = logging.getLogger('Tides')
|
|
self.logger.info("Creating D-Bus service: %s", SERVICE_NAME)
|
|
|
|
max_retries = 5
|
|
retry_delay = 1.0
|
|
for attempt in range(max_retries):
|
|
try:
|
|
self.dbus_service = VeDbusService(
|
|
SERVICE_NAME, self.bus, register=False)
|
|
break
|
|
except dbus.exceptions.NameExistsException:
|
|
if attempt < max_retries - 1:
|
|
self.logger.warning(
|
|
"D-Bus name exists, retrying in %.1fs (%d/%d)",
|
|
retry_delay, attempt + 1, max_retries)
|
|
time.sleep(retry_delay)
|
|
retry_delay *= 2
|
|
else:
|
|
raise
|
|
|
|
svc = self.dbus_service
|
|
svc.add_path('/Mgmt/ProcessName', 'dbus-tides')
|
|
svc.add_path('/Mgmt/ProcessVersion', VERSION)
|
|
svc.add_path('/Mgmt/Connection', 'local')
|
|
svc.add_path('/DeviceInstance', 0)
|
|
svc.add_path('/ProductId', 0xA162)
|
|
svc.add_path('/ProductName', 'Tide Prediction')
|
|
svc.add_path('/FirmwareVersion', VERSION)
|
|
svc.add_path('/Connected', 1)
|
|
|
|
def _status_text(p, v):
|
|
labels = {0: 'Idle', 1: 'Calibrating', 2: 'Ready', 3: 'Error'}
|
|
return labels.get(v, 'Unknown') if v is not None else 'Unknown'
|
|
|
|
svc.add_path('/Status', 0, gettextcallback=_status_text)
|
|
svc.add_path('/ErrorMessage', '')
|
|
svc.add_path('/IsStationary', 0)
|
|
svc.add_path('/Depth/Current', None)
|
|
svc.add_path('/Depth/History', '')
|
|
svc.add_path('/Depth/FullHistory', '')
|
|
svc.add_path('/Tide/Predictions', '')
|
|
svc.add_path('/Tide/Station/Predictions', '')
|
|
svc.add_path('/Tide/Local/Predictions', '')
|
|
svc.add_path('/Tide/Local/TimeOffset', 0.0)
|
|
svc.add_path('/Tide/Local/AmpScale', 1.0)
|
|
svc.add_path('/Tide/Local/MatchCount', 0)
|
|
svc.add_path('/Tide/Local/Status', 0)
|
|
for slot in ('NextHigh1', 'NextHigh2', 'NextLow1', 'NextLow2',
|
|
'PrevHigh1', 'PrevHigh2', 'PrevLow1', 'PrevLow2'):
|
|
svc.add_path(f'/Tide/{slot}/Time', None)
|
|
svc.add_path(f'/Tide/{slot}/Depth', None)
|
|
svc.add_path(f'/Tide/{slot}/ObsTime', None)
|
|
svc.add_path(f'/Tide/{slot}/ObsDepth', None)
|
|
svc.add_path('/Tide/LastModelRun', 0)
|
|
svc.add_path('/Tide/DataSource', '')
|
|
svc.add_path('/Tide/Station/Id', '')
|
|
svc.add_path('/Tide/Station/Name', '')
|
|
svc.add_path('/Tide/Station/Distance', None)
|
|
svc.add_path('/Tide/Station/Lat', None)
|
|
svc.add_path('/Tide/Station/Lon', None)
|
|
svc.add_path('/Tide/Station/Type', '')
|
|
svc.add_path('/Tide/Station/RefId', '')
|
|
svc.add_path('/Tide/NearbyStations', '[]')
|
|
svc.add_path('/Tide/ChartDepth', 0.0)
|
|
svc.add_path('/Tide/DatumOffset', 0.0)
|
|
svc.add_path('/Tide/ModelLocation/Lat', None)
|
|
svc.add_path('/Tide/ModelLocation/Lon', None)
|
|
svc.add_path('/Settings/Enabled', 1,
|
|
writeable=True,
|
|
onchangecallback=self._on_setting_changed)
|
|
svc.add_path('/Settings/MinTideAmplitude', 0.3,
|
|
writeable=True,
|
|
onchangecallback=self._on_setting_changed)
|
|
svc.add_path('/Settings/Units', 0,
|
|
writeable=True,
|
|
onchangecallback=self._on_setting_changed)
|
|
svc.add_path('/Settings/DatumOffset', -1.0,
|
|
writeable=True,
|
|
onchangecallback=self._on_setting_changed)
|
|
svc.add_path('/Settings/StationOverride', '',
|
|
writeable=True,
|
|
onchangecallback=self._on_setting_changed)
|
|
|
|
svc.register()
|
|
self.logger.info("D-Bus service created")
|
|
|
|
def _setup_settings(self):
|
|
self.settings = None
|
|
self.enabled = True
|
|
try:
|
|
path = '/Settings/Tides'
|
|
settings_def = {
|
|
'Enabled': [path + '/Enabled', 1, 0, 1],
|
|
'Units': [path + '/Units', 0, 0, 1],
|
|
}
|
|
self.settings = SettingsDevice(
|
|
self.bus, settings_def,
|
|
self._on_persistent_setting_changed)
|
|
if self.settings:
|
|
self.enabled = bool(self.settings['Enabled'])
|
|
self.dbus_service['/Settings/Enabled'] = (
|
|
1 if self.enabled else 0)
|
|
self.dbus_service['/Settings/Units'] = (
|
|
int(self.settings['Units']))
|
|
self.logger.info("Persistent settings initialized")
|
|
except Exception as e:
|
|
self.logger.warning("Could not init persistent settings: %s", e)
|
|
|
|
def _on_persistent_setting_changed(self, setting, old_value, new_value):
|
|
self.logger.info("Setting changed: %s = %s", setting, new_value)
|
|
if self.settings:
|
|
self.enabled = bool(self.settings['Enabled'])
|
|
|
|
def _on_setting_changed(self, path, value):
|
|
self.logger.info("Setting changed: %s = %s", path, value)
|
|
if path == '/Settings/Enabled':
|
|
self.enabled = bool(value)
|
|
if self.settings:
|
|
try:
|
|
self.settings['Enabled'] = 1 if self.enabled else 0
|
|
except Exception:
|
|
pass
|
|
elif path == '/Settings/MinTideAmplitude':
|
|
try:
|
|
self.tide_detector.min_amplitude = float(value)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
elif path == '/Settings/Units':
|
|
if self.settings:
|
|
try:
|
|
self.settings['Units'] = int(value)
|
|
except Exception:
|
|
pass
|
|
elif path == '/Settings/StationOverride':
|
|
self.tide_predictor.station_override = str(value) if value else None
|
|
self.last_prediction_run = 0
|
|
return True
|
|
|
|
def _main_loop(self):
|
|
try:
|
|
if not self.enabled:
|
|
self.dbus_service['/Status'] = 0
|
|
return True
|
|
|
|
now = time.time()
|
|
self._check_gps(now)
|
|
self._check_depth(now)
|
|
self._check_predictions(now)
|
|
self._refresh_event_slots(now)
|
|
|
|
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 _check_gps(self, now):
|
|
if now - self.last_gps_check < GPS_SAMPLE_INTERVAL:
|
|
return
|
|
|
|
pos = self.gps.get_position()
|
|
if pos:
|
|
self.current_lat, self.current_lon = pos
|
|
|
|
if (now - self.last_gps_history_sample
|
|
>= GPS_HISTORY_SAMPLE_INTERVAL):
|
|
self.movement_stationary.add_position(
|
|
self.current_lat, self.current_lon, now)
|
|
self.movement_model.add_position(
|
|
self.current_lat, self.current_lon, now)
|
|
self.last_gps_history_sample = now
|
|
|
|
is_stationary = not self.movement_stationary.is_moving(
|
|
self.current_lat, self.current_lon)
|
|
self.dbus_service['/IsStationary'] = 1 if is_stationary else 0
|
|
|
|
self.last_gps_check = now
|
|
|
|
def _check_depth(self, now):
|
|
if now - self.last_depth_sample < DEPTH_SAMPLE_INTERVAL:
|
|
return
|
|
|
|
speed = self.gps.get_speed()
|
|
if speed is not None and speed > SPEED_THRESHOLD_MS:
|
|
self.last_depth_sample = now
|
|
return
|
|
|
|
depth = self.depth_reader.get_depth()
|
|
if depth is None:
|
|
self.last_depth_sample = now
|
|
return
|
|
|
|
result = self.depth_recorder.add_reading(
|
|
depth, self.current_lat, self.current_lon, now)
|
|
|
|
smoothed = self.depth_recorder.get_latest_depth()
|
|
if smoothed is not None:
|
|
self.dbus_service['/Depth/Current'] = round(smoothed, 2)
|
|
|
|
if result is not None:
|
|
history = self.depth_recorder.get_history(
|
|
lat=self.current_lat, lon=self.current_lon)
|
|
new_event = self.tide_detector.update(history)
|
|
|
|
self._recalibrate_if_moved(history)
|
|
|
|
now_ts = time.time()
|
|
display_cutoff = now_ts - 24 * 3600
|
|
obs_points = [
|
|
{'ts': int(ts), 'depth': round(d, 2)}
|
|
for ts, d in history if ts > display_cutoff
|
|
]
|
|
self.dbus_service['/Depth/History'] = json.dumps(obs_points)
|
|
|
|
full_obs = [
|
|
{'ts': int(ts), 'depth': round(d, 2)}
|
|
for ts, d in history
|
|
]
|
|
self.dbus_service['/Depth/FullHistory'] = json.dumps(full_obs)
|
|
|
|
if new_event is not None:
|
|
self._refresh_adaptation()
|
|
|
|
self.last_depth_sample = now
|
|
|
|
def _recalibrate_if_moved(self, history):
|
|
"""Recalibrate chart_depth if vessel has moved to a new position."""
|
|
if not self.tide_merger.needs_recalibration(
|
|
self.current_lat, self.current_lon):
|
|
return
|
|
pred_ts, pred_heights, _ = (
|
|
self.tide_predictor.get_last_prediction())
|
|
if pred_ts is None:
|
|
return
|
|
if self.tide_merger.calibrate(
|
|
history, pred_ts, pred_heights,
|
|
lat=self.current_lat, lon=self.current_lon):
|
|
self._publish_adaptation(pred_ts, pred_heights,
|
|
self.tide_predictor._last_extrema)
|
|
|
|
def _refresh_adaptation(self):
|
|
"""Recompute adaptation from observed events and last prediction."""
|
|
pred_ts, pred_heights, pred_extrema = (
|
|
self.tide_predictor.get_last_prediction())
|
|
if pred_extrema is None:
|
|
return
|
|
|
|
observed_events = self.tide_detector.get_recent_events(20)
|
|
if not observed_events:
|
|
return
|
|
|
|
if self.tide_adapter.update(observed_events, pred_extrema):
|
|
self._publish_adaptation(pred_ts, pred_heights, pred_extrema)
|
|
|
|
def _refresh_event_slots(self, now):
|
|
"""Re-classify next/prev tide events every 60s using cached extrema.
|
|
|
|
Keeps the dbus slots current as predicted event times pass,
|
|
without waiting for a new prediction run or observed event.
|
|
"""
|
|
if now - self.last_event_refresh < 60:
|
|
return
|
|
|
|
self.last_event_refresh = now
|
|
|
|
pred_ts, pred_heights, extrema = (
|
|
self.tide_predictor.get_last_prediction())
|
|
if extrema is None:
|
|
return
|
|
|
|
chart_depth = self.tide_merger.chart_depth or 0.0
|
|
|
|
best_extrema = extrema
|
|
if self.tide_adapter.status > 0:
|
|
best_extrema = self.tide_adapter.adapt_extrema(extrema)
|
|
|
|
calibrated_extrema = [
|
|
(ts, chart_depth + h, t) for ts, h, t in best_extrema]
|
|
next_tides = self.tide_merger.get_next_tides(
|
|
calibrated_extrema, count=2)
|
|
prev_tides = self.tide_merger.get_prev_tides(
|
|
calibrated_extrema, count=2)
|
|
observed_events = self.tide_detector.get_recent_events(20)
|
|
self._publish_tide_events(next_tides, prev_tides, observed_events)
|
|
|
|
def _check_predictions(self, now):
|
|
if self._predict_in_progress:
|
|
return
|
|
if self.current_lat is None or self.current_lon is None:
|
|
return
|
|
|
|
needs_model = False
|
|
|
|
if self.model_lat is None:
|
|
needs_model = True
|
|
elif now - self.last_prediction_run >= MODEL_RERUN_INTERVAL:
|
|
needs_model = True
|
|
elif self.movement_model.is_moving(
|
|
self.current_lat, self.current_lon):
|
|
needs_model = True
|
|
|
|
if needs_model:
|
|
self._predict_in_progress = True
|
|
thread = threading.Thread(
|
|
target=self._run_prediction,
|
|
args=(self.current_lat, self.current_lon, now),
|
|
daemon=True)
|
|
thread.start()
|
|
|
|
def _run_prediction(self, lat, lon, now):
|
|
"""Run in background thread."""
|
|
try:
|
|
self.logger.info(
|
|
"Running tide prediction for %.2f, %.2f", lat, lon)
|
|
self.dbus_service['/Status'] = 1
|
|
|
|
history = self.depth_recorder.get_history(lat=lat, lon=lon)
|
|
|
|
if not self.tide_predictor.update_location(
|
|
lat, lon, depth_history=history):
|
|
self.dbus_service['/Status'] = 3
|
|
self.dbus_service['/ErrorMessage'] = 'No constituent data'
|
|
return
|
|
|
|
pred_ts, pred_heights, extrema = (
|
|
self.tide_predictor.predict(
|
|
now - DEPTH_HISTORY_HOURS * 3600,
|
|
duration_hours=DEPTH_HISTORY_HOURS + PREDICTION_HOURS))
|
|
if pred_ts is None:
|
|
self.dbus_service['/Status'] = 3
|
|
self.dbus_service['/ErrorMessage'] = 'Prediction failed'
|
|
return
|
|
|
|
if history:
|
|
self.tide_merger.calibrate(
|
|
history, pred_ts, pred_heights, lat=lat, lon=lon)
|
|
|
|
if history:
|
|
self.tide_detector.update(
|
|
[(ts, d) for ts, d in history])
|
|
observed_events = self.tide_detector.get_recent_events(20)
|
|
if observed_events and extrema:
|
|
self.tide_adapter.update(observed_events, extrema)
|
|
|
|
self._publish_adaptation(pred_ts, pred_heights, extrema)
|
|
|
|
manual_offset = self.dbus_service['/Settings/DatumOffset']
|
|
if manual_offset is not None and manual_offset >= 0:
|
|
datum_offset = float(manual_offset)
|
|
else:
|
|
datum_offset = self.tide_predictor.estimate_datum_offset()
|
|
self.dbus_service['/Tide/DatumOffset'] = round(datum_offset, 3)
|
|
|
|
self.model_lat = lat
|
|
self.model_lon = lon
|
|
self.last_prediction_run = time.time()
|
|
|
|
self.dbus_service['/Tide/LastModelRun'] = int(
|
|
self.last_prediction_run)
|
|
self.dbus_service['/Tide/DataSource'] = (
|
|
self.tide_predictor.current_source or '')
|
|
self._publish_station_info()
|
|
self.dbus_service['/Tide/ModelLocation/Lat'] = lat
|
|
self.dbus_service['/Tide/ModelLocation/Lon'] = lon
|
|
self.dbus_service['/ErrorMessage'] = ''
|
|
self.dbus_service['/Status'] = 2
|
|
|
|
self.logger.info(
|
|
"Prediction complete, %d extrema found",
|
|
len(extrema) if extrema else 0)
|
|
|
|
except Exception as e:
|
|
self.logger.exception("Prediction error: %s", e)
|
|
self.dbus_service['/Status'] = 3
|
|
self.dbus_service['/ErrorMessage'] = str(e)[:200]
|
|
finally:
|
|
self._predict_in_progress = False
|
|
|
|
def _publish_station_info(self):
|
|
"""Publish selected station metadata and nearby stations list."""
|
|
svc = self.dbus_service
|
|
st = self.tide_predictor.current_station
|
|
if st:
|
|
svc['/Tide/Station/Id'] = st.get('id', '')
|
|
svc['/Tide/Station/Name'] = st.get('name', '')
|
|
svc['/Tide/Station/Distance'] = round(
|
|
st.get('distance', 0) / 1000, 1)
|
|
svc['/Tide/Station/Lat'] = st.get('lat')
|
|
svc['/Tide/Station/Lon'] = st.get('lon')
|
|
svc['/Tide/Station/Type'] = st.get('type', 'R')
|
|
svc['/Tide/Station/RefId'] = st.get('ref_id', '')
|
|
else:
|
|
svc['/Tide/Station/Id'] = ''
|
|
svc['/Tide/Station/Name'] = ''
|
|
svc['/Tide/Station/Distance'] = None
|
|
svc['/Tide/Station/Lat'] = None
|
|
svc['/Tide/Station/Lon'] = None
|
|
svc['/Tide/Station/Type'] = ''
|
|
svc['/Tide/Station/RefId'] = ''
|
|
|
|
nearby = self.tide_predictor.nearby_stations
|
|
svc['/Tide/NearbyStations'] = json.dumps(nearby)
|
|
|
|
def _publish_adaptation(self, pred_ts, pred_heights, extrema):
|
|
"""Build and publish station + local curves and next-tide data."""
|
|
history = self.depth_recorder.get_history(
|
|
lat=self.current_lat, lon=self.current_lon)
|
|
chart_depth = self.tide_merger.chart_depth or 0.0
|
|
|
|
self.dbus_service['/Tide/ChartDepth'] = round(chart_depth, 3)
|
|
station_depths = [chart_depth + h for h in pred_heights]
|
|
|
|
local_ts = None
|
|
local_depths = None
|
|
best_extrema = extrema
|
|
|
|
if self.tide_adapter.status > 0:
|
|
adapted_ts, adapted_h = self.tide_adapter.adapt_curve(
|
|
pred_ts, pred_heights)
|
|
local_depths = [chart_depth + h for h in adapted_h]
|
|
local_ts = adapted_ts
|
|
best_extrema = self.tide_adapter.adapt_extrema(extrema)
|
|
|
|
residual = self.tide_adapter.compute_residual_correction(
|
|
history, self.tide_predictor.predict_at, chart_depth)
|
|
if residual:
|
|
local_depths = self.tide_adapter.apply_residual_correction(
|
|
local_ts, local_depths, residual)
|
|
|
|
curves = self.tide_merger.build_dual_curves(
|
|
history, pred_ts, station_depths, local_ts, local_depths)
|
|
|
|
self.dbus_service['/Depth/History'] = json.dumps(
|
|
curves.get('observed', []))
|
|
|
|
full_obs = [
|
|
{'ts': int(ts), 'depth': round(d, 2)}
|
|
for ts, d in history
|
|
]
|
|
self.dbus_service['/Depth/FullHistory'] = json.dumps(full_obs)
|
|
self.dbus_service['/Tide/Station/Predictions'] = json.dumps(
|
|
curves.get('station', []))
|
|
self.dbus_service['/Tide/Local/Predictions'] = json.dumps(
|
|
curves.get('local', []))
|
|
|
|
best_curve = curves.get('local', []) or curves.get('station', [])
|
|
self.dbus_service['/Tide/Predictions'] = json.dumps(best_curve)
|
|
|
|
self.dbus_service['/Tide/Local/TimeOffset'] = round(
|
|
self.tide_adapter.time_offset, 1)
|
|
self.dbus_service['/Tide/Local/AmpScale'] = round(
|
|
self.tide_adapter.amp_scale, 3)
|
|
self.dbus_service['/Tide/Local/MatchCount'] = (
|
|
self.tide_adapter.match_count)
|
|
self.dbus_service['/Tide/Local/Status'] = self.tide_adapter.status
|
|
|
|
if best_extrema:
|
|
calibrated_extrema = [
|
|
(ts, chart_depth + h, t) for ts, h, t in best_extrema]
|
|
next_tides = self.tide_merger.get_next_tides(
|
|
calibrated_extrema, count=2)
|
|
prev_tides = self.tide_merger.get_prev_tides(
|
|
calibrated_extrema, count=2)
|
|
observed_events = self.tide_detector.get_recent_events(20)
|
|
self._publish_tide_events(
|
|
next_tides, prev_tides, observed_events)
|
|
|
|
def _publish_tide_events(self, next_tides, prev_tides,
|
|
observed_events=None):
|
|
svc = self.dbus_service
|
|
|
|
obs_by_type = {'high': [], 'low': []}
|
|
if observed_events:
|
|
for obs_ts, obs_depth, obs_type in observed_events:
|
|
obs_by_type.setdefault(obs_type, []).append(
|
|
(obs_ts, obs_depth))
|
|
|
|
def _find_obs(pred_ts, tide_type, max_window=10800):
|
|
"""Find the observed event closest to a predicted timestamp."""
|
|
best, best_dist = None, max_window
|
|
for obs_ts, obs_depth in obs_by_type.get(tide_type, []):
|
|
dist = abs(obs_ts - pred_ts)
|
|
if dist < best_dist:
|
|
best_dist = dist
|
|
best = (obs_ts, obs_depth)
|
|
return best
|
|
|
|
def _set_slot(slot, pred_list, idx, tide_type):
|
|
if idx < len(pred_list):
|
|
ts, depth = pred_list[idx]
|
|
svc[f'/Tide/{slot}/Time'] = int(ts)
|
|
svc[f'/Tide/{slot}/Depth'] = round(depth, 2)
|
|
obs = _find_obs(ts, tide_type)
|
|
if obs:
|
|
svc[f'/Tide/{slot}/ObsTime'] = int(obs[0])
|
|
svc[f'/Tide/{slot}/ObsDepth'] = round(obs[1], 2)
|
|
else:
|
|
svc[f'/Tide/{slot}/ObsTime'] = None
|
|
svc[f'/Tide/{slot}/ObsDepth'] = None
|
|
else:
|
|
for suffix in ('Time', 'Depth', 'ObsTime', 'ObsDepth'):
|
|
svc[f'/Tide/{slot}/{suffix}'] = None
|
|
|
|
highs = next_tides.get('high', [])
|
|
lows = next_tides.get('low', [])
|
|
_set_slot('NextHigh1', highs, 0, 'high')
|
|
_set_slot('NextHigh2', highs, 1, 'high')
|
|
_set_slot('NextLow1', lows, 0, 'low')
|
|
_set_slot('NextLow2', lows, 1, 'low')
|
|
|
|
prev_highs = prev_tides.get('high', [])
|
|
prev_lows = prev_tides.get('low', [])
|
|
_set_slot('PrevHigh1', prev_highs, 0, 'high')
|
|
_set_slot('PrevHigh2', prev_highs, 1, 'high')
|
|
_set_slot('PrevLow1', prev_lows, 0, 'low')
|
|
_set_slot('PrevLow2', prev_lows, 1, 'low')
|
|
|
|
|
|
def main():
|
|
DBusGMainLoop(set_as_default=True)
|
|
|
|
print("=" * 60)
|
|
print("dbus-tides v%s" % VERSION)
|
|
print("=" * 60)
|
|
|
|
mainloop = None
|
|
|
|
def signal_handler(signum, frame):
|
|
try:
|
|
sig_name = signal.Signals(signum).name
|
|
except ValueError:
|
|
sig_name = str(signum)
|
|
logging.info("Received %s, shutting down...", sig_name)
|
|
if mainloop is not None:
|
|
mainloop.quit()
|
|
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
|
|
try:
|
|
controller = TidesController()
|
|
mainloop = GLib.MainLoop()
|
|
mainloop.run()
|
|
except KeyboardInterrupt:
|
|
print("\nShutdown requested")
|
|
except Exception as e:
|
|
logging.error("Fatal error: %s", e, exc_info=True)
|
|
sys.exit(1)
|
|
finally:
|
|
logging.info("Service stopped")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|