root
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
import time
|
|
|
|
from protocol import (
|
|
status_request,
|
|
set_relay_request,
|
|
enable_wifi_request,
|
|
toggle_ignition_request,
|
|
toggle_sensor_fault_request,
|
|
)
|
|
|
|
|
|
class PicoSimulator:
|
|
def __init__(self, controller):
|
|
self.controller = controller
|
|
self.last_status = None
|
|
self.primary_link = "rs485"
|
|
self.rs485_connected = True
|
|
self.backup_link_available = True
|
|
self.messages_sent = 0
|
|
self.messages_received = 0
|
|
self.last_message_time = None
|
|
|
|
def send_message(self, message):
|
|
self.messages_sent += 1
|
|
|
|
if not self.rs485_connected:
|
|
return {
|
|
"type": "error",
|
|
"success": False,
|
|
"error": "RS-485 disconnected"
|
|
}
|
|
|
|
response = self.controller.handle_message(message)
|
|
self.messages_received += 1
|
|
self.last_message_time = int(time.time())
|
|
return response
|
|
|
|
def get_status(self):
|
|
response = self.send_message(status_request())
|
|
|
|
if response.get("type") == "error":
|
|
if self.last_status:
|
|
self.last_status["network"]["rs485_connected"] = False
|
|
self.last_status["network"]["communication_lost"] = True
|
|
self.last_status["network"]["messages_sent"] = self.messages_sent
|
|
self.last_status["network"]["messages_received"] = self.messages_received
|
|
self.last_status["network"]["last_message_time"] = self.last_message_time
|
|
return self.last_status
|
|
|
|
return {
|
|
"timestamp": int(time.time()),
|
|
"vehicle": {"ignition_on": True},
|
|
"battery": {},
|
|
"temps": {},
|
|
"sensor_health": {},
|
|
"relays": {},
|
|
"network": {
|
|
"rs485_connected": False,
|
|
"communication_lost": True,
|
|
"wifi_enabled": False,
|
|
"wifi_override_active": False,
|
|
"starlink_enabled": False,
|
|
"messages_sent": self.messages_sent,
|
|
"messages_received": self.messages_received,
|
|
"last_message_time": self.last_message_time
|
|
},
|
|
"config": {}
|
|
}
|
|
|
|
self.last_status = response["data"]
|
|
self.last_status["network"]["rs485_connected"] = self.rs485_connected
|
|
self.last_status["network"]["communication_lost"] = False
|
|
self.last_status["network"]["messages_sent"] = self.messages_sent
|
|
self.last_status["network"]["messages_received"] = self.messages_received
|
|
self.last_status["network"]["last_message_time"] = self.last_message_time
|
|
return self.last_status
|
|
|
|
def set_relay(self, relay, state):
|
|
return self.send_message(set_relay_request(relay, state))
|
|
|
|
def enable_wifi(self, minutes=10):
|
|
return self.send_message(enable_wifi_request(minutes))
|
|
|
|
def toggle_ignition(self):
|
|
return self.send_message(toggle_ignition_request())
|
|
|
|
def toggle_sensor_fault(self, sensor):
|
|
return self.send_message(toggle_sensor_fault_request(sensor))
|
|
|
|
def disconnect_rs485(self):
|
|
self.rs485_connected = False
|
|
|
|
def restore_rs485(self):
|
|
self.rs485_connected = True
|
|
|
|
def get_comms(self):
|
|
return {
|
|
"primary": self.primary_link,
|
|
"backup_available": self.backup_link_available,
|
|
"rs485_connected": self.rs485_connected,
|
|
"messages_sent": self.messages_sent,
|
|
"messages_received": self.messages_received,
|
|
"last_message_time": self.last_message_time
|
|
}
|