All checks were successful
Build Raspberry Pi Binary / build-arm64 (push) Successful in 12m6s
192 lines
5.8 KiB
Python
192 lines
5.8 KiB
Python
import sys
|
|
import os
|
|
import json
|
|
import math
|
|
from PySide6.QtGui import QGuiApplication
|
|
from PySide6.QtQml import QQmlApplicationEngine
|
|
from PySide6.QtCore import QObject, Property, Signal, Slot, QTimer
|
|
|
|
# Try importing w1thermsensor, handle failure for dev environments
|
|
try:
|
|
from w1thermsensor import W1ThermSensor, Sensor
|
|
HAS_W1 = True
|
|
except ImportError:
|
|
HAS_W1 = False
|
|
print("Warning: w1thermsensor not found. Running in simulation mode.")
|
|
|
|
from PySide6.QtCore import QObject, Property, Signal, Slot, QTimer, QThread
|
|
|
|
class SensorWorker(QObject):
|
|
tempsUpdated = Signal(float, float) # left, right
|
|
|
|
def __init__(self, config, demo_mode=False):
|
|
super().__init__()
|
|
self.config = config
|
|
self.demo_mode = demo_mode
|
|
self.running = False
|
|
self.left_sensor = None
|
|
self.right_sensor = None
|
|
self._sim_step = 0
|
|
|
|
def init_sensors(self):
|
|
if not HAS_W1:
|
|
return
|
|
|
|
print(f"Worker: Loading sensors from {self.config}")
|
|
try:
|
|
available_sensors = W1ThermSensor.get_available_sensors()
|
|
|
|
# Left Sensor
|
|
left_id = self.config.get("left_sensor_id")
|
|
if left_id:
|
|
self.left_sensor = next((s for s in available_sensors if s.id == left_id), None)
|
|
if not self.left_sensor:
|
|
# Fallback
|
|
self.left_sensor = W1ThermSensor(sensor_id=left_id)
|
|
|
|
# Auto-assign Left if missing but sensors exist
|
|
if not self.left_sensor and available_sensors:
|
|
self.left_sensor = available_sensors[0]
|
|
|
|
# Right Sensor
|
|
right_id = self.config.get("right_sensor_id")
|
|
if right_id:
|
|
self.right_sensor = W1ThermSensor(sensor_id=right_id)
|
|
|
|
except Exception as e:
|
|
print(f"Worker Error init sensors: {e}")
|
|
|
|
@Slot()
|
|
def start_polling(self):
|
|
self.running = True
|
|
self.init_sensors()
|
|
|
|
while self.running:
|
|
left_val = 0.0
|
|
right_val = 0.0
|
|
|
|
# --- Left Read ---
|
|
if self.left_sensor:
|
|
try:
|
|
left_val = self.left_sensor.get_temperature()
|
|
# print(f"L: {left_val}")
|
|
except: pass
|
|
elif self.demo_mode:
|
|
self._sim_step += 0.1
|
|
left_val = 5 + 10 * math.sin(self._sim_step)
|
|
|
|
# --- Right Read ---
|
|
if self.right_sensor:
|
|
try:
|
|
right_val = self.right_sensor.get_temperature()
|
|
except: pass
|
|
elif self.demo_mode:
|
|
right_val = 10 + 10 * math.cos(self._sim_step)
|
|
|
|
# Emit result
|
|
self.tempsUpdated.emit(left_val, right_val)
|
|
|
|
# Sleep (30 seconds)
|
|
QThread.msleep(30000)
|
|
|
|
@Slot()
|
|
def stop(self):
|
|
self.running = False
|
|
|
|
class Backend(QObject):
|
|
leftTempChanged = Signal()
|
|
rightTempChanged = Signal()
|
|
|
|
def __init__(self, demo_mode=False):
|
|
super().__init__()
|
|
self._left_temp = 0.0
|
|
self._right_temp = 0.0
|
|
|
|
# Load Config
|
|
self.config = self.load_config()
|
|
|
|
# Setup Threading
|
|
self.thread = QThread()
|
|
self.worker = SensorWorker(self.config, demo_mode)
|
|
self.worker.moveToThread(self.thread)
|
|
|
|
# Connect signals
|
|
self.thread.started.connect(self.worker.start_polling)
|
|
self.worker.tempsUpdated.connect(self.handle_temps_update)
|
|
|
|
# Start
|
|
self.thread.start()
|
|
|
|
def handle_temps_update(self, left, right):
|
|
# Update Properties (This runs on Main Thread)
|
|
if abs(self._left_temp - left) > 0.1:
|
|
self._left_temp = left
|
|
self.leftTempChanged.emit()
|
|
|
|
if abs(self._right_temp - right) > 0.1:
|
|
self._right_temp = right
|
|
self.rightTempChanged.emit()
|
|
|
|
def load_config(self):
|
|
# Determine path based on run mode (Script vs Frozen/Compiled)
|
|
if getattr(sys, 'frozen', False):
|
|
# If run as a compiled exe, look in the same directory as the executable
|
|
application_path = os.path.dirname(sys.executable)
|
|
else:
|
|
# If run as a script, look in the directory of the script
|
|
application_path = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
config_path = os.path.join(application_path, "config.json")
|
|
if os.path.exists(config_path):
|
|
try:
|
|
with open(config_path, 'r') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"Error loading config: {e}")
|
|
return {}
|
|
|
|
# Cleanup
|
|
def stop(self):
|
|
if self.worker:
|
|
self.worker.stop()
|
|
self.thread.quit()
|
|
self.thread.wait()
|
|
|
|
@Property(float, notify=leftTempChanged)
|
|
def leftTemp(self):
|
|
return self._left_temp
|
|
|
|
@Property(float, notify=rightTempChanged)
|
|
def rightTemp(self):
|
|
return self._right_temp
|
|
|
|
def main():
|
|
# Redirect output to log.txt
|
|
sys.stdout = open("log.txt", "w", buffering=1)
|
|
sys.stderr = sys.stdout
|
|
print("--- Volvo Display Log ---")
|
|
|
|
# Check for demo mode
|
|
demo_mode = "--demo" in sys.argv
|
|
if demo_mode:
|
|
print("Starting in DEMO MODE (Simulation Enabled)")
|
|
|
|
app = QGuiApplication(sys.argv)
|
|
engine = QQmlApplicationEngine()
|
|
|
|
backend = Backend(demo_mode=demo_mode)
|
|
engine.rootContext().setContextProperty("backend", backend)
|
|
|
|
# Get absolute path to main.qml
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
qml_file = os.path.join(current_dir, "main.qml")
|
|
|
|
engine.load(qml_file)
|
|
if not engine.rootObjects():
|
|
sys.exit(-1)
|
|
|
|
sys.exit(app.exec())
|
|
|
|
if __name__ == "__main__":
|
|
main()
|