88 lines
1.9 KiB
Python
88 lines
1.9 KiB
Python
from app.qingping.models import QingpingState
|
|
|
|
|
|
def parse_cgdn1(data: bytes, rssi: int | None = None) -> QingpingState | None:
|
|
|
|
# Восемь первых байт — заголовок CGDN1.
|
|
if len(data) < 8:
|
|
return None
|
|
|
|
temperature = None
|
|
humidity = None
|
|
pm25 = None
|
|
pm10 = None
|
|
co2 = None
|
|
|
|
pos = 8
|
|
|
|
while pos + 2 <= len(data):
|
|
field_type = data[pos]
|
|
length = data[pos + 1]
|
|
|
|
pos += 2
|
|
|
|
if pos + length > len(data):
|
|
return None
|
|
|
|
value = data[pos:pos + length]
|
|
pos += length
|
|
|
|
# Temperature + Humidity
|
|
if field_type == 0x01 and length == 4:
|
|
temperature = (
|
|
int.from_bytes(
|
|
value[0:2],
|
|
byteorder="little",
|
|
signed=True,
|
|
)
|
|
/ 10
|
|
)
|
|
|
|
humidity = (
|
|
int.from_bytes(
|
|
value[2:4],
|
|
byteorder="little",
|
|
signed=False,
|
|
)
|
|
/ 10
|
|
)
|
|
|
|
# PM2.5 + PM10
|
|
elif field_type == 0x12 and length == 4:
|
|
pm25 = int.from_bytes(
|
|
value[0:2],
|
|
byteorder="little",
|
|
)
|
|
|
|
pm10 = int.from_bytes(
|
|
value[2:4],
|
|
byteorder="little",
|
|
)
|
|
|
|
# CO2
|
|
elif field_type == 0x13 and length == 2:
|
|
co2 = int.from_bytes(
|
|
value,
|
|
byteorder="little",
|
|
)
|
|
|
|
if any(
|
|
value is None
|
|
for value in (
|
|
temperature,
|
|
humidity,
|
|
pm25,
|
|
pm10,
|
|
co2,
|
|
)
|
|
):
|
|
return None
|
|
|
|
return QingpingState(
|
|
temperature=temperature,
|
|
humidity=humidity,
|
|
co2=co2,
|
|
pm25=pm25,
|
|
pm10=pm10,
|
|
rssi=rssi,
|
|
) |