Skip to content

naneos.partector_ble.decoders

Decoders for the BLE characteristics and advertisement payloads of a Partector.

Every payload is a fixed 20 byte layout of little-endian unsigned integers. Each decoder is a table of fields (name, byte slice, scaling); the generic decode() loop reads them onto a NaneosDeviceDataPoint. Only the size distribution packs its channels as 20 bit values and needs its own loop.

PartectorBleDecoderAux

Bases: PartectorBleDecoderBlueprint

The aux characteristic, also the scan response half of the advertisement.

Source code in src/naneos/partector_ble/decoders.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class PartectorBleDecoderAux(PartectorBleDecoderBlueprint):
    """The aux characteristic, also the scan response half of the advertisement."""

    FIELDS = (
        Field("corona_voltage", slice(0, 2)),
        Field("diffusion_current", slice(2, 4), factor=0.01),
        Field("deposition_voltage", slice(4, 6)),
        Field("flow_from_dp", slice(6, 8), divisor=1000.0),
        Field("ambient_pressure", slice(8, 10)),
        Field("electrometer_1_amplitude", slice(10, 12)),
        Field("electrometer_2_amplitude", slice(12, 14)),
        Field("electrometer_1_gain", slice(14, 16)),
        Field("electrometer_2_gain", slice(16, 18)),
        Field("diffusion_current_offset", slice(18, 20)),
    )

PartectorBleDecoderAuxError

Bases: PartectorBleDecoderBlueprint

The aux characteristic in error mode, marked by 0xFFFF in the first two bytes.

Source code in src/naneos/partector_ble/decoders.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class PartectorBleDecoderAuxError(PartectorBleDecoderBlueprint):
    """The aux characteristic in error mode, marked by 0xFFFF in the first two bytes."""

    MARKER = b"\xff\xff"

    FIELDS = (
        Field("device_status", slice(2, 6), cast=int),
        Field("diffusion_current_delay_on", slice(6, 7), cast=int),
        Field("diffusion_current_delay_off", slice(7, 8), cast=int),
        Field("diffusion_current_average", slice(8, 10), factor=0.01),
        Field("diffusion_current_stddev", slice(10, 12), factor=0.01),
        Field("diffusion_current_max", slice(12, 14), factor=0.01),
        Field("corona_voltage_onset", slice(14, 16), cast=int),
    )

    @classmethod
    def is_error_frame(cls, data: bytes) -> bool:
        return len(data) >= 2 and bytes(data[:2]) == cls.MARKER

PartectorBleDecoderBlueprint

Shared decode loop. Subclasses define FIELDS or override _decode_fields().

Source code in src/naneos/partector_ble/decoders.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class PartectorBleDecoderBlueprint:
    """Shared decode loop. Subclasses define FIELDS or override _decode_fields()."""

    FIELDS: tuple[Field, ...] = ()
    FIELD_NAMES: frozenset[str] = frozenset()

    def __init_subclass__(cls, **kwargs) -> None:
        super().__init_subclass__(**kwargs)
        # Derive the names from the table unless the subclass lists them itself.
        if cls.FIELDS and "FIELD_NAMES" not in cls.__dict__:
            cls.FIELD_NAMES = frozenset(f.name for f in cls.FIELDS)

    @classmethod
    def decode(
        cls, data: bytes, data_structure: NaneosDeviceDataPoint | None = None
    ) -> NaneosDeviceDataPoint:
        """Decode the payload. If data_structure is given it is filled in place and returned."""
        point = data_structure if data_structure is not None else NaneosDeviceDataPoint()
        for name, value in cls._decode_fields(data).items():
            setattr(point, name, value)
        return point

    @classmethod
    def _decode_fields(cls, data: bytes) -> dict[str, float | int]:
        return {field.name: field.read(data) for field in cls.FIELDS}

decode(data, data_structure=None) classmethod

Decode the payload. If data_structure is given it is filled in place and returned.

Source code in src/naneos/partector_ble/decoders.py 
39
40
41
42
43
44
45
46
47
@classmethod
def decode(
    cls, data: bytes, data_structure: NaneosDeviceDataPoint | None = None
) -> NaneosDeviceDataPoint:
    """Decode the payload. If data_structure is given it is filled in place and returned."""
    point = data_structure if data_structure is not None else NaneosDeviceDataPoint()
    for name, value in cls._decode_fields(data).items():
        setattr(point, name, value)
    return point

PartectorBleDecoderSize

Bases: PartectorBleDecoderBlueprint

The size distribution: 8 channels packed as consecutive 20 bit values.

Source code in src/naneos/partector_ble/decoders.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class PartectorBleDecoderSize(PartectorBleDecoderBlueprint):
    """The size distribution: 8 channels packed as consecutive 20 bit values."""

    CHANNELS = ("10", "16", "26", "43", "70", "114", "185", "300")
    FIELD_NAMES = frozenset(f"particle_number_{ch}nm" for ch in CHANNELS)
    BITS_PER_CHANNEL = 20

    @classmethod
    def _decode_fields(cls, data: bytes) -> dict[str, float | int]:
        packed = int.from_bytes(data[:20], byteorder="little")
        mask = (1 << cls.BITS_PER_CHANNEL) - 1
        return {
            f"particle_number_{channel}nm": float((packed >> (cls.BITS_PER_CHANNEL * i)) & mask)
            for i, channel in enumerate(cls.CHANNELS)
        }

PartectorBleDecoderStd

Bases: PartectorBleDecoderBlueprint

The std characteristic, also the first half of the advertisement.

Source code in src/naneos/partector_ble/decoders.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class PartectorBleDecoderStd(PartectorBleDecoderBlueprint):
    """The std characteristic, also the first half of the advertisement."""

    FIELDS = (
        Field("serial_number", slice(14, 16), cast=int),
        Field("ldsa", slice(0, 3), factor=0.01),
        Field("average_particle_diameter", slice(3, 5)),
        Field("particle_number_concentration", slice(5, 8)),
        Field("temperature", slice(8, 9)),
        Field("relative_humidity", slice(9, 10)),
        Field("battery_voltage", slice(12, 14), factor=0.01),
        Field("particle_mass", slice(16, 19), factor=0.01),
    )
    FIELD_NAMES = frozenset(f.name for f in FIELDS) | {"device_status"}

    OFFSET_DEVICE_STATE_LOW = slice(10, 12)
    INDEX_DEVICE_STATE_HIGH = 19

    @classmethod
    def get_serial_number(cls, data: bytes) -> int:
        return int(int.from_bytes(data[slice(14, 16)], byteorder="little"))

    @classmethod
    def _decode_fields(cls, data: bytes) -> dict[str, float | int]:
        fields = super()._decode_fields(data)
        # 16 status bits, plus 7 more packed into the upper bits of byte 19.
        status = int.from_bytes(data[cls.OFFSET_DEVICE_STATE_LOW], byteorder="little")
        status += ((int(data[cls.INDEX_DEVICE_STATE_HIGH]) >> 1) & 0b01111111) << 16
        fields["device_status"] = status
        return fields