######################################################################### # Simple Python3 TPMS scanner # # Sept 2025 ride # # Originally based on "tpms-bleak.py" at https://github.com/andi38/TPMS # # Added QT window and other functionality # # Lots of fixes and help from ChatGPT actually. # # It was useful. https://chatgpt.com/g/g-cKrO5THiU-python-code-fixer # ######################################################################### #basic system stuff and async modules import sys, asyncio, qasync #GUI stuff from PyQt5 import QtWidgets, QtCore, QtGui from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QDesktopWidget from PyQt5.QtGui import QPainter, QBrush, QColor, QFont, QIcon from PyQt5.QtCore import Qt, QRect import json import shutil import subprocess # Instead of BLE sensors, use a hardcoded mapping of example 433MHz sensor IDs to tire positions # These would be the IDs emitted by the tire sensors seen by rtl_433. Replace with your real IDs. ID_TO_TIRE = { "0xA1B2": "FL", "0xA1B3": "FR", "0xA1B4": "RL", "0xA1B5": "RR", } def kpa_to_psi(kpa: float) -> float: # 1 kPa = 0.1450377377 psi return round(kpa * 0.1450377377, 1) async def rtl_433_reader(window): """Spawn rtl_433 (if available) and parse its JSON output line-by-line. This expects rtl_433 to be installed and in PATH. If rtl_433 is not available the function will simulate no data but keep the GUI running. """ # check for rtl_433 rtl_path = shutil.which("rtl_433") if not rtl_path: # rtl_433 not found; periodically repaint and do nothing while True: await asyncio.sleep(5.0) window.trigger_repaint() return # run rtl_433 in JSON mode; -F json prints JSON lines # limit sample rate and use a generic device; the user can customize command cmd = [rtl_path, "-F", "json"] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: # read lines asynchronously assert proc.stdout is not None while True: line = await proc.stdout.readline() if not line: # process ended break try: obj = json.loads(line.decode(errors="ignore")) except Exception: # ignore parse errors continue # Expected JSON structure from rtl_433 varies by sensor. We'll look for a sensible # id field (id, id_hex, idString, etc.) and a pressure field in kPa or similar. sensor_id = None for key in ("id_hex", "id", "idString", "sensor_id"): if key in obj: sensor_id = str(obj[key]) break if sensor_id is None and "id" in obj: sensor_id = str(obj.get("id")) # Normalize sensor_id to 0x... hex string if numeric try: if sensor_id is not None and sensor_id.isdigit(): sensor_id = hex(int(sensor_id)) except Exception: pass tire = ID_TO_TIRE.get(sensor_id) # Some sensors report pressure as 'pressure_kpa' or 'pressure' in kPa pressure_kpa = None for pkey in ("pressure_kpa", "pressure_kPa", "pressure", "press_kpa"): if pkey in obj: try: pressure_kpa = float(obj[pkey]) except Exception: pressure_kpa = None break temp_c = None for tkey in ("temperature_C", "temperature", "temp_c", "temp"): if tkey in obj: try: temp_c = float(obj[tkey]) except Exception: temp_c = None break batt = None for bkey in ("battery", "battery_level", "batt"): if bkey in obj: try: batt = float(obj[bkey]) except Exception: batt = None break if tire and pressure_kpa is not None: psi = kpa_to_psi(pressure_kpa) bar = round((pressure_kpa / 100.0) / 0.986923, 2) # approximate conversion tempc = int(temp_c) if temp_c is not None else 32 battv = round(batt, 1) if batt is not None else 0 window.tire_info[tire].update({ "BATT": battv, "TEMPc": tempc, "TEMPf": round((tempc * (9 / 5)) + 32, 2), "PSI": psi, "BAR": bar, "KPA": round(pressure_kpa, 1), }) window.activity = 1 window.activity_timer.start(750) window.trigger_repaint() # keep GUI alive even when no messages await asyncio.sleep(0) except asyncio.CancelledError: try: proc.terminate() except Exception: pass raise finally: try: await proc.wait() except Exception: pass #This draws the window class MainWindow(QMainWindow): def __init__(self, loop): super().__init__() self.loop = loop #Stick the Window in the bottom left corner self.setGeometry(0, 800, 350, 400) #window title self.setWindowTitle("Truckputer TPMS") #program icon self.setWindowIcon(QIcon("./TPMSicon.png")) #removes the window's top bar and borders self.setWindowFlags(QtCore.Qt.FramelessWindowHint) #window background set to black self.setStyleSheet("background-color: black") #window transparency self.setWindowOpacity(0.9) self.ble_task = None #activity light toggle on/off self.activity = 0 #activity light timer (ChatGPT help) self.activity_timer = QtCore.QTimer() self.activity_timer.setSingleShot(True) self.activity_timer.timeout.connect(self.reset_activity) #Array to store the tire info and which units we are looking at self.tire_info = { "UNITS":"PSI", "FL":{"BATT":0, "TEMPf":0, "TEMPc":32, "PSI":0, "BAR":0.0, "KPA":0}, "FR":{"BATT":0, "TEMPf":0, "TEMPc":32, "PSI":0, "BAR":0.0, "KPA":0}, "RL":{"BATT":0, "TEMPf":0, "TEMPc":32, "PSI":0, "BAR":0.0, "KPA":0}, "RR":{"BATT":0, "TEMPf":0, "TEMPc":32, "PSI":0, "BAR":0.0, "KPA":0}, "SP":{"BATT":0, "TEMPf":0, "TEMPc":32, "PSI":0, "BAR":0.0, "KPA":0} } #the units button to cycle PSI, BAR, KPA values self.unitsButton = QPushButton("PSI", self) self.unitsButton.setFont(QFont("Noto Sans", 20, QFont.Bold)) self.unitsButton.setStyleSheet("background-color: #303030") self.unitsButton.clicked.connect(self.cycleUnits) self.unitsButton.setGeometry(145, 40, 60, 50) #reset the activity dot def reset_activity(self): self.activity = 0 self.update() #Draws the basic vehicle wheels and axles def drawVehicleFrame(self, qp): frtLeft = QRect(15,15,100,155) frtRight = QRect(235,15,100,155) rearLeft = QRect(15,215,100,155) rearRight = QRect(235,215,100,155) #Store the wheel rectangles so they are easier to use later self.tireRects = { "FL": frtLeft, "FR": frtRight, "RL": rearLeft, "RR": rearRight, } #draw wheels for rect in self.tireRects.values(): qp.drawRect(rect) #draw front "axle" qp.drawLine(102,95,250,95) #draw rear "axle" qp.drawLine(102,295,250,295) #draw "driveline" qp.drawLine(175,95,175,295) #make a little circle for the rear diff #(makes it easy to tell front and rear! ;-) ) qp.setBrush(QBrush(Qt.white, Qt.SolidPattern)) qp.drawEllipse(165,285,20,20) #x,y,width,height #draw the little activity light at top center of window def activityIndicator(self, qp): if self.activity == 1: qp.setBrush(QBrush(Qt.green, Qt.SolidPattern)) else: qp.setBrush(QBrush(Qt.NoBrush)) qp.drawEllipse(170,10,10,10) #draw the sensor info in each "wheel" def drawTireInfo(self, qp, rect, tire, units): upperText = (Qt.AlignTop | Qt.AlignHCenter) centerText = Qt.AlignCenter bottomText = (Qt.AlignBottom | Qt.AlignHCenter) qp.setFont(QFont("Noto Sans", 30, QFont.Bold)) value = tire[units] #Check pressure and change color based on range #PSI if units == "PSI": if 30 <= value <= 40: qp.setPen(QtCore.Qt.GlobalColor.green) elif 20 <= value < 30: qp.setPen(QtCore.Qt.GlobalColor.yellow) else: qp.setPen(QtCore.Qt.GlobalColor.red) #BAR if units == "BAR": if 2.1 <= value <= 2.8: qp.setPen(QtCore.Qt.GlobalColor.green) elif 1.4 <= value < 2.1: qp.setPen(QtCore.Qt.GlobalColor.yellow) else: qp.setPen(QtCore.Qt.GlobalColor.red) #KPA if units == "KPA": qp.setFont(QFont("Noto Sans", 21, QFont.Bold)) if 210 <= value <= 280: qp.setPen(QtCore.Qt.GlobalColor.green) elif 140 <= value < 210: qp.setPen(QtCore.Qt.GlobalColor.yellow) else: qp.setPen(QtCore.Qt.GlobalColor.red) qp.drawText(rect, upperText, str(value)) qp.setFont(QFont("Noto Sans", 11, QFont.Bold)) qp.setPen(QtCore.Qt.GlobalColor.white) qp.drawText(rect, bottomText, f"{tire['TEMPf']}°F\n{tire['TEMPc']}°C\n{tire['BATT']}v\n") #Actually draw the window by calling all the other functions above. def paintEvent(self, event): qp = QPainter(self) qp.setPen(QtCore.Qt.GlobalColor.white) #Draw vehicle and wheels outline self.drawVehicleFrame(qp) #Activity indicator self.activityIndicator(qp) #Fill in Sensor Info units = self.tire_info['UNITS'] for tire, rect in self.tireRects.items(): self.drawTireInfo(qp, rect, self.tire_info[tire], units) #Stop drawing the Vehicle and other stuff. qp.end() #Repaint window? def trigger_repaint(self): self.update() #Start Scanner when window is opened def showEvent(self, event): """Start BLE scanner when window is shown""" if self.ble_task is None or self.ble_task.done(): self.ble_task = self.loop.create_task(ble_device_scanner(self)) #print("scan disabled...") super().showEvent(event) #Stop scanner when window is closed def closeEvent(self, event): #Stop BLE scanner when window is closed if self.ble_task and not self.ble_task.done(): self.ble_task.cancel() #print("scan cancelled...") super().closeEvent(event) #pprint.pprint(self.tire_info) #FOR DEBUGGING #keeps track of which unit of measure we are looking at def cycleUnits(self): current = self.tire_info['UNITS'] order = ["PSI", "BAR", "KPA"] nextUnit = order[(order.index(current) + 1) % len(order)] self.tire_info['UNITS'] = nextUnit self.unitsButton.setText(f"{nextUnit}") self.trigger_repaint() #This ties it all together and runs everything #(and adds the Quit button!) async def main_window(): app = QApplication(sys.argv) loop = qasync.QEventLoop(app) asyncio.set_event_loop(loop) window = MainWindow(loop) #Quit button for when we remove title bar, etc. quitButton = QPushButton("Quit", window) quitButton.setFont(QFont("Noto Sans", 14, QFont.Bold)) quitButton.setStyleSheet("background-color: #303030") quitButton.clicked.connect(window.close) quitButton.setGeometry(145, 345, 60, 50) window.show() window.trigger_repaint() # ensure coroutine is "awaited" before run_forever blocks #no idea what that means, but thanks ChatGPT! await asyncio.sleep(0) with loop: loop.run_forever() if __name__ == "__main__": asyncio.run(main_window())