#!/usr/bin/env python3

"""

OpenP3 - Unofficial Driver and Viewer for Thermal Master P3 (Linux)

Released under MIT License.

"""

import sys

import time

import usb.core

import usb.util

import numpy as np

import cv2

# --- CONFIGURATION ---

SCALE = 3 # UI scaling factor (256*3 = 768px width)

VID = 0x3474 # Vendor ID

PID = 0x45a2 # Product ID

# Calibration constants (Standard InfiRay Tiny1-C approximation)

# Temp (C) = (Raw / 64) - 273.15

RAW_DIVISOR = 64.0

KELVIN_OFFSET = 273.15

class ThermalP3:

"""Handles low-level USB communication with the Thermal Master P3."""

# The magic 8-step handshake sequence derived from reverse engineering

HANDSHAKE_SEQUENCE = [

b'\x01\x01\x81\x00\x01\x00\x00\x00\x00\x00\x00\x00\x1e\x00\x00\x00\x4f\x90',

b'\x01\x01\x81\x00\x02\x00\x00\x00\x00\x00\x00\x00\x0c\x00\x00\x00\x1f\x63',

b'\x01\x01\x81\x00\x06\x00\x00\x00\x00\x00\x00\x00\x40\x00\x00\x00\x65\x4f',

b'\x01\x01\x81\x00\x07\x00\x00\x00\x00\x00\x00\x00\x40\x00\x00\x00\x10\x4c',

b'\x01\x01\x81\x00\x0a\x00\x00\x00\x00\x00\x00\x00\x40\x00\x00\x00\x19\x59',

b'\x10\x21\x81\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x95\xd1',

b'\x01\x01\x81\x00\x0f\x00\x00\x00\x00\x00\x00\x00\x40\x00\x00\x00\xb8\x57',

b'\x01\x2f\x81\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x49\x30'

]

HEARTBEAT_PAYLOAD = b'\x01\x2f\x81\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x49\x30'

def __init__(self):

self.dev = usb.core.find(idVendor=VID, idProduct=PID)

if self.dev is None:

raise ValueError("Thermal Master P3 device not found. Is it plugged in?")

self.last_heartbeat = 0

def connect(self):

"""Initializes the connection, performs handshake, and starts the stream."""

print(f"[+] Device found: {hex(VID)}:{hex(PID)}")

# 1. Detach kernel drivers (if any)

for i in [0, 1]:

if self.dev.is_kernel_driver_active(i):

self.dev.detach_kernel_driver(i)

print(f"[*] Detached kernel driver from interface {i}")

self.dev.set_configuration()

# 2. Perform Handshake on Interface 0

print("[*] Sending handshake sequence...")

for i, payload in enumerate(self.HANDSHAKE_SEQUENCE):

try:

# Request 32, Type 0x41, Value 0, Index 0

self.dev.ctrl_transfer(0x41, 32, 0, 0, payload)

time.sleep(0.01)

except usb.core.USBError as e:

print(f"[!] Warning during handshake step {i}: {e}")

time.sleep(0.1)

# 3. Enable Video Interface (Alt Setting 1)

print("[*] Enabling video interface (Alt Setting 1)...")

try:

self.dev.set_interface_altsetting(interface=1, alternate_setting=1)

except usb.core.USBError as e:

raise RuntimeError(f"Failed to set alt setting: {e}")

# 4. Send Start Command (Request 238)

print("[*] Sending start command (Request 238)...")

try:

# Request 238 (0xEE), Type 0x40, Value 0, Index 1

self.dev.ctrl_transfer(0x40, 238, 0, 1, None)

except usb.core.USBError as e:

print(f"[!] Start command warning: {e}")

print("[+] Camera initialized successfully.")

def get_frame(self):

"""Reads a frame from the device. Returns a numpy array (192x256)."""

# Send heartbeat every 2 seconds to keep connection alive

if time.time() - self.last_heartbeat > 2.0:

try:

self.dev.ctrl_transfer(0x41, 32, 0, 0, self.HEARTBEAT_PAYLOAD)

self.last_heartbeat = time.time()

