Support for 480x480 HDMI display, optimized sensor polling, and touch tools
All checks were successful
Build Raspberry Pi Binary / build-arm64 (push) Successful in 12m6s
All checks were successful
Build Raspberry Pi Binary / build-arm64 (push) Successful in 12m6s
This commit is contained in:
199
main.py
199
main.py
@@ -14,29 +14,118 @@ 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.demo_mode = demo_mode
|
||||
self._left_temp = 0.0
|
||||
self._right_temp = 0.0
|
||||
|
||||
self.left_sensor = None
|
||||
self.right_sensor = None
|
||||
|
||||
# Load Config
|
||||
self.config = self.load_config()
|
||||
|
||||
# Initialize Sensors
|
||||
self.init_sensors()
|
||||
# Setup Threading
|
||||
self.thread = QThread()
|
||||
self.worker = SensorWorker(self.config, demo_mode)
|
||||
self.worker.moveToThread(self.thread)
|
||||
|
||||
# Timer for polling
|
||||
self.timer = QTimer()
|
||||
self.timer.timeout.connect(self.update_temps)
|
||||
self.timer.start(2000) # Poll every 2 seconds
|
||||
# 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)
|
||||
@@ -56,99 +145,21 @@ class Backend(QObject):
|
||||
print(f"Error loading config: {e}")
|
||||
return {}
|
||||
|
||||
def init_sensors(self):
|
||||
if not HAS_W1:
|
||||
print("W1ThermSensor not installed.")
|
||||
return
|
||||
|
||||
print(f"Loading sensors using config: {self.config}")
|
||||
try:
|
||||
available_sensors = W1ThermSensor.get_available_sensors()
|
||||
print(f"Available W1 sensors: {[s.id for s in available_sensors]}")
|
||||
|
||||
# Map Left Sensor
|
||||
left_id = self.config.get("left_sensor_id")
|
||||
if left_id:
|
||||
try:
|
||||
# Check if ID exists in available to be sure
|
||||
found = next((s for s in available_sensors if s.id == left_id), None)
|
||||
if found:
|
||||
self.left_sensor = found
|
||||
print(f"✅ Successfully bound Left Sensor to {left_id}")
|
||||
else:
|
||||
print(f"❌ Configured Left Sensor {left_id} not found in available list!")
|
||||
# Try direct init anyway just in case
|
||||
self.left_sensor = W1ThermSensor(sensor_id=left_id)
|
||||
except Exception as e:
|
||||
print(f"❌ Failed to bind Left Sensor {left_id}: {e}")
|
||||
|
||||
# Auto-assign fallback
|
||||
if not self.left_sensor and len(available_sensors) > 0:
|
||||
self.left_sensor = available_sensors[0]
|
||||
print(f"⚠️ Auto-assigned Left Sensor to {self.left_sensor.id}")
|
||||
|
||||
# Map Right Sensor
|
||||
right_id = self.config.get("right_sensor_id")
|
||||
if right_id:
|
||||
try:
|
||||
self.right_sensor = W1ThermSensor(sensor_id=right_id)
|
||||
print(f"Bound Right Sensor to {right_id}")
|
||||
except Exception as e:
|
||||
print(f"Failed to bind Right Sensor {right_id}: {e}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error initializing sensors: {e}")
|
||||
|
||||
def update_temps(self):
|
||||
# Simulation counter
|
||||
if not hasattr(self, '_sim_step'):
|
||||
self._sim_step = 0
|
||||
self._sim_step += 0.1
|
||||
|
||||
# Left
|
||||
if self.left_sensor:
|
||||
try:
|
||||
t = self.left_sensor.get_temperature()
|
||||
print(f"Left Sensor ({self.left_sensor.id}) Read: {t}°C")
|
||||
self._set_left_temp(t)
|
||||
except Exception as e:
|
||||
print(f"Error reading left sensor: {e}")
|
||||
elif self.demo_mode:
|
||||
# Simulation
|
||||
print("Left Sensor Missing - Simulating...")
|
||||
sim_val = 5 + 10 * math.sin(self._sim_step)
|
||||
self._set_left_temp(sim_val)
|
||||
|
||||
# Right
|
||||
if self.right_sensor:
|
||||
try:
|
||||
t = self.right_sensor.get_temperature()
|
||||
self._set_right_temp(t)
|
||||
except Exception as e:
|
||||
print(f"Error reading right sensor: {e}")
|
||||
elif self.demo_mode:
|
||||
# Simulation
|
||||
sim_val = 10 + 10 * math.cos(self._sim_step)
|
||||
self._set_right_temp(sim_val)
|
||||
# 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
|
||||
|
||||
def _set_left_temp(self, val):
|
||||
if abs(self._left_temp - val) > 0.1:
|
||||
self._left_temp = val
|
||||
self.leftTempChanged.emit()
|
||||
|
||||
@Property(float, notify=rightTempChanged)
|
||||
def rightTemp(self):
|
||||
return self._right_temp
|
||||
|
||||
def _set_right_temp(self, val):
|
||||
if abs(self._right_temp - val) > 0.1:
|
||||
self._right_temp = val
|
||||
self.rightTempChanged.emit()
|
||||
|
||||
def main():
|
||||
# Redirect output to log.txt
|
||||
sys.stdout = open("log.txt", "w", buffering=1)
|
||||
|
||||
Reference in New Issue
Block a user