"""Parser for Inkbird BLE advertisements. This file is shamelessly copied from the following repository: https://github.com/Ernst79/bleparser/blob/c42ae922e1abed2720c7fac993777e1bd59c0c93/package/bleparser/inkbird.py MIT License applies. """ from __future__ import annotations import asyncio import contextlib import logging import struct from dataclasses import dataclass from enum import Enum, StrEnum, auto from functools import lru_cache from typing import TYPE_CHECKING, Any, ClassVar from uuid import UUID from bleak.exc import BleakCharacteristicNotFoundError, BleakError from bleak_retry_connector import BleakClientWithServiceCache, establish_connection from bluetooth_data_tools import ( monotonic_time_coarse, short_address, ) from bluetooth_sensor_state_data import BluetoothData, SensorUpdate from sensor_state_data import SensorLibrary, Units if TYPE_CHECKING: from collections.abc import Callable, Coroutine from bleak import BleakGATTCharacteristic, BLEDevice from habluetooth import BluetoothServiceInfoBleak _LOGGER = logging.getLogger(__name__) class Model(StrEnum): IBBQ_1 = "iBBQ-1" IBBQ_2 = "iBBQ-2" IBBQ_4 = "iBBQ-4" IBBQ_6 = "iBBQ-6" IBS_TH = "IBS-TH" IBS_TH2 = "IBS-TH2" IBS_P02B = "IBS-P02B" ITH_11_B = "ITH-11-B" ITH_13_B = "ITH-13-B" ITH_21_B = "ITH-21-B" GENERIC_18 = "Generic 18 byte model" IAM_T1 = "IAM-T1" IAM_T2 = "IAM-T2" IHT_2PB = "IHT-2PB" INT_11P_B = "INT-11P-B" class ModelType(Enum): BBQ = auto() SENSOR = auto() @dataclass(frozen=True) class ModelInfo: """Model information.""" name: str model_type: ModelType local_name: str | None message_length: int unpacker: Callable[[bytes], tuple[int, ...]] service_uuid: UUID | None characteristic_uuid: UUID | None notify_uuid: UUID | None use_local_name_for_device: bool parse_adv: bool # Commands written after subscribing to notifications to make a device # start streaming (each entry is ``(characteristic_uuid, payload)``). Most # notify models stream unprompted, so this defaults to empty. notify_init_writes: tuple[tuple[UUID, bytes], ...] = () # Whether the device may be refreshed via a connectable GATT poll. Probe # thermometers like the IBS-P02B broadcast their full reading in the # advertisement and become unstable under active connections — the firmware # stops responding until the batteries are pulled — so polling is disabled # for them. The advertisement already carries every field a poll would read. # See https://github.com/Bluetooth-Devices/inkbird-ble/issues/116 supports_polling: bool = True INKBIRD_SERVICE_UUID = UUID("0000fff0-0000-1000-8000-00805f9b34fb") EIGHTEEN_BYTE_SENSOR_DATA_CHARACTERISTIC_UUID = UUID( "0000fff7-0000-1000-8000-00805f9b34fb" ) NINE_BYTE_SENSOR_DATA_CHARACTERISTIC_UUID = UUID("0000fff2-0000-1000-8000-00805f9b34fb") IAM_T1_CHARACTERISTIC_UUID = UUID("0000fff4-0000-1000-8000-00805f9b34fb") INKBIRD_UNPACK = struct.Struct("= this marks a sub-zero reading IHT_2PB_POS_MAX_HI = 11 # high byte <= this marks a valid positive reading # INT-11P-B GATT support. This connectable BBQ probe carries no readings in its # advertisement; the values are read from the ``fff1`` characteristic on the # ``fff0`` service. The byte layout was reverse-engineered by the community # (https://github.com/Bluetooth-Devices/inkbird-ble/issues/41 and the linked # Home Assistant forum thread) and is not yet verified against hardware here: # [0] header (0xAA) # [1] probe (internal) temperature, °C # [2] flags (bit 7 = probe charging) # [3] ambient temperature, °C (0 means "no ambient reading") # [4] probe battery: low 7 bits = percentage, bit 7 = flag # [5] case battery: bits 1-7 = percentage (>> 1), bit 0 = case charging # [6] unknown INT_11P_B_DATA_CHARACTERISTIC_UUID = UUID("0000fff1-0000-1000-8000-00805f9b34fb") INT_11P_B_MIN_READ_LEN = 6 INT_11P_B_PROBE_TEMP_INDEX = 1 INT_11P_B_AMBIENT_TEMP_INDEX = 3 INT_11P_B_PROBE_BATTERY_INDEX = 4 INT_11P_B_CASE_BATTERY_INDEX = 5 INT_11P_B_BATTERY_MASK = 0x7F MODEL_INFO = { Model.IBBQ_1: ModelInfo( name="iBBQ-1", model_type=ModelType.BBQ, local_name=None, message_length=12, unpacker=struct.Struct(" 0xFFCE) into ~6548C, the # same #155 wraparound family. Signed parsing keeps the 0xFFFF "no # probe" sentinel as -1, which BBQ_PROBE_NOT_CONNECTED still drops. unpacker=struct.Struct(" bytes | None: """Connect to the device and read the data characteristic.""" for attempt in range(2): client = await establish_connection( BleakClientWithServiceCache, ble_device, ble_device.name or ble_device.address, ) try: return await action(client) except BleakCharacteristicNotFoundError: if attempt == 0: await client.clear_cache() continue raise except BleakError: if attempt == 0: continue raise finally: await client.disconnect() msg = "unreachable" # pragma: no cover raise AssertionError(msg) # pragma: no cover @lru_cache def try_parse_model(value: str | Model | None) -> Model | None: """Try to parse the value into a model. Return None if parsing fails. """ with contextlib.suppress(ValueError): return Model(value) # type: ignore[arg-type] return None # A BBQ probe that is not plugged in reports 0xFFFF. All BBQ models now use # signed unpackers, so this surfaces as -1; the unsigned 65535 form is kept # defensively in case a future model is added with an unsigned unpacker. Either # way it means "no probe attached" and must be dropped rather than reported as # a bogus 6553.5°C reading. BBQ_PROBE_NOT_CONNECTED = frozenset((0xFFFF, -1)) # Inkbird hygrometers occasionally emit a corrupt advertisement where the # unsigned humidity field is garbage (e.g. 0xFFFF -> 6553.5%) and the # temperature reads 0. Relative humidity cannot exceed 100%, so a reading # above this is treated as a corrupt packet and dropped rather than polluting # the sensor history. See https://github.com/Bluetooth-Devices/inkbird-ble/issues/141 MAX_PLAUSIBLE_HUMIDITY = 100.0 # Companion plausibility ceiling for ambient/indoor-air decoders (currently # the IAM-T1 notify path). That protocol encodes temperature as an unsigned # 16-bit value plus a separate sign nibble, so a garbage ``0xFFFF`` field with # ``sign == 0`` decodes to 6553.5 °C — the same wraparound shape as the # #155 / #188 / #193 advertisement family, which the signed advertisement # parsers now block at the source. Notify cannot switch to signed parsing # without breaking the sign-nibble protocol, so the guard runs here instead: # any decoded reading whose absolute value exceeds this ceiling marks a # corrupt packet and is dropped. The ceiling matches the invariant the # temperature boundary-net test enforces against the advertisement parsers # (see ``test_adv_temperature_boundary_invariant``). # # NOT valid for BBQ probe decoders (iBBQ-1/2/4/6), which spec up to ~300 °C # and would need their own higher ceiling (e.g. # ``MAX_PLAUSIBLE_PROBE_TEMPERATURE_CELSIUS``) if a real corruption case ever # appears there. Today the ADV BBQ decoders are already signed, so applying # this guard to them would be dead defensive code. MAX_PLAUSIBLE_AMBIENT_TEMPERATURE_CELSIUS = 200.0 def convert_temperature(temp: float) -> float: """Temperature converter. Signed BBQ probes can legitimately read below 0°C (e.g. an ambient probe in a freezer or a cold smoker). Disconnected probes are dropped upstream via ``BBQ_PROBE_NOT_CONNECTED``, so no clamping is needed here — clamping sub-zero readings to 0 would corrupt valid data. """ return temp / 10.0 def is_bbq(lower_name: str) -> bool: """Check if the device is a BBQ sensor.""" return bool("xbbq" in lower_name or "ibbq" in lower_name) class INKBIRDBluetoothDeviceData(BluetoothData): """Date update for INKBIRD Bluetooth devices.""" def __init__( self, device_type: Model | str | None = None, device_data: dict[str, Any] | None = None, update_callback: Callable[[SensorUpdate], None] | None = None, device_data_changed_callback: Callable[[dict[str, Any]], None] | None = None, ) -> None: """Initialize the class.""" super().__init__() self._device_type = try_parse_model(device_type) # Last time we got a full update from ADV data self._last_full_update = 0.0 self._notify_task: asyncio.Task[None] | None = None self._running = True self._device_data = device_data.copy() if device_data else {} self._update_callback = update_callback self._device_data_changed_callback = device_data_changed_callback @property def uses_notify(self) -> bool: """Return True if the device uses notifications.""" return self._device_type in NOTIFY_MODELS async def async_start( self, service_info: BluetoothServiceInfoBleak, ble_device: BLEDevice ) -> None: """Start the device.""" self._set_name_and_manufacturer(service_info) if TYPE_CHECKING: assert self._device_type is not None self._running = True if self._device_type not in NOTIFY_MODELS: return self._notify_task = asyncio.create_task(self._async_start_notify(ble_device)) async def async_stop(self) -> None: """Stop the device.""" self._running = False if self._notify_task: self._notify_task.cancel() with contextlib.suppress(asyncio.CancelledError): await self._notify_task self._notify_task = None async def _async_start_notify(self, ble_device: BLEDevice) -> None: """Start the notification loop.""" while self._running: _LOGGER.debug("Starting notification for %s", self.name) try: await async_connect_action(ble_device, self._async_notify_action) except (BleakError, TimeoutError) as err: _LOGGER.debug("Error starting notification: %s", str(err) or type(err)) _LOGGER.debug("Notification loop for %s finished", self.name) # Wait for 5 seconds before trying again # This is needed to avoid a busy loop if the device is not # available await asyncio.sleep(5) async def _async_notify_action(self, client: BleakClientWithServiceCache) -> None: if TYPE_CHECKING: assert self._device_type is not None dev_info = MODEL_INFO[self._device_type] notify_uuid = dev_info.notify_uuid loop = asyncio.get_running_loop() disconnect_future = loop.create_future() def _resolve_disconnect_callback(_: BleakClientWithServiceCache) -> None: if not disconnect_future.done(): disconnect_future.set_result(None) client.set_disconnected_callback(_resolve_disconnect_callback) await client.start_notify(notify_uuid, self._notify_callback) for char_uuid, payload in dev_info.notify_init_writes: # Some devices (e.g. IHT-2PB) only start streaming after an # activation command is written. These writes are best-effort: # at least one known device rejects the write yet still begins # notifying, so a write error must not abort the session. with contextlib.suppress(BleakError): await client.write_gatt_char(char_uuid, payload, response=False) await disconnect_future # wait for disconnect def _notify_callback( self, sender: BleakGATTCharacteristic, data: bytearray ) -> None: """Dispatch a notification to the handler for the current model.""" _LOGGER.debug("Received notification from %s: %s", sender, data) if not self._running: return handler = self._notify_dispatch.get(self._device_type) if handler is not None: handler(self, sender, data) def _notify_iam_t1(self, sender: BleakGATTCharacteristic, data: bytearray) -> None: """Parse an IAM-T1 notification.""" if ( len(data) == IAM_T1_STATE_NOTIFY_LENGTH and bytes(data[1:3]) == IAM_T1_NOTIFY_STATE_PREFIX ): in_f = data[10] & 0xF unit = Units.TEMP_FAHRENHEIT if in_f else Units.TEMP_CELSIUS _LOGGER.debug("IAM-T1 unit: %s (%s)", unit, self._device_data) if unit != self._device_data.get("temp_unit"): self._device_data["temp_unit"] = unit if TYPE_CHECKING: assert self._device_data_changed_callback is not None _LOGGER.debug("IAM-T1 unit changed: %s (%s)", unit, self._device_data) self._device_data_changed_callback(self._device_data) elif ( len(data) == IAM_T1_DATA_NOTIFY_LENGTH and bytes(data[1:3]) == IAM_T1_NOTIFY_DATA_PREFIX ): sign = data[4] & 0xF temp = data[5] << 8 | data[6] signed_temp = (temp if sign == 0 else -temp) / 10 humidity = (data[7] << 8 | data[8]) / 10 if not self._is_humidity_plausible(humidity): # A garbage humidity field marks a corrupt notification; drop # the whole packet rather than publish any of its fields (#141). return _LOGGER.debug("IAM-T1 temperature: %s (%s)", signed_temp, self._device_data) if self._device_data.get("temp_unit") == Units.TEMP_FAHRENHEIT: # Convert to Celsius signed_temp = round((signed_temp - 32) * 5 / 9, 2) if not self._is_temperature_plausible(signed_temp): # Temperature here is unsigned 16-bit + a separate sign # nibble, so a garbage ``0xFFFF`` field decodes to ~6553 °C # (or ~-6553 with the sign bit set) — the same wraparound # shape the signed advertisement parsers now block at the # source. Treat it as a corrupt notification and drop the # whole packet rather than publish any of its fields. return self.update_predefined_sensor( SensorLibrary.TEMPERATURE__CELSIUS, signed_temp ) self.update_predefined_sensor(SensorLibrary.HUMIDITY__PERCENTAGE, humidity) self.update_predefined_sensor( SensorLibrary.CO2__CONCENTRATION_PARTS_PER_MILLION, data[9] << 8 | data[10], ) self.update_predefined_sensor( SensorLibrary.PRESSURE__HPA, data[11] << 8 | data[12] ) if TYPE_CHECKING: assert self._update_callback is not None self._update_callback(self._finish_update()) else: _LOGGER.debug( "Unexpected notification from %s length: %d header: %s", sender, len(data), bytes(data[:3]), ) def _notify_iht_2pb( self, _sender: BleakGATTCharacteristic, data: bytearray ) -> None: """Parse an IHT-2PB notification. Each packet reports one probe: ``data[2]`` selects the probe (2/4/6 -> probe 1/2/3) and ``data[4:6]`` is the temperature in a base-255 encoding (high byte <= 11 positive, >= 254 sub-zero; the 12-253 range means the probe is unplugged and is skipped). Decoding follows the community reference (ebw44/ESPHome-Inkbird-ITH-2PB). """ if len(data) < IHT_2PB_MIN_NOTIFY_LEN: return probe_num = IHT_2PB_PROBE_SELECTORS.get(data[2]) if probe_num is None: return hi = data[4] lo = data[5] if hi >= IHT_2PB_NEG_HI: temp = ( IHT_2PB_TEMP_BASE * (hi - IHT_2PB_TEMP_BASE) + (lo - IHT_2PB_TEMP_BASE) ) / 10 elif hi <= IHT_2PB_POS_MAX_HI: temp = (IHT_2PB_TEMP_BASE * hi + lo) / 10 else: # Probe not plugged in; skip rather than reporting a bogus value. return _LOGGER.debug("IHT-2PB probe %d temperature: %s", probe_num, temp) self.update_predefined_sensor( SensorLibrary.TEMPERATURE__CELSIUS, temp, key=f"temperature_probe_{probe_num}", name=f"Temperature Probe {probe_num}", ) if TYPE_CHECKING: assert self._update_callback is not None self._update_callback(self._finish_update()) _notify_dispatch: ClassVar[ dict[ Model | None, Callable[ [INKBIRDBluetoothDeviceData, BleakGATTCharacteristic, bytearray], None, ], ] ] @property def device_type(self) -> Model | None: """Return the device type.""" return self._device_type @property def name(self) -> str: """Return the device name.""" if (info := self._get_device_info(None)) and info.name: return info.name return self._device_type.name if self._device_type else "Unknown" def _set_name_and_manufacturer( self, service_info: BluetoothServiceInfoBleak ) -> None: if self._device_type is None: return self.set_device_manufacturer("INKBIRD") local_name = service_info.name address = service_info.address dev_info = MODEL_INFO[self._device_type] dev_type_name = dev_info.name if dev_info.use_local_name_for_device: self.set_device_name(f"{local_name} {short_address(address)}") self.set_device_type(f"{local_name[0]}{dev_type_name[1:]}") else: self.set_device_name(f"{dev_type_name} {short_address(address)}") self.set_device_type(dev_type_name) def _detect_device_type( self, service_info: BluetoothServiceInfoBleak, manufacturer_data: dict[int, bytes], data: bytes, msg_length: int, ) -> bool: """Identify the device type from advertisement data. Set ``self._device_type`` and return ``True`` when a known model is recognised, or return ``False`` when the advertisement does not match any supported device. The branchy match chain keeps this above the mccabe threshold; it is a single linear dispatch by design. """ lower_name = service_info.name.lower() if (lower_name in INKBIRD_NAMES) and ( msg_length in SENSOR_MSG_LENGTHS or "0000fff0-0000-1000-8000-00805f9b34fb" in service_info.service_uuids ): self._device_type = INKBIRD_NAMES[lower_name] elif lower_name.startswith("ink@iht-2pb"): # The IHT-2PB advertises as "Ink@IHT-2PB#" and carries # no usable payload; identify it by name prefix and let the # notify flow (async_start) read its probes over GATT. self._device_type = Model.IHT_2PB elif is_bbq(lower_name) and msg_length in BBQ_LENGTH_TO_TYPE: self._device_type = BBQ_LENGTH_TO_TYPE[msg_length] elif ( msg_length == EIGHTEEN_BYTE_MESSAGE_LENGTH and GENERIC_18_MANUFACTURER_ID in manufacturer_data and "0000fff0-0000-1000-8000-00805f9b34fb" in service_info.service_uuids and manufacturer_data[GENERIC_18_MANUFACTURER_ID].endswith(b"\x00\x00\x00") ): self._device_type = Model.GENERIC_18 elif IAM_T1_MANUFACTURER_ID in manufacturer_data and manufacturer_data[ IAM_T1_MANUFACTURER_ID ].startswith(b"AC-6200"): # AC-6200 self._device_type = Model.IAM_T1 elif ( msg_length == SEVENTEEN_BYTE_MESSAGE_LENGTH and IAM_T2_MANUFACTURER_ID in manufacturer_data and data[2:4] == IAM_T2_MAC_PREFIX # MAC starts with 00:62 ): # IAM-T2 self._device_type = Model.IAM_T2 else: return False return True def _start_update(self, service_info: BluetoothServiceInfoBleak) -> None: """Update from BLE advertisement data.""" _LOGGER.debug("Parsing inkbird BLE advertisement data: %s", service_info) if not (manufacturer_data := service_info.manufacturer_data): self._set_name_and_manufacturer(service_info) return last_id = list(manufacturer_data)[-1] data = int(last_id).to_bytes(2, byteorder="little") + manufacturer_data[last_id] msg_length = len(data) # If we do not know the device type yet, try to determine it from the # advertisement data. if self._device_type in ( None, Model.GENERIC_18, ) and not self._detect_device_type( service_info, manufacturer_data, data, msg_length ): return self._set_name_and_manufacturer(service_info) if TYPE_CHECKING: assert self._device_type is not None if not MODEL_INFO[self._device_type].parse_adv: # Device does not support parsing advertisement data return excludes = MANUFACTURER_DATA_ID_EXCLUDES if len(manufacturer_data) > 1 else None changed_manufacturer_data = self.changed_manufacturer_data( service_info, excludes ) if not changed_manufacturer_data: return if service_info.raw is None and len(changed_manufacturer_data) > 1: # Without raw advertisement bytes, multiple changed entries are # ambiguous (missed packets / new source) so we wait for the # next update. When raw is available, changed_manufacturer_data # reflects only the current packet, so trust the last entry. return last_id = list(changed_manufacturer_data)[-1] data = ( int(last_id).to_bytes(2, byteorder="little") + changed_manufacturer_data[last_id] ) _LOGGER.debug("Parsing INKBIRD BLE advertisement data: %s", data) self._device_type_dispatch[self._device_type](self, data, msg_length) self._last_full_update = service_info.time def poll_needed( self, service_info: BluetoothServiceInfoBleak, last_poll: float | None ) -> bool: """Return whether the device needs a connectable poll for this update. Called every time we get a ``service_info`` for a device, or manually. For models that broadcast their readings, the recency check uses ``service_info.time`` rather than ``self._last_full_update`` so a healthy device whose readings have not changed (and whose repeat advertisements are therefore deduplicated before the parser runs) does not get marked as needing a connectable poll. For poll-only models (``GATT_POLL_MODELS``) the advertisement carries no readings, so its freshness is irrelevant; the gate is instead the time since the last successful poll (``last_poll`` is the number of seconds since the last poll, or ``None`` if the device has never been polled). """ if not self._supports_polling: poll_needed = False elif self._device_type in GATT_POLL_MODELS: poll_needed = last_poll is None or last_poll > MIN_POLL_INTERVAL else: poll_needed = ( not self._last_full_update or (monotonic_time_coarse() - service_info.time) > MIN_POLL_INTERVAL ) _LOGGER.debug("Poll needed for INKBIRD device %s: %s", self.name, poll_needed) return poll_needed @property def _supports_polling(self) -> bool: """Return True if the device supports polling.""" return self._device_type is not None and ( ( self._device_type in SENSOR_MODELS and MODEL_INFO[self._device_type].supports_polling ) or self._device_type in GATT_POLL_MODELS ) async def _async_connect_and_read(self, ble_device: BLEDevice) -> bytes: """Connect to the device and read the data characteristic.""" _LOGGER.debug("Polling INKBIRD device %s", self.name) # Try to connect to the device and read the data characteristic # up to 2 times. # If the first attempt fails, clear the cache and try again. # This is needed because the cache may contain old data. # If the second attempt fails, raise an error. data = await async_connect_action(ble_device, self._async_poll_action) if TYPE_CHECKING: assert data is not None return data async def _async_poll_action( self, client: BleakClientWithServiceCache ) -> bytes | None: """Poll the device for updates.""" if TYPE_CHECKING: assert self._device_type is not None dev_info = MODEL_INFO[self._device_type] service = client.services.get_service(dev_info.service_uuid) char = service.get_characteristic(dev_info.characteristic_uuid) return await client.read_gatt_char(char) def _poll_read_too_short(self, payload: bytes, minimum: int) -> bool: """Return ``True`` (and log) when a GATT poll read is undersized. A device that returns fewer bytes than a decode path needs would otherwise crash that path while slicing; callers skip the decode so the poll yields no values rather than raising. Mirrors the INT-11P-B guard. """ if len(payload) < minimum: _LOGGER.debug( "%s poll read too short (%d bytes, need %d): %s", self.name, len(payload), minimum, payload, ) return True return False async def async_poll(self, ble_device: BLEDevice) -> SensorUpdate: """Poll the device for updates.""" payload = await self._async_connect_and_read(ble_device) if self._device_type in EIGHTEEN_BYTE_SENSOR_MODELS: if not self._poll_read_too_short(payload, EIGHTEEN_BYTE_POLL_MIN_READ_LEN): self._update_eighteen_byte_model_from_raw(payload[5:9], payload[9]) elif self._device_type in NINE_BYTE_SENSOR_MODELS: # Battery doesn't seem to be available for these models # but it is in the advertisement data if not self._poll_read_too_short(payload, NINE_BYTE_POLL_MIN_READ_LEN): self._update_nine_byte_model_from_raw(payload[0:4], None) elif self._device_type == Model.INT_11P_B: self._update_int_11p_b_from_raw(payload) return self._finish_update() def _update_bbq_model(self, data: bytes, _msg_length: int) -> None: """Update a BBQ sensor model.""" # Some are iBBQ, some are xBBQ if TYPE_CHECKING: assert self._device_type is not None xvalue = data[10:] for idx, temp in enumerate(MODEL_INFO[self._device_type].unpacker(xvalue)): if temp in BBQ_PROBE_NOT_CONNECTED: # Probe not plugged in; skip it instead of reporting 6553.5°C. continue num = idx + 1 self.update_predefined_sensor( SensorLibrary.TEMPERATURE__CELSIUS, convert_temperature(temp), key=f"temperature_probe_{num}", name=f"Temperature Probe {num}", ) def _update_nine_byte_model(self, data: bytes, _msg_length: int) -> None: """Update the sensor values for a 9 byte model.""" self._update_nine_byte_model_from_raw(data[0:4], data[7]) def _update_nine_byte_model_from_raw( self, temp_hum_bytes: bytes, bat: int | None ) -> None: if TYPE_CHECKING: assert self._device_type is not None temp, hum = MODEL_INFO[self._device_type].unpacker(temp_hum_bytes) # Only some models report humidity: IBS-TH always, IBS-TH2 when non-zero. reports_humidity = self._device_type == Model.IBS_TH or ( self._device_type == Model.IBS_TH2 and hum != 0 ) humidity = hum / 100 if reports_humidity and not self._is_humidity_plausible(humidity): # Humidity is parsed unsigned (`` None: """Update the sensor values for a 18 byte model.""" self._update_eighteen_byte_model_from_raw(data[6:10], data[10]) def _is_humidity_plausible(self, humidity: float) -> bool: """Return ``False`` for a physically impossible humidity reading. Relative humidity cannot exceed 100%; a higher value (e.g. a garbage ``0xFFFF`` field -> 6553.5%) marks a corrupt packet. Callers drop the whole reading rather than let it pollute the sensor history. Shared by every humidity-bearing decode path. See #141. """ if humidity > MAX_PLAUSIBLE_HUMIDITY: _LOGGER.debug( "Ignoring corrupt reading from %s: humidity %.1f%% exceeds 100%%", self.name, humidity, ) return False return True def _is_temperature_plausible(self, temperature_c: float) -> bool: """Return ``False`` for an ambient temperature outside plausible Celsius range. Scoped to **ambient/indoor-air** decoders whose protocol cannot be made signed at the source (currently only the IAM-T1 sign-nibble notify packet). A garbage 16-bit field there decodes to ~6553 °C, the same wraparound the signed advertisement parsers now block — this guard catches it on the notify side. Callers drop the whole packet rather than publish a temperature outside ``MAX_PLAUSIBLE_AMBIENT_TEMPERATURE_CELSIUS``. **Not valid for BBQ probe decoders** (iBBQ-1/2/4/6), which spec up to ~300 °C — those would need their own higher ceiling if a real corruption case ever appears there. """ if abs(temperature_c) > MAX_PLAUSIBLE_AMBIENT_TEMPERATURE_CELSIUS: _LOGGER.debug( "Ignoring corrupt reading from %s: temperature %.1f °C " "exceeds plausible range", self.name, temperature_c, ) return False return True def _update_eighteen_byte_model_from_raw( self, temp_hum_bytes: bytes, bat: int ) -> None: """Update the sensor values for a 18 byte model.""" if TYPE_CHECKING: assert self._device_type is not None temp, hum = MODEL_INFO[self._device_type].unpacker(temp_hum_bytes) humidity = hum / 10 if not self._is_humidity_plausible(humidity): return self.update_predefined_sensor(SensorLibrary.TEMPERATURE__CELSIUS, temp / 10) self.update_predefined_sensor(SensorLibrary.BATTERY__PERCENTAGE, bat) if hum != 0: self.update_predefined_sensor(SensorLibrary.HUMIDITY__PERCENTAGE, humidity) def _update_seventeen_byte_model(self, data: bytes, _msg_length: int) -> None: """Update the sensor values for 17-byte sensor models (IAM-T2).""" # Data format is 17 bytes total: a 2-byte manufacturer ID followed by # a 15-byte payload laid out as 6 bytes of MAC, 1 unknown byte, 1 # status byte, then 2 bytes each of temperature, humidity and CO2, and # finally 1 battery byte. # Parse status byte status = data[9] # Parse sensor values (all big-endian). Temperature is signed: sub-zero # readings arrive as two's-complement (e.g. -5.0C -> 0xFFCE). Parsing it # unsigned reports ~6553C for any negative temperature (see #155 family). temperature_raw = int.from_bytes(data[10:12], "big", signed=True) humidity = ((data[12] << 8) | data[13]) / 10.0 co2 = (data[14] << 8) | data[15] # Temperature is in tenths of degrees if status & 0x02: # Fahrenheit mode temperature_f = temperature_raw / 10.0 temperature_c = (temperature_f - 32) * 5 / 9 else: # Celsius mode temperature_c = temperature_raw / 10.0 # Battery is intentionally not reported: its encoding is unconfirmed # (one device reported a raw value of 145 for 75%), so it is omitted # rather than published incorrectly. if not self._is_humidity_plausible(humidity): # A garbage humidity field marks a corrupt advertisement; drop the # whole reading rather than publish any of its fields (#141). return self.update_predefined_sensor(SensorLibrary.TEMPERATURE__CELSIUS, temperature_c) self.update_predefined_sensor(SensorLibrary.HUMIDITY__PERCENTAGE, humidity) self.update_predefined_sensor( SensorLibrary.CO2__CONCENTRATION_PARTS_PER_MILLION, co2 ) def _update_int_11p_b_from_raw(self, payload: bytes) -> None: """Update the sensor values for an INT-11P-B GATT read. The probe exposes a small fixed-layout buffer on its ``fff1`` characteristic instead of broadcasting readings. Decoding follows the community reverse-engineering referenced from issue #41. """ if len(payload) < INT_11P_B_MIN_READ_LEN: _LOGGER.debug( "INT-11P-B read too short (%d bytes): %s", len(payload), payload ) return self.update_predefined_sensor( SensorLibrary.TEMPERATURE__CELSIUS, payload[INT_11P_B_PROBE_TEMP_INDEX], key="temperature_probe", name="Probe Temperature", ) ambient_temp = payload[INT_11P_B_AMBIENT_TEMP_INDEX] if ambient_temp: # An ambient reading of 0 means the probe is not reporting an # ambient value (the community config filters it out), so skip it. self.update_predefined_sensor( SensorLibrary.TEMPERATURE__CELSIUS, ambient_temp, key="temperature_ambient", name="Ambient Temperature", ) self.update_predefined_sensor( SensorLibrary.BATTERY__PERCENTAGE, payload[INT_11P_B_PROBE_BATTERY_INDEX] & INT_11P_B_BATTERY_MASK, key="probe_battery", name="Probe Battery", ) self.update_predefined_sensor( SensorLibrary.BATTERY__PERCENTAGE, payload[INT_11P_B_CASE_BATTERY_INDEX] >> 1, key="case_battery", name="Case Battery", ) _device_type_dispatch: ClassVar[ dict[Model, Callable[[INKBIRDBluetoothDeviceData, bytes, int], None]] ] INKBIRDBluetoothDeviceData._device_type_dispatch = { # noqa: SLF001 **dict.fromkeys( BBQ_MODELS, INKBIRDBluetoothDeviceData._update_bbq_model, # noqa: SLF001 ), **dict.fromkeys( NINE_BYTE_SENSOR_MODELS, INKBIRDBluetoothDeviceData._update_nine_byte_model, # noqa: SLF001 ), **dict.fromkeys( EIGHTEEN_BYTE_SENSOR_MODELS, INKBIRDBluetoothDeviceData._update_eighteen_byte_model, # noqa: SLF001 ), **dict.fromkeys( SEVENTEEN_BYTE_SENSOR_MODELS, INKBIRDBluetoothDeviceData._update_seventeen_byte_model, # noqa: SLF001 ), } INKBIRDBluetoothDeviceData._notify_dispatch = { # noqa: SLF001 Model.IAM_T1: INKBIRDBluetoothDeviceData._notify_iam_t1, # noqa: SLF001 Model.IHT_2PB: INKBIRDBluetoothDeviceData._notify_iht_2pb, # noqa: SLF001 }