except:

pass

try:

# Read enough data for one frame (256*192*2 bytes = 98304)

# We read a bit more to be safe regarding headers/padding

data = self.dev.read(0x81, 0x40000, timeout=1000)

raw = np.frombuffer(data, dtype=np.uint16)

if len(raw) >= 256 * 192:

# Extract the last full frame

return raw[-256*192:].reshape(192, 256)

return None

except usb.core.USBError as e:

if e.errno == 110: # Timeout

return None

raise e

# --- UI HELPER FUNCTIONS ---

mouse_x, mouse_y = 0, 0

def mouse_callback(event, x, y, flags, param):

global mouse_x, mouse_y

if event == cv2.EVENT_MOUSEMOVE:

mouse_x, mouse_y = x, y

def raw_to_temp(raw_val):

"""Converts raw 16-bit sensor value to Celsius."""

return (raw_val / RAW_DIVISOR) - KELVIN_OFFSET

def main():

print("--- OpenP3 Viewer ---")

# Initialize Camera

try:

cam = ThermalP3()

cam.connect()

except Exception as e:

print(f"\n❌ Error: {e}")

print("Hint: Did you set up the udev rules? Try running with sudo.")

sys.exit(1)

print("\nControls:")

print(" [q] Quit")

print(" [c] Cycle Color Palette")

# UI Setup

cv2.namedWindow("Thermal P3", cv2.WINDOW_GUI_NORMAL)

cv2.setMouseCallback("Thermal P3", mouse_callback)

colormaps = [cv2.COLORMAP_INFERNO, cv2.COLORMAP_JET, cv2.COLORMAP_HOT, cv2.COLORMAP_BONE]

cmap_idx = 0

fps_start = time.time()

frames = 0

while True:

frame_raw = cam.get_frame()

if frame_raw is not None:

# 1. Processing

# Normalize to 0-255 for display

img_norm = cv2.normalize(frame_raw, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8)

img_color = cv2.applyColorMap(img_norm, colormaps[cmap_idx])

# Upscale image

img_disp = cv2.resize(img_color, (256*SCALE, 192*SCALE), interpolation=cv2.INTER_CUBIC)

# 2. Measurements

min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(frame_raw)

min_temp = raw_to_temp(min_val)

max_temp = raw_to_temp(max_val)

# 3. Mouse Measurement

# Map screen coordinates back to raw sensor coordinates

raw_x = min(max(0, mouse_x // SCALE), 255)

raw_y = min(max(0, mouse_y // SCALE), 191)

hover_temp = raw_to_temp(frame_raw[raw_y, raw_x])

# 4. Drawing Overlay

# Draw Hot/Cold spots

cv2.circle(img_disp, (max_loc[0]*SCALE, max_loc[1]*SCALE), 5, (0,0,255), 2)

cv2.circle(img_disp, (min_loc[0]*SCALE, min_loc[1]*SCALE), 5, (255,0,0), 2)

# Draw Status Bar (Background)

cv2.rectangle(img_disp, (0,0), (img_disp.shape[1], 40), (0,0,0), -1)

# Status Text

status_text = f"Max: {max_temp:.1f}C Min: {min_temp:.1f}C | Cursor: {hover_temp:.1f}C"

cv2.putText(img_disp, status_text, (10, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255,255,255), 1)

# Draw Crosshair at cursor

cv2.line(img_disp, (mouse_x-10, mouse_y), (mouse_x+10, mouse_y), (200,200,200), 1)

cv2.line(img_disp, (mouse_x, mouse_y-10), (mouse_x, mouse_y+10), (200,200,200), 1)

cv2.imshow("Thermal P3", img_disp)

frames += 1

key = cv2.waitKey(1) & 0xFF

if key == ord('q'):

break

elif key == ord('c'):

cmap_idx = (cmap_idx + 1) % len(colormaps)

# Cleanup

fps_end = time.time()

print(f"\nSession ended. Avg FPS: {frames / (fps_end - fps_start):.2f}")

cv2.destroyAllWindows()

if __name__ == "__main__":

main()