diff --git a/README.md b/README.md index 409cad6..692723b 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,88 @@ https://www.eevblog.com/forum/thermal-imaging/infiray-and-their-p2-pro-discussio Check out Leo's Github here: https://github.com/LeoDJ/P2Pro-Viewer/tree/main +## Universal viewer — `src/thermalcam.py` + +The original program (`src/tc001v4.2.py`, below) is hard-coded for the **Topdon +TC001** frame format. `src/thermalcam.py` is a camera-agnostic rewrite that also +works with newer Topdon / InfiRay models such as the **TC002C Duo**, which expose +a different USB frame geometry. + +What it adds: + +- **Auto-detects** the thermal camera on the USB bus — no `--device` needed. +- **Auto-detects the frame geometry** instead of assuming `256x384`, so it + handles the TC001 (`256x384`) and the TC002C Duo family (`256x392`, …). +- **Real error handling** (the original has none). +- **Real temperatures.** It auto-detects the camera's radiometric mode and + decodes true °C — the TC001 at `256x384` and the **TC002C Duo at `512x484`** + (which also gives a crisp 512×384 image). Cameras with no 16-bit mode fall back + to a clearly-labelled *relative* scale. The absolute reading can be offset- + calibrated with `--temp-offset` / the `[` `]` keys. See `docs/TC002C-DUO.md`. +- Same niceties: colormaps, HUD, recording, snapshots, scaling, blur, contrast, + hot/cold spot tracking — plus fixed-pattern-noise removal and a contrast stretch + for cameras that don't pre-AGC their preview. + +### Install + +```bash +sudo apt-get install python3-opencv # Debian/Ubuntu/Raspberry Pi +# or: pip install -r requirements.txt +``` + +### Run + +```bash +python3 src/thermalcam.py # auto-detect everything +python3 src/thermalcam.py --device /dev/video2 +python3 src/thermalcam.py --resolution 256x384 # e.g. force TC001 radiometric mode +python3 src/thermalcam.py --selftest 20 # headless: save a snapshot + diagnostics, no GUI +``` + +Useful options: `--temp-offset N` to correct the absolute °C, `--temp-scale +{auto,64,16}` / `--temp-order {le,be}` to force a radiometric decode, +`--no-stretch` / `--no-destripe` / `--smooth` for the image cleanups. + +### Keys + +`a/z` blur · `s/x` min-max threshold · `d/c` scale · `f/v` contrast · `m` colormap +· `h` HUD · `n` swap bands · `g` temporal smoothing · `e/w` fullscreen on/off · +`r/t` record/stop · `p` snapshot · `[`/`]` temperature-offset calibration · `q`/ESC quit + +### Feed it to other apps (MJPEG bridge) + +`src/thermal_bridge.py` re-serves the parsed, colormapped thermal image as an +MJPEG-over-HTTP stream, so any app can consume it like a webcam — no kernel +module, no root, no dependency: + +```bash +python3 src/thermal_bridge.py --colormap inferno --rotate 90 # serves http://127.0.0.1:8090 +``` + +```python +import cv2 +cap = cv2.VideoCapture("http://127.0.0.1:8090/stream.mjpg") # OpenCV, ffmpeg, browsers, ... +``` + +Options mirror the viewer (`--colormap`, `--rotate`, `--flip`, `--width/--height`, +`--temp-scale`, `--device`, …). Open `http://127.0.0.1:8090/` in a browser for a +live preview. + +### Orientation + +The sensor is landscape (4:3). Rotate/mirror in software with `--rotate +{0,90,180,270}` and `--flip {none,h,v}`, or live with the `o` key. (The InfiRay +sensor also has a hardware mirror/flip, reachable only via the vendor command +protocol — software rotation is simpler.) + +### Camera support + +| Camera | Image | Temperature | +|--------|-------|-------------| +| Topdon TC001 | ✅ | ✅ real °C (`256x384`, /64) | +| Topdon TC002C Duo | ✅ hi-res 512×384 | ✅ real °C (`512x484`, /16; offset-calibratable) | +| Other InfiRay UVC | ✅ likely | auto-detected if a radiometric mode is present | + ## Introduction diff --git a/docs/TC002C-DUO.md b/docs/TC002C-DUO.md new file mode 100644 index 0000000..a9825a0 --- /dev/null +++ b/docs/TC002C-DUO.md @@ -0,0 +1,169 @@ +# Topdon TC002C Duo (and other InfiRay UVC cameras) on Linux + +Notes from getting `src/thermalcam.py` working with a **Topdon TC002C Duo**, and a +reverse-engineering log of the radiometric (true-temperature) protocol pulled +out of the Android app. Written so the work can be continued. + +The TC002C Duo is a dual-sensor (thermal + visible) camera built on the InfiRay +`iruvc` engine — the same family as the TC001 this repo originally targeted, but +it behaves differently over USB. + +USB id: `2bdf:0102` (`Topdon`). It enumerates as a standard UVC camera +(`/dev/videoN`), so `thermalcam.py` auto-detects it. + +## Frame geometry + +`v4l2-ctl -d /dev/video2 --list-formats-ext` reports a partly-garbage table (the +firmware exposes bogus descriptors like `4x12305`, `60x3299`). The real, usable +YUYV modes are: + +| Resolution | What it contains | +|-----------|------------------| +| `256x192` | thermal preview only (8-bit) | +| `256x392` | stacked: two 8-bit thermal planes (**no** 16-bit data) | +| `512x384` | 2× thermal preview, 8-bit only | +| `512x484` | **16-bit temperature band (top) + 512x384 thermal image** ← use this | +| `644x384`, `520x192` | other banded/fused layouts | + +**The `512x484` mode is the one to use.** It is a stack of horizontal bands: + +``` +rows 0 - 95 : 16-bit temperature data (256x192, little-endian, /16 Kelvin), + packed two sensor rows per 512-wide line -> 96*512*2 = 256*192*2 bytes +rows 96 - 97 : telemetry +rows 98 -481 : 512x384 thermal AGC image (8-bit, what we display) +rows 482 -483 : telemetry +``` + +Read naively as one YUYV image the temperature band looks like a green/torn +stripe (its high byte is mis-read as chroma) — which is exactly why it was first +mistaken for garbage. Decoded as 16-bit it is a clean radiometric frame. + +Two gotchas cost real debugging time and are handled in `thermalcam.py`: + +* **OpenCV buffering** hands back stale, mid-frame-desynced buffers. Set + `CAP_PROP_BUFFERSIZE = 1`. (`v4l2-ctl --stream-mmap` never desyncs.) +* **Opening an unsupported resolution desyncs the next stream.** Enumerate the + advertised modes (`VIDIOC_ENUM_FRAMESIZES`) and never probe one the camera + doesn't list. The occasional desynced frame is also detected (its shifted + bytes decode to absurd temperatures) and skipped. + +## Temperature + +Decode the band as little-endian 16-bit and apply the standard InfiRay TPD +formula: + +``` +temp_C = raw / 16 - 273.15 +``` + +This gives a stable, spatially-correct field (the warm/cool structure matches the +scene exactly). The **absolute** value can be offset by several °C because we are +not running the camera's shutter-based NUC or ambient/emissivity correction — use +`--temp-offset` / the `[` `]` keys to dial it in against a known reference (e.g. +your own skin ~34 °C). The original TC001 uses the same scheme at `256x384` but +with a `/64` divisor; `thermalcam.py` knows both via `KNOWN_LAYOUTS`. + +> Earlier notes in this file claimed the Duo exposes *no* 16-bit data over UVC. +> That was wrong — it was based on only testing `256x392`. The `512x484` mode +> carries full radiometric data with no vendor command required. + +## Vendor command protocol (still useful for image quality) + +True temperatures need no command, but the **image** still shows the sensor's +uncorrected non-uniformity (blotchy on flat scenes) because we can't trigger the +camera's shutter NUC over plain UVC. That — and finer temperature calibration — +is what the vendor command protocol below is for. It was reverse-engineered from +the Android app's native libraries before the `512x484` mode was found. + +## The vendor command protocol (reverse-engineered) + +The Android APK's Kotlin/Java only holds the InfiRay SDK *wrapper* +(`com.energy.iruvc.*`); the real protocol is in the native `.so` libraries, +which live in the **`split_config.arm64_v8a.apk`** split (pull from the phone via +`adb`, see below). The relevant libs: + +- `libircmd.so` — the `ir_cmd` command set + temperature math +- `libUSBUVCCamera.so` — a libuvc + libusb fork; `UVCCamera::sendCommand` is the transport +- `libUSBDualCamera.so` — the Duo dual-sensor path +- `libadvirtemp.so`, `libdualcalibration.so` — Y16→°C + Duo calibration + +### Transport: `UVCCamera::sendCommand` + +Commands are **USB vendor control transfers** (not UVC-class), i.e. plain +`libusb_control_transfer`: + +``` +sendCommand(this, bmRequestType, bRequest, wValue, wIndex, data, wLength) + -> libusb_control_transfer(handle, bmRequestType, bRequest, + wValue, wIndex, data, wLength, timeout=1000) +``` + +| Operation | bmRequestType | bRequest | wValue | +|-----------|--------------|----------|--------| +| register WRITE | `0x41` (OUT, vendor, interface) | `0x45` (`'E'`) | `0x0078` | +| register READ | `0xc1` (IN, vendor, interface) | `0x44` (`'D'`) | `0x0078` | + +`wIndex` is a **device register**: + +| Register | Meaning | +|----------|---------| +| `0x9d00` | command descriptor (opcode + arg length) | +| `0x9d08` | command payload (bulk data) | +| `0x1d00`, `0x1d08`, `0x1d40` | command parameters | +| `0x1d04`, `0x1d08` | result read-back | +| `0x200` | status register (poll: bit0 = busy, retry; error if >3) | + +### Worked example: `preview_start` + +Decoded from `Java_com_energy_iruvc_ircmd_LibIRCMD_preview_1start` in `libircmd.so`: + +``` +1. WRITE 0x9d00 <- 0f c1 00 00 00 00 00 08 # opcode 0xc10f, 8-byte arg block +2. WRITE 0x1d08 <- [fps, + (source==1 ? 0x80 : 0x00), # source: 0=SENSOR, 1=FIX_PATTERN + width_hi, width_lo, # big-endian + height_hi, height_lo, # big-endian + mode, # StartPreviewMode (0=VOC_DVP_MODE) + path] # PreviewPathChannel +3. POLL 0x200 (read 1 byte) until (status & 1) == 0; abort if status > 3 +``` + +`thermal_protocol.py` in this directory implements this transport and command. +**It is experimental and unverified on hardware** — see the safety notes there. + +## Optional next steps (nice-to-have, not required) + +Basic temperatures and a clean hi-res image already work from the `512x484` +stream with no commands. The vendor protocol would let us go further: + +* **Shutter NUC** — periodically trigger the sensor's flat-field correction + (`shutter_manual_switch` / `shutter_sta_set`) to remove the slowly-drifting + blotchy non-uniformity that's visible on flat scenes. +* **Calibrated °C** — apply the camera's emissivity / ambient / NUC correction + (`libadvirtemp.so` + calibration assets `nuc_table_high.bin`, `tau*.bin`, + `kt_high.bin`, `bt_high.bin`, `dual_calibration_parameters2.bin`) instead of the + raw linear formula, removing the absolute offset. +* **Visible / fused image** — the `644x384` and other modes appear to carry the + Duo's visible camera for thermal+visible fusion (`libUSBDualCamera.so`). + +To pursue these, either continue the static RE the way `preview_start` was +decoded, or capture the real app's USB traffic (phone with `usbmon`+root, or +Waydroid on Linux with the camera passed through). None of it is needed for the +working radiometric viewer. + +Once the stream delivers genuine 16-bit data, `thermalcam.py` already +auto-detects it (`auto_detect_radiometric`) and shows real °C — or force it with +`--temp-scale`. + +## Pulling the native libs from the phone + +```bash +adb shell pm path com.topdon.topInfrared # find base + split APKs +adb pull <.../split_config.arm64_v8a.apk> /tmp/split.apk +unzip -o /tmp/split.apk 'lib/*' -d /tmp/topdon_libs +ls /tmp/topdon_libs/lib/arm64-v8a/ # *.so +``` + +The libs are ARM64 Android (won't run on an x86_64 laptop) but disassemble fine +(Ghidra/radare2), which is how the protocol above was recovered. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ffe746e --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +# Core viewer (src/thermalcam.py) +opencv-python>=4.5 +numpy>=1.20 + +# Only for the experimental radiometric-unlock work (src/thermal_protocol.py). +# Not needed for the viewer. +# pyusb>=1.2 diff --git a/src/radiometry/__init__.py b/src/radiometry/__init__.py new file mode 100644 index 0000000..4f303cc --- /dev/null +++ b/src/radiometry/__init__.py @@ -0,0 +1,12 @@ +"""Radiometric calibration for InfiRay/Topdon UVC cameras (e.g. TC002C Duo). + +Reproduces the camera's factory true-temperature pipeline on Linux/Pi by +reimplementing the math from the Android native libraries in pure numpy. The +native libs are arm64/Bionic/JNI and cannot be loaded off-device, so the model's +correction lookup tables are extracted once (`tables.py` -> `adv_tables.npz`) and +the per-pixel correction is reimplemented (`corrector.py`). Per-unit factory +parameters (NUC / kt / bt) are read off each camera over the USB vendor protocol +and stored in a per-unit profile (`profile.py`). + +See docs/TC002C-DUO.md for the reverse-engineering background. +""" diff --git a/src/radiometry/adv_tables.npz b/src/radiometry/adv_tables.npz new file mode 100644 index 0000000..111d05f Binary files /dev/null and b/src/radiometry/adv_tables.npz differ diff --git a/src/radiometry/corrector.py b/src/radiometry/corrector.py new file mode 100644 index 0000000..472dc87 --- /dev/null +++ b/src/radiometry/corrector.py @@ -0,0 +1,186 @@ +"""Radiometric correction pipeline: raw Y16 -> true object temperature (°C). + +Reimplements the InfiRay libadvirtemp math in vectorised numpy. The full firmware +pipeline is: + + raw_u16 --[NUC + kt/bt]--> apparent temp --[ems LUT]--> effective emissivity + --[temp_correct: emissivity / reflected-temp / transmission]--> object °C + --[distance/atmosphere]--> reported °C + +The emissivity/reflected-temp/transmission inversion (`emissivity_correct`, the +firmware's `temp_correct`) and its exact inverse (`reverse_temp_correct`) are fully +recovered and validated here by round-trip. The NUC + kt/bt stage that turns raw +counts into the apparent temperature needs the camera's per-unit factory +parameters (read over the vendor protocol, see `vendor.py`/`profile.py`); until a +profile supplies them, `raw_to_apparent` falls back to the legacy linear decode and +the corrector still applies the emissivity correction the legacy path never did. +""" + +import numpy as np + +from . import tables as _tables + +KELVIN = 273.15 +EMS_Q14 = 16384 # emissivity fixed-point scale (2^14), per the firmware +_TAU_FLOOR = 1e-4 # firmware guards tau < 1e-4 / emissivity == 0 + + +def emissivity_correct(t_meas, emissivity, t_refl=20.0, tau=1.0): + """True object temperature from an apparent/measured temperature. + + Inverts the graybody radiance-mixing model (firmware `temp_correct`): the + measured temperature carries the object's own emission plus reflected ambient, + attenuated by atmospheric transmission ``tau``:: + + Tmeas^4 = (1 - tau*e) * Trefl^4 + tau*e * Tobj^4 (T in K, W proportional T^4) + => Tobj = ([Tmeas^4 - (1 - tau*e) * Trefl^4] / (tau*e)) ** 1/4 + + ``emissivity`` is quantised to the camera's Q14 fixed point (e*16384) to match + the firmware. ``t_meas`` may be a scalar or a numpy array; ``t_refl`` is the + reflected apparent temperature (°C, ~ambient). Returns °C. + """ + e_int = np.rint(np.clip(emissivity, _TAU_FLOOR, 1.0) * EMS_Q14) + f = tau * e_int / EMS_Q14 # tau * emissivity + meas4 = (np.asarray(t_meas, dtype=np.float64) + KELVIN) ** 4 + refl4 = (t_refl + KELVIN) ** 4 + obj4 = np.maximum(meas4 - (1.0 - f) * refl4, 0.0) / f + return np.sqrt(np.sqrt(obj4)) - KELVIN + + +def reverse_temp_correct(t_obj, emissivity, t_refl=20.0, tau=1.0): + """Apparent temperature a body of true temperature ``t_obj`` would read. + + The exact inverse of `emissivity_correct` (firmware `reverse_temp_correct`); + kept so the forward correction can be validated by round-trip without the lib. + """ + e_int = np.rint(np.clip(emissivity, _TAU_FLOOR, 1.0) * EMS_Q14) + f = tau * e_int / EMS_Q14 + obj4 = (np.asarray(t_obj, dtype=np.float64) + KELVIN) ** 4 + refl4 = (t_refl + KELVIN) ** 4 + meas4 = (1.0 - f) * refl4 + f * obj4 + return np.sqrt(np.sqrt(meas4)) - KELVIN + + +def compatible_emissivity(target_temp_C, org_ems, tables, version=2): + """Effective in-band emissivity for an object at ``target_temp_C``. + + The user-set broadband emissivity differs from what the sensor sees in its + spectral band, and the gap varies with object temperature; the firmware looks + the corrected value up in a 2-D table over (target temperature, set emissivity) + -- `read_compatible_ems_version1`. We reproduce it as a clamped bilinear + interpolation of the extracted `ems_correct_table_v{version}` (axes in Kelvin / + emissivity). ``target_temp_C`` may be a numpy array (per-pixel); ``org_ems`` is + the scalar scene emissivity. + """ + temp_axis = tables[f"target_temp_list_of_ems_table_v{version}"] # Kelvin + ems_axis = tables[f"org_ems_list_of_ems_table_v{version}"] + table = tables[f"ems_correct_table_v{version}"] # (n_temp, n_ems) + # Interpolate the columns at the scalar set-emissivity -> a curve over temp, + # then interpolate that curve per pixel. np.interp clamps outside the axes, + # matching the firmware's range guard (it errors out of range; we hold the edge). + j = np.interp(np.clip(org_ems, ems_axis[0], ems_axis[-1]), + ems_axis, np.arange(ems_axis.size)) + j0 = int(np.floor(j)); j1 = min(j0 + 1, ems_axis.size - 1); fj = j - j0 + col = table[:, j0] * (1.0 - fj) + table[:, j1] * fj + tk = np.asarray(target_temp_C, dtype=np.float64) + KELVIN + return np.interp(tk, temp_axis, col) + + +def vapor_pressure(rh, temp_C): + """Water-vapour partial pressure (Pa) via the firmware's Magnus form. + + `calculate_vapor_pressure`: 611.2 * exp(17.67 * T / (T + 243.5)) * RH, with the + firmware's [800, 3000] Pa clamp. Feeds the atmospheric-transmission chain; at + indoor range that chain leaves tau ~ 1, so this is provided for completeness and + the pipeline defaults to tau = 1. + """ + es = 611.2 * np.exp(17.67 * temp_C / (temp_C + 243.5)) * rh + return np.clip(es, 800.0, 3000.0) + + +class RadiometricCorrector: + """Apply the radiometric correction to a frame given scene + unit parameters. + + Scene parameters (``emissivity``, ``t_refl``, ``tau``) default to sensible + indoor values; per-unit NUC/kt/bt come from a `profile` once provisioned. + """ + + def __init__(self, emissivity=0.95, t_refl=20.0, tau=1.0, profile=None, + temp_scale=16.0, tables=None, use_ems_lut=False, ems_version=2): + self.emissivity = emissivity + self.t_refl = t_refl + self.tau = tau + self.profile = profile + self.temp_scale = temp_scale + self.use_ems_lut = use_ems_lut + self.ems_version = ems_version + self._tables = tables + if use_ems_lut and self._tables is None: + self._tables = _tables.load() + + def raw_to_apparent(self, raw_u16): + """Raw Y16 counts -> apparent temperature (°C). + + With a provisioned profile this will apply the per-unit NUC + kt/bt kernel + (firmware `temp_measure_with_NUC_value`); without one it is the legacy + linear decode, which is spatially correct but carries the uncalibrated + absolute offset. + """ + if self.profile is not None and self.profile.has_nuc(): + raise NotImplementedError("NUC/kt/bt kernel lands with vendor reads (Phase 3)") + return np.asarray(raw_u16, dtype=np.float64) / self.temp_scale - KELVIN + + def apparent_to_object(self, t_apparent): + """Apply the emissivity / reflected-temp / transmission correction. + + With ``use_ems_lut`` the per-pixel effective emissivity is looked up from + the temperature-dependent table; otherwise the scalar scene emissivity is + used directly (the rigorously round-trip-validated path). + """ + emis = self.emissivity + if self.use_ems_lut: + emis = compatible_emissivity(t_apparent, self.emissivity, + self._tables, self.ems_version) + return emissivity_correct(t_apparent, emis, self.t_refl, self.tau) + + def apply(self, raw_u16): + """Full raw -> corrected object °C for a frame (or scalar).""" + return self.apparent_to_object(self.raw_to_apparent(raw_u16)) + + +def _self_test(): + """Round-trip: emissivity_correct must invert reverse_temp_correct exactly.""" + rng = np.random.default_rng(0) + t_obj = rng.uniform(-20, 120, 100000) + for emis in (0.5, 0.8, 0.95, 0.98, 1.0): + for t_refl in (-10.0, 20.0, 35.0): + for tau in (0.7, 0.9, 1.0): + meas = reverse_temp_correct(t_obj, emis, t_refl, tau) + back = emissivity_correct(meas, emis, t_refl, tau) + err = np.nanmax(np.abs(back - t_obj)) + assert err < 1e-6, f"round-trip {emis=} {t_refl=} {tau=}: {err}" + # Physical sanity: a perfect emitter (e=1) needs no correction. With e<1 the + # camera blends in reflected ambient, so a body HOTTER than its surroundings is + # under-read (correct up) and a body COOLER than its surroundings is over-read + # (correct down). + assert abs(emissivity_correct(30.0, 1.0, 20.0, 1.0) - 30.0) < 1e-9 + assert emissivity_correct(30.0, 0.9, 25.0, 1.0) > 30.0 # warm body, cool room + assert emissivity_correct(20.0, 0.9, 25.0, 1.0) < 20.0 # cool body, warm room + + # Effective-emissivity LUT: in-range, smooth, and a near-identity for skin. + tbl = _tables.load() + for ver in (1, 2): + e = compatible_emissivity(np.array([20.0, 34.0, 60.0]), 0.95, tbl, ver) + assert np.all((0.4 < e) & (e <= 1.0)), f"ems LUT v{ver} out of range: {e}" + # The LUT-enabled corrector must still behave near the scalar path for skin. + c = RadiometricCorrector(emissivity=0.95, t_refl=22.0, use_ems_lut=True) + assert abs(float(c.apparent_to_object(34.0)) - 34.0) < 3.0 + + # Magnus vapour pressure: monotone in RH, within the firmware clamp. + assert vapor_pressure(0.5, 20.0) < vapor_pressure(0.9, 20.0) + assert 800.0 <= vapor_pressure(0.5, 20.0) <= 3000.0 + print("corrector self-test OK: round-trip < 1e-6 °C; ems LUT + Magnus sane") + + +if __name__ == "__main__": + _self_test() diff --git a/src/radiometry/duo.py b/src/radiometry/duo.py new file mode 100644 index 0000000..b4ad69e --- /dev/null +++ b/src/radiometry/duo.py @@ -0,0 +1,85 @@ +"""Topdon TC002C Duo radiometric decode: raw stream frame -> temperature (°C). + +The Duo's true-temperature mode is the UVC resolution the camera advertises as the +bogus-looking ``8x12578`` (= 100624 "pixels"; as YUYV that's 201248 bytes). Each +frame is 100624 little-endian u16: + + [0:2320] header / telemetry (magic 0x70827773, dims, scene config, and + the app's own min/max/avg of the scene as float32 at byte 144) + [2320:51472] temperature plane, 256x192, raw sensor counts + [51472:100624] display image, 256x192 (unpopulated over plain UVC) + +In the indoor operating range the firmware's conversion is linear in the raw counts: + + temp_C = raw / 64 - K + +The gain (1/64 °C per count) is validated **exactly** against the official app: for a +captured frame whose header carries the app's own min/max/avg (22.6 / 34.7 / 27.8 °C), +``raw_min/64 - 50 = 22.62`` and ``raw_max/64 - 50 = 34.69`` -- a 0.1 °C match on all +three. (The camera's full kt/bt + LUT pipeline, reverse-engineered from +libadvirtempac020.so, reduces to this line in-range; the LUT only matters at +temperature extremes outside indoor monitoring.) + +``K`` is a per-frame offset that tracks the sensor's Vtemp / focal-plane temperature +(it drifts as the sensor warms). The firmware reads Vtemp from an internal sensor we +can't see over plain UVC, so `estimate_offset` pins a robust cold-baseline percentile +of the frame to an assumed ambient (`ambient_bg`); this auto-tracks Vtemp drift and is +exact for temperature *differences* (e.g. person-vs-background, the reflection gate), +while the absolute level can be fine-tuned with thermalcam's ``--temp-offset`` / ``[`` +``]`` keys. Emissivity / reflected-temperature correction is applied on top by +`radiometry.corrector.RadiometricCorrector`. +""" + +import numpy as np + +GAIN_DIV = 64.0 # °C per raw count = 1/64 (validated vs the app) +FRAME_U16 = 100624 # 201248 bytes / 2 +FRAME_MAGIC = 0x70827773 # header[0:2] as a little-endian u32 +HEADER_U16 = 2320 +SENSOR_W, SENSOR_H = 256, 192 +_PLANE = SENSOR_W * SENSOR_H # 49152 +TEMP_OFFSET = HEADER_U16 # 2320 +IMAGE_OFFSET = HEADER_U16 + _PLANE # 51472 + +# `estimate_offset` pins this low percentile of the raw plane (the coldest stable +# background, ~ambient) to `AMBIENT_BG` °C. The captured calibration frame's cold +# baseline sat at ~23 °C, which reproduces its app min/max/avg; rooms differ, so this +# is the absolute anchor to nudge via --temp-offset (the gain above is fixed/exact). +BG_PERCENTILE = 2.0 +AMBIENT_BG = 22.0 + + +def is_duo_frame(u16): + """True if a flat u16 buffer looks like a Duo radiometric frame (size + magic).""" + u = np.asarray(u16).reshape(-1) + return u.size >= FRAME_U16 and (int(u[0]) | (int(u[1]) << 16)) == FRAME_MAGIC + + +def split_frame(u16): + """Split a flat u16 frame into (header, temp_raw[192,256], image[192,256] u8).""" + u = np.asarray(u16).reshape(-1)[:FRAME_U16] + header = u[:HEADER_U16] + temp = u[TEMP_OFFSET:TEMP_OFFSET + _PLANE].reshape(SENSOR_H, SENSOR_W) + image = (u[IMAGE_OFFSET:IMAGE_OFFSET + _PLANE] & 0xFF).astype(np.uint8).reshape(SENSOR_H, SENSOR_W) + return header, temp, image + + +def estimate_offset(temp_raw, ambient_bg=AMBIENT_BG): + """Per-frame offset K so the cold background sits at ``ambient_bg`` °C. + + Tracks the sensor's Vtemp drift (the whole plane shifts with focal-plane + temperature) by anchoring a robust low percentile to ambient. + """ + return float(np.percentile(temp_raw, BG_PERCENTILE)) / GAIN_DIV - ambient_bg + + +def apparent_celsius(temp_raw, offset=None, ambient_bg=AMBIENT_BG): + """Raw temperature-plane counts -> temperature (°C): ``raw/64 - K``. + + ``offset`` (K) is auto-estimated from the frame's cold baseline when ``None``. + Returns a float32 array (apparent temperature, before emissivity correction). + """ + raw = np.asarray(temp_raw, dtype=np.float32) + if offset is None: + offset = estimate_offset(raw, ambient_bg) + return raw / GAIN_DIV - np.float32(offset) diff --git a/src/radiometry/profile.py b/src/radiometry/profile.py new file mode 100644 index 0000000..ea39206 --- /dev/null +++ b/src/radiometry/profile.py @@ -0,0 +1,100 @@ +"""Per-unit radiometric calibration profile (keyed by USB serial). + +A profile is the artifact that makes deploying to many homes a non-event: each +camera is provisioned once (its factory NUC / kt / bt read off the device, see +`provision.py`) into a `.json`, and any host that opens that camera loads +the matching profile and gets corrected °C with no manual per-site calibration. + +Scene parameters (emissivity, reflected temperature, transmission) live in the +profile too so a deployment's defaults travel with it. Until a camera is +provisioned the per-unit fields are absent and the corrector falls back to the +legacy linear decode -- still applying the emissivity correction on top. +""" + +import json +import os +from pathlib import Path + +import numpy as np + +from .corrector import RadiometricCorrector + +DEFAULT_DIR = Path(os.environ.get( + "PYTHERMALCAM_PROFILES", + Path.home() / ".config" / "pythermalcam" / "profiles")) + +# Per-unit factory fields, stored as plain lists in JSON, numpy arrays in memory. +_ARRAY_FIELDS = ("nuc", "kt", "bt") + + +class Profile: + """Calibration state for one physical camera. + + Per-unit (from the device): ``nuc`` (NUC table), ``kt``, ``bt``, ``gain_mode``. + Scene defaults: ``emissivity``, ``t_refl`` (°C), ``tau``, ``use_ems_lut``, + ``ems_version``. ``serial`` keys the file; ``temp_scale`` is the raw-units/Kelvin + divisor (16 for the TC002C Duo). + """ + + def __init__(self, serial, model="TC002C-Duo", temp_scale=16.0, + emissivity=0.95, t_refl=20.0, tau=1.0, + use_ems_lut=False, ems_version=2, + nuc=None, kt=None, bt=None, gain_mode=None): + self.serial = serial + self.model = model + self.temp_scale = temp_scale + self.emissivity = emissivity + self.t_refl = t_refl + self.tau = tau + self.use_ems_lut = use_ems_lut + self.ems_version = ems_version + self.nuc = None if nuc is None else np.asarray(nuc, dtype=np.float64) + self.kt = None if kt is None else np.asarray(kt, dtype=np.float64) + self.bt = None if bt is None else np.asarray(bt, dtype=np.float64) + self.gain_mode = gain_mode + + def has_nuc(self): + """True once the per-unit factory parameters have been provisioned.""" + return all(getattr(self, f) is not None for f in _ARRAY_FIELDS) + + def to_dict(self): + d = {"serial": self.serial, "model": self.model, + "temp_scale": self.temp_scale, "emissivity": self.emissivity, + "t_refl": self.t_refl, "tau": self.tau, + "use_ems_lut": self.use_ems_lut, "ems_version": self.ems_version, + "gain_mode": self.gain_mode} + for f in _ARRAY_FIELDS: + a = getattr(self, f) + d[f] = None if a is None else a.tolist() + return d + + @classmethod + def from_dict(cls, d): + return cls(**d) + + @classmethod + def path_for(cls, serial, directory=DEFAULT_DIR): + return Path(directory) / f"{serial}.json" + + @classmethod + def load(cls, serial, directory=DEFAULT_DIR): + return cls.from_dict(json.loads(cls.path_for(serial, directory).read_text())) + + @classmethod + def try_load(cls, serial, directory=DEFAULT_DIR): + """Load the profile for ``serial`` or None if it hasn't been provisioned.""" + p = cls.path_for(serial, directory) + return cls.from_dict(json.loads(p.read_text())) if p.exists() else None + + def save(self, directory=DEFAULT_DIR): + path = self.path_for(self.serial, directory) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(self.to_dict(), indent=2)) + return path + + def corrector(self, tables=None): + """Build a `RadiometricCorrector` configured from this profile.""" + return RadiometricCorrector( + emissivity=self.emissivity, t_refl=self.t_refl, tau=self.tau, + profile=self, temp_scale=self.temp_scale, tables=tables, + use_ems_lut=self.use_ems_lut, ems_version=self.ems_version) diff --git a/src/radiometry/tables.py b/src/radiometry/tables.py new file mode 100644 index 0000000..fb052ad --- /dev/null +++ b/src/radiometry/tables.py @@ -0,0 +1,146 @@ +"""Extract the InfiRay correction lookup tables embedded in libadvirtemp.so. + +The Android `libadvirtemp.so` carries the model-wide emissivity / distance / +target-temperature LUTs as plain `.data` arrays of little-endian float64. They are +constants for the sensor family (not per-unit), so we dump them once into +`adv_tables.npz`, which ships with the package; the runtime never needs the .so. + +Run once to (re)generate the artifact: + + python -m radiometry.tables extract /path/to/libadvirtemp.so + +Everything else loads the shipped npz via `load()`. + +Determined by reverse-engineering (see docs/TC002C-DUO.md): + * arrays are little-endian float64; + * the 2-D ems correction table is row-major [target_temp][org_ems], i.e. its + shape is (len(target_temp_list), len(org_ems_list)) -- confirmed both against + the axis lengths and by which reshape varies smoothly along both axes. +""" + +import struct +import sys +from pathlib import Path + +import numpy as np + +DEFAULT_NPZ = Path(__file__).with_name("adv_tables.npz") + +# 2-D ems-correction tables, each reshaped to (target_temp_axis, org_ems_axis). +EMS_TABLES = { + "ems_correct_table_v1": ("target_temp_list_of_ems_table_v1", + "org_ems_list_of_ems_table_v1"), + "ems_correct_table_v2": ("target_temp_list_of_ems_table_v2", + "org_ems_list_of_ems_table_v2"), +} + +# 1-D axis / lookup tables to carry through verbatim. +LIST_TABLES = [ + "org_ems_list_of_ems_table_v1", "target_temp_list_of_ems_table_v1", + "org_ems_list_of_ems_table_v2", "target_temp_list_of_ems_table_v2", + "dist_table", "new_dist_table", "dist_table_v3", + "temp_table", "target_temp_table", "new_target_temp_table", + "target_temp_table_v3", +] + + +def _read_elf_symbols(path): + """Minimal ELF64 reader: return (file_bytes, {name: (vma, size, file_off)}). + + Maps every defined symbol's virtual address to a file offset through its own + section header, so extraction never depends on a hand-computed delta. + """ + data = Path(path).read_bytes() + if data[:4] != b"\x7fELF" or data[4] != 2: + raise ValueError(f"{path}: not an ELF64 object") + e_shoff, = struct.unpack_from("= len(secs): + continue + sec = secs[st_shndx] + file_off = st_value - sec["addr"] + sec["off"] + syms[cstr(strsec["off"], st_name)] = (st_value, st_size, file_off) + return data, syms + + +def _grab_f64(data, syms, name): + vma, size, off = syms[name] + return np.frombuffer(data[off:off + size], dtype=" caught here. + if not (np.all(np.diff(a) >= 0) and a.max() > a.min()): + raise ValueError(f"{k}: not monotonic non-decreasing -- bad offset/dtype?") + if not (lo <= a.min() and a.max() <= hi): + raise ValueError(f"{k}: out of range [{a.min()},{a.max()}] not in [{lo},{hi}]") + for name, (taxis, eaxis) in EMS_TABLES.items(): + t = tables[name] + if t.shape != (tables[taxis].size, tables[eaxis].size): + raise ValueError(f"{name}: shape {t.shape} != axes " + f"({tables[taxis].size},{tables[eaxis].size})") + if not (0.4 <= t.min() and t.max() <= 1.0): + raise ValueError(f"{name}: values {t.min()}..{t.max()} outside [0.4,1.0]") + + +def extract(so_path, out_path=DEFAULT_NPZ): + """Dump the LUTs from `so_path` (libadvirtemp.so) into `out_path` (.npz).""" + data, syms = _read_elf_symbols(so_path) + tables = {n: _grab_f64(data, syms, n) for n in LIST_TABLES} + for name, (taxis, eaxis) in EMS_TABLES.items(): + flat = _grab_f64(data, syms, name) + tables[name] = flat.reshape(syms[taxis][1] // 8, syms[eaxis][1] // 8) + _validate(tables) + np.savez_compressed(out_path, **tables) + return tables + + +def load(path=DEFAULT_NPZ): + """Load the shipped correction tables as a plain dict of numpy arrays.""" + with np.load(path) as z: + return {k: z[k] for k in z.files} + + +if __name__ == "__main__": + if len(sys.argv) >= 3 and sys.argv[1] == "extract": + out = Path(sys.argv[3]) if len(sys.argv) > 3 else DEFAULT_NPZ + t = extract(sys.argv[2], out) + print(f"Wrote {out} with {len(t)} tables:") + for k, v in sorted(t.items()): + print(f" {k:34} {str(v.shape):12} " + f"[{v.min():.4g}, {v.max():.4g}]") + else: + print(__doc__) + sys.exit("usage: python -m radiometry.tables extract [out.npz]") diff --git a/src/thermal_bridge.py b/src/thermal_bridge.py new file mode 100644 index 0000000..890c2c6 --- /dev/null +++ b/src/thermal_bridge.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +""" +thermal_bridge.py - Serve the thermal camera as an MJPEG-over-HTTP stream. + +Re-exposes an InfiRay/Topdon thermal camera (parsed + colormapped by thermalcam) +as a plain MJPEG stream that ANY app can open like a webcam - no kernel module, +no root, no reboot, no code dependency: + + python3 src/thermal_bridge.py --colormap inferno --rotate 90 + + # then, in any consumer: + OpenCV: cv2.VideoCapture("http://127.0.0.1:8090/stream.mjpg") + ffmpeg: ffmpeg -i http://127.0.0.1:8090/stream.mjpg ... + browser: http://127.0.0.1:8090/ (preview page) + +This exists because the camera's default UVC output is a torn/garbage frame; the +bridge does the 512x484 radiometric parse and hands out a clean thermal image. +""" + +import argparse +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import cv2 + +import thermalcam + +BOUNDARY = "frame" + +# Latest encoded JPEG, published by the capture thread and consumed by clients. +_latest = {"jpeg": None} +_new_frame = threading.Condition() +_stop = threading.Event() + + +def capture_loop(app, out_size, quality): + """Grab frames, colormap them, publish the newest JPEG. Skips desynced/tiled + frames, and reopens the camera if the stream wedges into a stuck desync.""" + encode = [cv2.IMWRITE_JPEG_QUALITY, quality] + skips = 0 + while not _stop.is_set(): + ok, frame = app.cap.read() + raw = app._as_raw(frame) if ok else None + bgr = app.colormap_frame(raw) if raw is not None else None + if bgr is None: # bad frame; keep serving the previous one + skips += 1 + if skips % 60 == 0: # ~2-4 s of nothing usable -> recover + print(f"[bridge] {skips} unusable frames, reopening camera...", flush=True) + app.reopen() + continue + skips = 0 + if out_size is not None: + bgr = cv2.resize(bgr, out_size, interpolation=cv2.INTER_AREA) + ok, buf = cv2.imencode(".jpg", bgr, encode) + if not ok: + continue + with _new_frame: + _latest["jpeg"] = buf.tobytes() + _new_frame.notify_all() + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.0" + + def log_message(self, *a): # quiet + pass + + def do_GET(self): + if self.path.startswith("/stream") or self.path.rstrip("/") == "": + if self.path.rstrip("/") == "": + return self._page() + return self._stream() + if self.path.startswith("/snapshot"): + return self._snapshot() + self.send_error(404) + + def _page(self): + html = (b"" + b"") + self.send_response(200) + self.send_header("Content-Type", "text/html") + self.send_header("Content-Length", str(len(html))) + self.end_headers() + self.wfile.write(html) + + def _snapshot(self): + with _new_frame: + _new_frame.wait(timeout=2.0) + jpeg = _latest["jpeg"] + if jpeg is None: + return self.send_error(503) + self.send_response(200) + self.send_header("Content-Type", "image/jpeg") + self.send_header("Content-Length", str(len(jpeg))) + self.end_headers() + self.wfile.write(jpeg) + + def _stream(self): + self.send_response(200) + self.send_header("Content-Type", f"multipart/x-mixed-replace; boundary={BOUNDARY}") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + try: + while not _stop.is_set(): + with _new_frame: + _new_frame.wait(timeout=2.0) + jpeg = _latest["jpeg"] + if jpeg is None: + continue + self.wfile.write(b"--" + BOUNDARY.encode() + b"\r\n") + self.wfile.write(b"Content-Type: image/jpeg\r\n") + self.wfile.write(f"Content-Length: {len(jpeg)}\r\n\r\n".encode()) + self.wfile.write(jpeg) + self.wfile.write(b"\r\n") + except (BrokenPipeError, ConnectionResetError): + pass + + +def build_app(args): + """Open the camera through thermalcam with the requested settings.""" + names = [n.lower() for _c, n in thermalcam.COLORMAPS] + if args.colormap.lower() not in names: + sys.exit(f"--colormap must be one of: {', '.join(names)}") + + tc_argv = ["--scale", "1", "--rotate", str(args.rotate), "--flip", args.flip, + "--temp-scale", args.temp_scale, "--temp-order", args.temp_order] + if args.device: + tc_argv += ["--device", args.device] + if args.resolution: + tc_argv += ["--resolution", args.resolution] + if args.no_destripe: + tc_argv += ["--no-destripe"] + tc_argv += ["--smooth", str(args.smooth)] + + app = thermalcam.ThermalApp(thermalcam.parse_args(tc_argv)) + app.colormap = names.index(args.colormap.lower()) + app.open_camera() + return app + + +def main(argv=None): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--host", default="127.0.0.1", help="Bind address (0.0.0.0 to expose on the LAN).") + p.add_argument("--port", type=int, default=8090) + p.add_argument("--colormap", default="white_hot", + help="Palette (white_hot, black_hot, inferno, jet, ...). Default white_hot - " + "grayscale gives detectors far fewer false positives on empty scenes.") + p.add_argument("--width", type=int, help="Output width (default: camera native).") + p.add_argument("--height", type=int, help="Output height (default: camera native).") + p.add_argument("--quality", type=int, default=85, help="JPEG quality 1-100.") + p.add_argument("--device", help="Video device. Default: auto-detect.") + p.add_argument("--resolution", help="Force capture WxH.") + p.add_argument("--rotate", type=int, default=0, choices=[0, 90, 180, 270]) + p.add_argument("--flip", default="none", choices=["none", "h", "v"]) + p.add_argument("--temp-scale", default="auto") + p.add_argument("--temp-order", default="le", choices=["le", "be"]) + p.add_argument("--no-destripe", action="store_true") + p.add_argument("--smooth", type=float, default=0.5) + args = p.parse_args(argv) + + app = build_app(args) + out_size = (args.width, args.height) if args.width and args.height else None + + worker = threading.Thread(target=capture_loop, args=(app, out_size, args.quality), daemon=True) + worker.start() + + server = ThreadingHTTPServer((args.host, args.port), Handler) + url = f"http://{args.host}:{args.port}" + print(f"Thermal MJPEG stream at {url}/stream.mjpg (preview: {url}/ snapshot: {url}/snapshot.jpg)", flush=True) + print("Open it from any app, e.g.:", flush=True) + print(f" python3 -c \"import cv2; c=cv2.VideoCapture('{url}/stream.mjpg'); print(c.read()[0])\"", flush=True) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + _stop.set() + server.shutdown() + app.cap.release() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/thermal_protocol.py b/src/thermal_protocol.py new file mode 100644 index 0000000..f9f4f53 --- /dev/null +++ b/src/thermal_protocol.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +""" +thermal_protocol.py - EXPERIMENTAL InfiRay/Topdon vendor command transport. + +Reverse-engineered from the Topdon Android app's native libraries +(libircmd.so / libUSBUVCCamera.so). See docs/TC002C-DUO.md for the full writeup. + +NOTE: real temperatures already work in thermalcam.py with NO commands at all +(via the 512x484 radiometric mode). This module is NOT needed for that. It exists +for the optional next steps - shutter NUC, full °C calibration, the visible/fused +image - which do need the vendor command channel. + +>>> THIS IS UNVERIFIED ON HARDWARE AND NOT WIRED INTO thermalcam.py. <<< +By default it runs DRY (prints the control transfers it WOULD send). Pass --run +to actually talk to the camera, at your own risk. + +Safety: + * Only the safe register reads and the `preview_start` command are implemented. + * The flash/OEM/erase commands from the SDK are deliberately NOT implemented. + * A USB replug resets the camera if it ends up in a bad state. + +Linux caveat: these are USB *vendor* control transfers (bmRequestType 0x41/0xc1), +not UVC-class controls, so the kernel uvcvideo driver does not expose them. To +send them, libusb must talk to the device directly, which conflicts with +streaming the same interface via v4l2 at the same time. Expect to claim/detach +the interface here and do the video capture through libuvc rather than v4l2 once +this path works. That integration is the open task. +""" + +import argparse +import sys +import time + +VID, PID = 0x2BDF, 0x0102 + +# Transport constants (see docs/TC002C-DUO.md). +REQTYPE_WRITE = 0x41 # OUT | vendor | interface +REQTYPE_READ = 0xC1 # IN | vendor | interface +BREQUEST_WRITE = 0x45 # 'E' +BREQUEST_READ = 0x44 # 'D' +WVALUE = 0x0078 +TIMEOUT_MS = 1000 + +# Registers. +REG_CMD = 0x9D00 # command descriptor +REG_PAYLOAD = 0x9D08 # bulk payload +REG_PARAM = 0x1D08 # command parameters +REG_STATUS = 0x0200 # status poll + + +class InfiRayDevice: + def __init__(self, dry_run=True): + self.dry_run = dry_run + self.dev = None + self.usb = None + + def open(self): + import usb.core # pyusb; only needed for a real run + self.usb = usb.core + dev = usb.core.find(idVendor=VID, idProduct=PID) + if dev is None: + raise RuntimeError(f"No {VID:04x}:{PID:04x} device found") + self.dev = dev + if not self.dry_run: + # The kernel uvcvideo driver owns the interface; vendor control + # transfers need it detached. (This is what stops simultaneous v4l2.) + for cfg in dev: + for intf in cfg: + n = intf.bInterfaceNumber + if dev.is_kernel_driver_active(n): + dev.detach_kernel_driver(n) + return self + + def write_reg(self, reg, data): + """Vendor WRITE of `data` bytes to register `reg`.""" + data = bytes(data) + if self.dry_run: + print(f" WRITE reg=0x{reg:04x} <- {data.hex(' ')}") + return len(data) + return self.dev.ctrl_transfer(REQTYPE_WRITE, BREQUEST_WRITE, WVALUE, reg, data, TIMEOUT_MS) + + def read_reg(self, reg, length): + """Vendor READ of `length` bytes from register `reg`.""" + if self.dry_run: + print(f" READ reg=0x{reg:04x} ({length} bytes) [dry-run -> zeros]") + return bytes(length) + return bytes(self.dev.ctrl_transfer(REQTYPE_READ, BREQUEST_READ, WVALUE, reg, length, TIMEOUT_MS)) + + def poll_status(self, tries=1000): + """Poll REG_STATUS until the command completes (bit0 clear). Returns the + last status byte; values > 3 indicate an error per the SDK.""" + for _ in range(tries): + status = self.read_reg(REG_STATUS, 1)[0] + if self.dry_run: + return 0 + if status & 1: # busy + continue + if (status >> 1) & 1 and status > 3: + raise RuntimeError(f"command error, status=0x{status:02x}") + return status + raise TimeoutError("status poll timed out") + + def preview_start(self, width, height, fps=25, mode=0, source=0, path=0): + """The decoded `preview_start` command (see docs/TC002C-DUO.md). + + NOTE: starts the sensor preview; it does not by itself switch the UVC + stream to 16-bit. The data-flow-mode command is still to be recovered. + """ + # 1. command descriptor: opcode 0xc10f, 8-byte arg block. + self.write_reg(REG_CMD, bytes([0x0F, 0xC1, 0, 0, 0, 0, 0, 0x08])) + # 2. parameter block (width/height big-endian). + params = bytes([ + fps & 0xFF, + 0x80 if source == 1 else 0x00, + (width >> 8) & 0xFF, width & 0xFF, + (height >> 8) & 0xFF, height & 0xFF, + mode & 0xFF, + path & 0xFF, + ]) + self.write_reg(REG_PARAM, params) + # 3. wait for completion. + return self.poll_status() + + +def main(argv=None): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--run", action="store_true", + help="Actually send to the camera (default: dry-run, prints only).") + p.add_argument("--width", type=int, default=256) + p.add_argument("--height", type=int, default=384) + args = p.parse_args(argv) + + dev = InfiRayDevice(dry_run=not args.run) + if args.run: + try: + dev.open() + except Exception as exc: # pyusb missing, no perms, device busy, ... + sys.exit(f"open failed: {exc}\n(Try: pip install pyusb; run as root; unplug other users.)") + print(f"preview_start({args.width}x{args.height}){' [DRY-RUN]' if not args.run else ''}:") + dev.preview_start(args.width, args.height) + print("done.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/thermalcam.py b/src/thermalcam.py new file mode 100644 index 0000000..8784eda --- /dev/null +++ b/src/thermalcam.py @@ -0,0 +1,841 @@ +#!/usr/bin/env python3 +""" +thermalcam.py - Universal Linux viewer for InfiRay-based USB thermal cameras. + +Originally written by Les Wright for the Topdon TC001 (see tc001v4.2.py), this +version is a camera-agnostic rewrite that also works with newer Topdon/InfiRay +models such as the TC002C Duo. + +These cameras pack a frame as a stack of horizontal bands: a high-res thermal +*image* band (8-bit AGC, neutral chroma) plus a 16-bit *temperature* band (raw +radiometric data, which shows up with extreme chroma when mis-read as YUYV). The +band geometry differs per model: + + * TC001 256x384 : image 256x192 + temp 256x192 (16-bit, /64 Kelvin) + * TC002C Duo 512x484 : temp 256x192 (16-bit, /16 Kelvin) + image 512x384 + +This viewer auto-detects the camera, picks a mode that carries real temperature +data, decodes it, and shows real degrees C. If no radiometric mode is available +it falls back to a clearly-labelled relative scale. See docs/TC002C-DUO.md. +""" + +import argparse +import fcntl +import glob +import os +import re +import struct +import sys +import time + +try: + import cv2 + import numpy as np +except ImportError as exc: # pragma: no cover - environment guard + sys.exit( + f"Missing dependency: {exc.name}. Install with:\n" + " sudo apt-get install python3-opencv (Debian/Ubuntu/Raspberry Pi)\n" + " pip install opencv-python numpy (everything else)" + ) + +# TC002C Duo true-temperature decode (its own self-calibrating stream + bundled +# calibration LUTs). Optional: the viewer still runs without it on other cameras. +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +try: + from radiometry import duo as _duo +except Exception: # pragma: no cover - radiometry is optional + _duo = None + +# USB vendor IDs known to ship InfiRay-based thermal cameras. 0x2BDF is Topdon. +KNOWN_THERMAL_VIDS = {0x2BDF, 0x0BDA, 0x1514, 0x3474} + +KELVIN = 273.15 # temp_C = raw/scale - 273.15 + +# Exact band geometry per capture resolution, keyed by (width, height): +# temp_rows, image_rows, sensor (w,h), scale (raw units per Kelvin), byte order. +# Fixed geometry is far more robust than per-frame chroma detection, which breaks +# on the occasional desynced/metadata-header frame these cameras emit. +KNOWN_LAYOUTS = { + (256, 384): ((192, 384), (0, 192), (256, 192), 64.0, "le"), # TC001 / P2 family + (512, 484): ((0, 96), (98, 482), (256, 192), 16.0, "le"), # TC002C Duo +} +# The TC002C Duo's true-°C mode: the camera advertises it as the bogus-looking +# "8x12578" (= 100624 u16 = 201248 B), a header+temperature+image frame the firmware +# self-calibrates. Decoded by radiometry/duo.py (real °C, no per-unit calibration). +DUO_MODE = (8, 12578) +# Resolutions to try for genuine temperature data, in order of preference. The Duo's +# self-calibrating mode (real absolute °C) is preferred over the 512x484 linear band. +RADIOMETRIC_MODES = ([DUO_MODE] if _duo is not None else []) + [(256, 384), (512, 484)] +# Plain image modes (no temperature), used only if no radiometric mode works. +IMAGE_MODES = [(512, 384), (256, 392), (256, 192)] + +# (cv2 colormap or None for the grayscale white/black-hot pseudo-palettes), name. +# white-hot is first (the default): on flat/noisy scenes a colour map turns sensor +# noise into person-shaped blobs that fool detectors, while grayscale stays clean. +COLORMAPS = [ + (None, "white_hot"), + (None, "black_hot"), + (cv2.COLORMAP_INFERNO, "inferno"), + (cv2.COLORMAP_JET, "jet"), + (cv2.COLORMAP_HOT, "ironbow"), + (cv2.COLORMAP_MAGMA, "magma"), + (cv2.COLORMAP_VIRIDIS, "viridis"), + (cv2.COLORMAP_BONE, "bone"), +] + + +def apply_palette(luma, idx): + """Colormap an 8-bit luma image; returns (bgr, name). Handles the grayscale + white-hot / black-hot pseudo-palettes (no cv2 colormap).""" + cmap, name = COLORMAPS[idx] + if cmap is None: + if name == "black_hot": + luma = 255 - luma + return cv2.cvtColor(luma, cv2.COLOR_GRAY2BGR), name + return cv2.applyColorMap(luma, cmap), name + + +def is_raspberrypi(): + try: + with open("/sys/firmware/devicetree/base/model", "r") as m: + return "raspberry pi" in m.read().lower() + except OSError: + return False + + +def _sysfs_usb_ids(video_node): + """Return (vid, pid, name) for a /dev/videoN node by walking sysfs, or None.""" + name = os.path.basename(video_node) + base = f"/sys/class/video4linux/{name}" + try: + modalias = open(os.path.join(base, "device", "modalias")).read().strip() + match = re.search(r"v([0-9A-Fa-f]{4})p([0-9A-Fa-f]{4})", modalias) + if not match: + return None + vid, pid = int(match.group(1), 16), int(match.group(2), 16) + except OSError: + return None + try: + product = open(os.path.join(base, "name")).read().strip() + except OSError: + product = "" + return vid, pid, product + + +def find_thermal_device(): + """Locate the thermal camera's /dev/videoN, preferring known thermal VIDs.""" + nodes = sorted(glob.glob("/dev/video*"), + key=lambda p: int(p[10:]) if p[10:].isdigit() else 999) + fallback = None + for node in nodes: + ids = _sysfs_usb_ids(node) + if ids is None: + continue + vid, _pid, product = ids + if vid in KNOWN_THERMAL_VIDS and _can_capture(node): + return node + if fallback is None and "USB Camera" in product and _can_capture(node): + fallback = node + return fallback + + +def supported_resolutions(device): + """Set of (w, h) YUYV modes the device advertises, via VIDIOC_ENUM_FRAMESIZES. + + Used to avoid even *opening* an unsupported resolution: doing so desyncs these + cameras' subsequent streams. Returns None if enumeration fails. + """ + yuyv = ord("Y") | (ord("U") << 8) | (ord("Y") << 16) | (ord("V") << 24) + # _IOWR('V', 74, sizeof(struct v4l2_frmsizeenum)==44) + vidioc_enum_framesizes = (3 << 30) | (44 << 16) | (ord("V") << 8) | 74 + try: + fd = os.open(device, os.O_RDWR) + except OSError: + return None + sizes = set() + try: + for idx in range(128): + buf = struct.pack("III", idx, yuyv, 0) + b"\x00" * 32 + try: + res = fcntl.ioctl(fd, vidioc_enum_framesizes, buf) + except OSError: + break + _i, _fmt, ftype, w, h = struct.unpack("IIIII", res[:20]) + if ftype != 1: # 1 == V4L2_FRMSIZE_TYPE_DISCRETE + break + sizes.add((w, h)) + finally: + os.close(fd) + return sizes or None + + +def _can_capture(node): + cap = cv2.VideoCapture(node, cv2.CAP_V4L2) + try: + if not cap.isOpened(): + return False + cap.set(cv2.CAP_PROP_CONVERT_RGB, 0) + ok, _ = cap.read() + return ok + finally: + cap.release() + + +# -- frame parsing -------------------------------------------------------- +def _largest_run(mask): + """Return (start, end) of the longest contiguous True run in a 1-D mask.""" + best_len = best = 0 + start = None + for i, v in enumerate(mask): + if v and start is None: + start = i + elif not v and start is not None: + if i - start > best_len: + best_len, best = i - start, (start, i) + start = None + if start is not None and len(mask) - start > best_len: + best = (start, len(mask)) + return best or None + + +def classify_bands(raw, swap=False): + """Split a raw YUYV frame into (image_band, temp_band) row ranges. + + The image band reads as a real YUYV picture (chroma near the neutral 128); + the temperature band is 16-bit data that reads as extreme chroma. Telemetry + rows (luma pinned to 0/255) are excluded. Either range may be None. + """ + chroma = raw[..., 1].mean(axis=1) + luma = raw[..., 0].mean(axis=1) + telemetry = (luma < 8) | (luma > 248) + is_image = (np.abs(chroma - 128.0) < 40) & ~telemetry + is_temp = ~is_image & ~telemetry + image_band = _largest_run(is_image) + temp_band = _largest_run(is_temp) + if swap: + image_band, temp_band = temp_band, image_band + return image_band, temp_band + + +class Frame: + """One parsed frame: a display image and (optionally) a temperature map, + both as float32 arrays at the image band's native resolution. + + `valid` is False when a known radiometric layout produced an implausible + temperature field - i.e. the camera emitted a desynced/garbage frame that the + caller should skip rather than display. + """ + + def __init__(self, image, temp, is_real, valid=True): + self.image = image # (H, W) luma + self.temp = temp # (H, W) degrees C, or None + self.is_real = is_real # True when temp is genuine radiometric + self.valid = valid + self.h, self.w = image.shape + + +def _plausible(temp): + """True for a physically sensible temperature field. Rejects desynced frames, + whose shifted image/telemetry bytes decode to a big chunk of absurd values, + while still allowing a small genuinely-hot region (a soldering iron, a flame).""" + median = float(np.median(temp)) + if not -25.0 <= median <= 160.0: + return False + garbage = float(np.mean((temp < -40.0) | (temp > 300.0))) + return garbage < 0.02 + + +def _image_aligned(luma): + """False if the image band looks horizontally tiled - the signature of a + desynced frame whose line stride is wrong (content repeats at width/4 or /2). + + Tested by mean-absolute-difference against a shifted copy: tiled panels are + near-identical (diff ~0), whereas a normal scene - even a smooth left-to-right + temperature gradient - differs substantially when shifted. Flat/near-uniform + scenes are exempt.""" + s = float(luma.std()) + if s < 8.0: # too uniform to judge; treat as fine + return True + w = luma.shape[1] + for period in (w // 4, w // 2): + if period < 2: + continue + diff = float(np.abs(luma[:, period:] - luma[:, :-period]).mean()) + if diff < max(3.0, 0.12 * s): # panels ~identical at this period => tiled + return False + return True + + +def _decode_temp(raw, temp_rows, sensor, scale, order): + """Decode a temperature band into a sensor-resolution °C array, or None.""" + t0, t1 = temp_rows + sw, sh = sensor + band = np.ascontiguousarray(raw[t0:t1]).reshape(-1) + lo, hi = band[0::2].astype(np.uint16), band[1::2].astype(np.uint16) + u16 = (lo + (hi << 8)) if order == "le" else (hi + (lo << 8)) + if u16.size < sw * sh: + return None + temp = u16[:sw * sh].astype(np.float32) / float(scale) - KELVIN + temp = temp.reshape(sh, sw) + return temp if _plausible(temp) else None + + +def _parse_duo(raw): + """Decode a TC002C Duo radiometric frame (the 8x12578 mode) into a Frame. + + The temperature plane carries the thermal scene (real °C from the firmware's + self-calibrating decode); the raw display-image plane is unpopulated over plain + UVC, so the temperature field doubles as the display source. + """ + u16 = np.frombuffer(np.ascontiguousarray(raw).tobytes(), dtype="= _duo.FRAME_U16 * 2 and _duo.is_duo_frame( + np.frombuffer(np.ascontiguousarray(raw).tobytes(), dtype=" skip it. + return Frame(image, None, False, valid=False) + if temp.shape != image.shape: + temp = cv2.resize(temp, (image.shape[1], image.shape[0]), interpolation=cv2.INTER_NEAREST) + return Frame(image, temp, True) + + # Unknown resolution: fall back to chroma-based band detection (best effort). + image_band, temp_band = classify_bands(raw, swap) + if image_band is None: + return Frame(raw[..., 0].astype(np.float32), None, False) + i0, i1 = image_band + image = raw[i0:i1, :, 0].astype(np.float32) + if temp_band is None or not scale: + return Frame(image, None, False) + ih = i1 - i0 + npix = (temp_band[1] - temp_band[0]) * w + sensor = (w, ih) if ih * w <= npix else (w // 2, ih // 2) + temp = _decode_temp(raw, temp_band, sensor, scale, order) + if temp is None: + return Frame(image, None, False) + if temp.shape != image.shape: + temp = cv2.resize(temp, (image.shape[1], image.shape[0]), interpolation=cv2.INTER_NEAREST) + return Frame(image, temp, True) + + +class ThermalApp: + def __init__(self, args): + self.args = args + self.is_pi = is_raspberrypi() + + self.alpha = args.contrast + self.colormap = 0 + self.blur = 0 + self.threshold = 2.0 + self.hud = True + self.recording = False + self.video_out = None + self.rec_start = 0.0 + self.elapsed = "00:00:00" + self.snaptime = "None" + self.fullscreen = False + self.swap = args.swap_halves + self.smooth = args.smooth + self._ema = None + self.rotate = args.rotate # 0/90/180/270 + self.flip = args.flip # none/h/v + + self.temp_scale = None # raw-units-per-Kelvin, or None for relative + self.temp_order = args.temp_order + self.temp_offset = args.temp_offset + self.radiometric = False + self.is_duo = False + + self.cap = None + self.native_w = self.native_h = 0 + self.scale = args.scale # finalised once native size is known + + # -- camera lifecycle ------------------------------------------------- + @staticmethod + def _open_at(device, w, h): + cap = cv2.VideoCapture(device, cv2.CAP_V4L2) + if not cap.isOpened(): + return None, None + cap.set(cv2.CAP_PROP_CONVERT_RGB, 0) + cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"YUYV")) + cap.set(cv2.CAP_PROP_FRAME_WIDTH, w) + cap.set(cv2.CAP_PROP_FRAME_HEIGHT, h) + # Without this OpenCV hands back stale buffered frames that are desynced + # mid-frame (the band layout shifts and the parse fails). + cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) + # Read a few frames: the first ones after a mode change can be warm-up + # garbage that wouldn't classify correctly. + raw = None + for _ in range(5): + ok, frame = cap.read() + raw = ThermalApp._as_raw(frame) if ok else None + if raw is None or raw.shape[1] != w or raw.shape[0] != h: + cap.release() + return None, None + return cap, raw + + def _probe_radiometric(self, cap, raw, scale): + """True if this mode yields genuine temperature data. Checks the probe + frame plus a few live ones, since the occasional frame desyncs.""" + if parse_frame(raw, scale, self.temp_order, self.swap).is_real: + return True + for _ in range(10): + ok, frame = cap.read() + r = self._as_raw(frame) if ok else None + if r is not None and parse_frame(r, scale, self.temp_order, self.swap).is_real: + return True + return False + + def open_camera(self): + device = self.args.device or find_thermal_device() + if device is None: + sys.exit( + "No thermal camera found. Plug it in and check it appears in\n" + " v4l2-ctl --list-devices\n" + "then pass it explicitly, e.g. --device /dev/video2" + ) + if device.isdigit(): + device = "/dev/video" + device + + forced = self.args.temp_scale != "auto" + forced_scale = float(self.args.temp_scale) if forced else None + + # Only probe modes the camera actually advertises - opening an + # unsupported resolution desyncs the stream that follows it. + supported = supported_resolutions(device) + ok = (lambda wh: supported is None or wh in supported) + + pick = None # (cap, frame, scale, order) + if self.args.resolution: + w, h = (int(x) for x in self.args.resolution.lower().split("x")) + cap, raw = self._open_at(device, w, h) + if cap is not None: + pick = (cap, raw, forced_scale, self.temp_order) + else: + # Prefer a mode that yields genuine temperature data. Probe a few + # frames since the occasional one desyncs. + for w, h in RADIOMETRIC_MODES: + if not ok((w, h)): + continue + cap, raw = self._open_at(device, w, h) + if cap is None: + continue + if (w, h) == DUO_MODE: + scale = "duo" # LUT-based decode, real °C + else: + scale = forced_scale if forced else KNOWN_LAYOUTS[(w, h)][3] + if self._probe_radiometric(cap, raw, scale): + pick = (cap, raw, scale, self.temp_order) + break + cap.release() + if pick is None: + for w, h in IMAGE_MODES: + if not ok((w, h)): + continue + cap, raw = self._open_at(device, w, h) + if cap is not None: + pick = (cap, raw, forced_scale, self.temp_order) + break + + if pick is None: + sys.exit(f"{device}: no usable YUYV mode. Try --resolution 256x192.") + + self.cap, raw, self.temp_scale, self.temp_order = pick + self._device = device + self._wh = (raw.shape[1], raw.shape[0]) # for reopen() after a desync + frame = parse_frame(raw, self.temp_scale, self.temp_order, self.swap) + self.is_duo = self.temp_scale == "duo" + self.radiometric = self.temp_scale is not None or frame.is_real + if not self.radiometric: + self.temp_scale = None + self.native_h, self.native_w = frame.h, frame.w + if not self.args.scale: # auto: aim for ~768px wide + self.scale = max(1, min(5, round(768 / self.native_w))) + print(f"Opened {device}: frame {raw.shape[1]}x{raw.shape[0]}, " + f"image {self.native_w}x{self.native_h}, " + f"temperature={'REAL °C' if self.radiometric else 'relative (uncalibrated)'}") + if self.is_duo: + print(f" radiometric: TC002C Duo self-calibrating decode (real °C), " + f"offset={self.temp_offset:+.1f} °C (adjust with [ and ])") + elif self.radiometric: + print(f" radiometric: scale=1/{self.temp_scale:g} K, order={self.temp_order}, " + f"offset={self.temp_offset:+.1f} °C (adjust with [ and ])") + else: + print(" no 16-bit data in this stream - see docs/TC002C-DUO.md") + + def reopen(self): + """Reopen the camera at the same resolution to recover from a stuck + desync (the stream can wedge into a misaligned state).""" + try: + self.cap.release() + except Exception: + pass + cap, _raw = self._open_at(self._device, *self._wh) + if cap is not None: + self.cap = cap + return True + return False + + @staticmethod + def _as_raw(frame): + if frame is None: + return None + if frame.ndim == 3 and frame.shape[2] == 2: + return frame + if frame.ndim == 2 and frame.shape[1] % 2 == 0: + return frame.reshape(frame.shape[0], frame.shape[1] // 2, 2) + return None + + # -- per-frame processing -------------------------------------------- + @staticmethod + def _destripe(luma): + """Remove per-row and per-column fixed-pattern offsets while keeping + smooth gradients (uncorrected previews carry 2D FPN that a stretch + turns into stripes; we can't run the camera's shutter NUC over UVC).""" + row = np.median(luma, axis=1) + luma = luma - (row - cv2.blur(row.reshape(-1, 1), (1, 9)).ravel())[:, None] + col = np.median(luma, axis=0) + return luma - (col - cv2.blur(col.reshape(1, -1), (9, 1)).ravel())[None, :] + + def _display_luma(self, image): + """Turn the raw image band into a stretched, denoised 8-bit luma.""" + luma = image.copy() + if self.smooth > 0: + if self._ema is None or self._ema.shape != luma.shape: + self._ema = luma + else: + self._ema = self.smooth * self._ema + (1.0 - self.smooth) * luma + luma = self._ema + if not self.args.no_destripe: + luma = self._destripe(luma) + if not self.args.no_stretch: + lo, hi = np.percentile(luma, [1, 99]) + span = max(hi - lo, 45.0) + luma = np.clip((luma - lo) * (255.0 / span), 0, 255) + return luma.astype(np.uint8) + + def _orient(self, arr): + """Apply the current rotation/flip to an image or temperature array.""" + if arr is None: + return None + if self.rotate == 90: + arr = cv2.rotate(arr, cv2.ROTATE_90_CLOCKWISE) + elif self.rotate == 180: + arr = cv2.rotate(arr, cv2.ROTATE_180) + elif self.rotate == 270: + arr = cv2.rotate(arr, cv2.ROTATE_90_COUNTERCLOCKWISE) + if self.flip == "h": + arr = cv2.flip(arr, 1) + elif self.flip == "v": + arr = cv2.flip(arr, 0) + return arr + + def render(self, raw): + frame = parse_frame(raw, self.temp_scale, self.temp_order, self.swap) + if not frame.valid: + return None # desynced/garbage frame - caller reuses the last one + is_real = frame.is_real + image = self._orient(frame.image) + if is_real: + temp_map = self._orient(frame.temp) + self.temp_offset + unit = "C" + else: + temp_map = self.args.rel_gain * image + self.args.rel_offset + unit = "lvl" + + h, w = image.shape + center = temp_map[h // 2, w // 2] + inner = temp_map[2:-2, 2:-2] + max_pos = tuple(p + 2 for p in np.unravel_index(np.argmax(inner), inner.shape)) + min_pos = tuple(p + 2 for p in np.unravel_index(np.argmin(inner), inner.shape)) + tmax, tmin = float(temp_map[max_pos]), float(temp_map[min_pos]) + tavg = float(temp_map.mean()) + + luma = self._display_luma(image) + luma = cv2.convertScaleAbs(luma, alpha=self.alpha) + new_w, new_h = w * self.scale, h * self.scale + luma = cv2.resize(luma, (new_w, new_h), interpolation=cv2.INTER_CUBIC) + if self.blur > 0: + luma = cv2.blur(luma, (self.blur, self.blur)) + + heatmap, cmap_name = apply_palette(luma, self.colormap) + + self._draw_crosshair(heatmap, new_w, new_h, center, unit) + self._draw_markers(heatmap, max_pos, min_pos, tmax, tmin, tavg, unit, w, h, new_w, new_h) + if self.hud: + self._draw_hud(heatmap, cmap_name, tavg, unit, is_real) + return heatmap + + def colormap_frame(self, raw): + """A clean colormapped BGR frame (oriented, no crosshair/HUD/markers), + for feeding other apps. Returns None for a desynced frame.""" + frame = parse_frame(raw, self.temp_scale, self.temp_order, self.swap) + if not frame.valid: + return None + image = self._orient(frame.image) + luma = self._display_luma(image) + luma = cv2.convertScaleAbs(luma, alpha=self.alpha) + nw, nh = image.shape[1] * self.scale, image.shape[0] * self.scale + luma = cv2.resize(luma, (nw, nh), interpolation=cv2.INTER_CUBIC) + if self.blur > 0: + luma = cv2.blur(luma, (self.blur, self.blur)) + return apply_palette(luma, self.colormap)[0] + + def _draw_crosshair(self, img, w, h, center_temp, unit): + cx, cy = w // 2, h // 2 + for color, thick in (((255, 255, 255), 2), ((0, 0, 0), 1)): + cv2.line(img, (cx, cy + 20), (cx, cy - 20), color, thick) + cv2.line(img, (cx + 20, cy), (cx - 20, cy), color, thick) + label = f"{center_temp:.1f} {unit}" + cv2.putText(img, label, (cx + 10, cy - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 0, 0), 2, cv2.LINE_AA) + cv2.putText(img, label, (cx + 10, cy - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 255, 255), 1, cv2.LINE_AA) + + def _draw_markers(self, img, max_pos, min_pos, tmax, tmin, tavg, unit, w, h, nw, nh): + sx, sy = nw / w, nh / h + for (row, col), col_bgr, value in ((max_pos, (0, 0, 255), tmax), (min_pos, (255, 0, 0), tmin)): + if abs(value - tavg) < self.threshold: + continue + x, y = int(col * sx), int(row * sy) + cv2.circle(img, (x, y), 5, (0, 0, 0), 2) + cv2.circle(img, (x, y), 5, col_bgr, -1) + text = f"{value:.1f} {unit}" + cv2.putText(img, text, (x + 10, y + 5), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 0, 0), 2, cv2.LINE_AA) + cv2.putText(img, text, (x + 10, y + 5), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 255, 255), 1, cv2.LINE_AA) + + def _draw_hud(self, img, cmap_name, tavg, unit, is_real): + cv2.rectangle(img, (0, 0), (180, 130), (0, 0, 0), -1) + rows = [ + f"Avg: {tavg:.1f} {unit}", + f"Threshold: {self.threshold:.0f}", + f"Colormap: {cmap_name}", + f"Blur: {self.blur} Scale: {self.scale}", + f"Contrast: {self.alpha:.1f}", + f"Snapshot: {self.snaptime}", + ] + for i, text in enumerate(rows): + cv2.putText(img, text, (8, 14 + i * 14), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1, cv2.LINE_AA) + mode = "RADIOMETRIC" if is_real else "RELATIVE (uncal)" + mode_color = (0, 255, 0) if is_real else (0, 165, 255) + cv2.putText(img, mode, (8, 14 + len(rows) * 14), cv2.FONT_HERSHEY_SIMPLEX, 0.4, mode_color, 1, cv2.LINE_AA) + rec_color = (40, 40, 255) if self.recording else (200, 200, 200) + cv2.putText(img, f"REC {self.elapsed}", (8, 14 + (len(rows) + 1) * 14), cv2.FONT_HERSHEY_SIMPLEX, 0.4, rec_color, 1, cv2.LINE_AA) + + # -- recording / snapshots ------------------------------------------- + def start_recording(self, frame_size): + now = time.strftime("%Y%m%d--%H%M%S") + self.video_out = cv2.VideoWriter(now + "-output.avi", cv2.VideoWriter_fourcc(*"XVID"), 25, frame_size) + self.recording = True + self.rec_start = time.time() + + def stop_recording(self): + self.recording = False + self.elapsed = "00:00:00" + if self.video_out is not None: + self.video_out.release() + self.video_out = None + + def snapshot(self, heatmap): + now = time.strftime("%Y%m%d-%H%M%S") + cv2.imwrite(f"thermal-{now}.png", heatmap) + self.snaptime = time.strftime("%H:%M:%S") + print(f"Saved thermal-{now}.png") + + # -- main loop -------------------------------------------------------- + def run(self): + self.open_camera() + win = "Thermal" + cv2.namedWindow(win, cv2.WINDOW_GUI_NORMAL) + cv2.resizeWindow(win, self.native_w * self.scale, self.native_h * self.scale) + self._print_keys() + while True: + ok, frame = self.cap.read() + if not ok or frame is None: + time.sleep(0.05) + continue + raw = self._as_raw(frame) + if raw is None: + continue + heatmap = self.render(raw) + if heatmap is None: # desynced frame; show the previous one + continue + cv2.imshow(win, heatmap) + if self.recording and self.video_out is not None: + self.elapsed = time.strftime("%H:%M:%S", time.gmtime(time.time() - self.rec_start)) + self.video_out.write(heatmap) + if self.handle_key(cv2.waitKey(1) & 0xFF, heatmap): + break + self.cleanup() + + def handle_key(self, key, heatmap): + if key in (ord("q"), 27): + return True + elif key == ord("a"): + self.blur += 1 + elif key == ord("z"): + self.blur = max(0, self.blur - 1) + elif key == ord("s"): + self.threshold += 1 + elif key == ord("x"): + self.threshold = max(0, self.threshold - 1) + elif key == ord("d"): + self.scale = min(5, self.scale + 1) + self._resize_window() + elif key == ord("c"): + self.scale = max(1, self.scale - 1) + self._resize_window() + elif key == ord("f"): + self.alpha = min(3.0, round(self.alpha + 0.1, 1)) + elif key == ord("v"): + self.alpha = max(0.0, round(self.alpha - 0.1, 1)) + elif key == ord("m"): + self.colormap = (self.colormap + 1) % len(COLORMAPS) + elif key == ord("h"): + self.hud = not self.hud + elif key == ord("n"): + self.swap = not self.swap + elif key == ord("g"): + self.smooth = 0.0 if self.smooth > 0 else (self.args.smooth or 0.5) + elif key == ord("o"): # cycle rotation 0->90->180->270 + self.rotate = (self.rotate + 90) % 360 + elif key == ord("w"): + self._set_fullscreen(False) + elif key == ord("e"): + self._set_fullscreen(True) + elif key == ord("p"): + self.snapshot(heatmap) + elif key == ord("r") and not self.recording: + self.start_recording((heatmap.shape[1], heatmap.shape[0])) + elif key == ord("t"): + self.stop_recording() + elif key == ord("]"): # calibration: nudge temperature offset / relative gain + if self.radiometric: + self.temp_offset = round(self.temp_offset + 0.5, 1) + else: + self.args.rel_gain = round(self.args.rel_gain + 0.05, 3) + elif key == ord("["): + if self.radiometric: + self.temp_offset = round(self.temp_offset - 0.5, 1) + else: + self.args.rel_gain = round(self.args.rel_gain - 0.05, 3) + return False + + def _resize_window(self): + if not self.fullscreen and not self.is_pi: + cv2.resizeWindow("Thermal", self.native_w * self.scale, self.native_h * self.scale) + + def _set_fullscreen(self, on): + self.fullscreen = on + cv2.setWindowProperty("Thermal", cv2.WND_PROP_FULLSCREEN, + cv2.WINDOW_FULLSCREEN if on else cv2.WINDOW_NORMAL) + if not on: + self._resize_window() + + def cleanup(self): + if self.recording: + self.stop_recording() + if self.cap is not None: + self.cap.release() + cv2.destroyAllWindows() + + def _print_keys(self): + print( + "\nKey bindings:\n" + " a/z blur +/- s/x min-max label threshold +/-\n" + " d/c scale +/- f/v contrast +/-\n" + " m cycle colormap h toggle HUD\n" + " n swap bands g toggle temporal smoothing\n" + " o rotate 90 deg e/w fullscreen on/off\n" + " r/t record / stop p snapshot\n" + " [/] temperature offset (or relative gain) calibration\n" + " q/ESC quit\n" + ) + + # -- headless self-test ---------------------------------------------- + def selftest(self, frames): + self.open_camera() + last = None + for i in range(frames): + ok, frame = self.cap.read() + if not ok or frame is None: + continue + raw = self._as_raw(frame) + if raw is None: + continue + f = parse_frame(raw, self.temp_scale, self.temp_order, self.swap) + rendered = self.render(raw) + if rendered is not None: + last = rendered + if i == frames - 1 and f.valid: + if f.is_real: + t = f.temp + self.temp_offset + print(f"frame {i}: REAL temps center={t[f.h//2, f.w//2]:.1f}C " + f"min={t.min():.1f} max={t.max():.1f}") + else: + print(f"frame {i}: relative (no radiometric data)") + self.cap.release() + if last is None: + print("SELFTEST FAILED: no frames decoded") + return 1 + cv2.imwrite("selftest-thermal.png", last) + print(f"SELFTEST OK: wrote selftest-thermal.png ({last.shape[1]}x{last.shape[0]})") + return 0 + + +def parse_args(argv=None): + p = argparse.ArgumentParser(description="Universal Linux thermal camera viewer (TC001, TC002C Duo, InfiRay).") + p.add_argument("--device", help="Video device, e.g. /dev/video2 or 2. Default: auto-detect.") + p.add_argument("--resolution", help="Force capture WxH, e.g. 512x484. Default: best radiometric mode.") + p.add_argument("--scale", type=int, default=0, help="Display upscaling 1-5 (0 = auto from image size).") + p.add_argument("--contrast", type=float, default=1.0, help="Initial contrast (0.0-3.0).") + p.add_argument("--temp-scale", default="auto", + help="Radiometric raw-units-per-Kelvin divisor: 'auto', 64 (TC001), or 16 (Duo).") + p.add_argument("--temp-order", default="le", choices=["le", "be"], help="16-bit byte order.") + p.add_argument("--temp-offset", type=float, default=0.0, + help="°C added to decoded temperatures (calibrate against a known reference; live keys [ ]).") + p.add_argument("--rel-gain", type=float, default=0.2, help="Relative-mode level->pseudo-temp gain.") + p.add_argument("--rel-offset", type=float, default=0.0, help="Relative-mode offset.") + p.add_argument("--no-stretch", action="store_true", help="Disable the display contrast stretch.") + p.add_argument("--no-destripe", action="store_true", help="Disable fixed-pattern-noise removal.") + p.add_argument("--smooth", type=float, default=0.5, help="Temporal smoothing 0.0-0.9 (0 disables). Live: 'g'.") + p.add_argument("--swap-halves", action="store_true", help="Swap which band is image vs data. Live: 'n'.") + p.add_argument("--rotate", type=int, default=0, choices=[0, 90, 180, 270], + help="Rotate the image clockwise. Live: cycle with 'o'.") + p.add_argument("--flip", default="none", choices=["none", "h", "v"], help="Mirror the image horizontally/vertically.") + p.add_argument("--selftest", type=int, metavar="N", help="Headless: grab N frames, save a snapshot, exit.") + return p.parse_args(argv) + + +def main(argv=None): + args = parse_args(argv) + app = ThermalApp(args) + if args.selftest: + return app.selftest(args.selftest) + try: + app.run() + except KeyboardInterrupt: + app.cleanup() + print("\nBye.") + return 0 + + +if __name__ == "__main__": + sys.exit(main())