Skip to content

naneos.ble.partector.readout

The UI curve and the pulse form of a Partector over BLE (firmware 418 or newer).

Both come as a stream of 20 byte packets on the aux characteristic, one packet every 2 s, after "UI?" or "pulse?" (see PartectorBleDiagnosticsPackets). The reader collects them until the last one and assembles the result.

BleDiagnosticsReader

Reads the diagnostics of one device. Must be used on the connection's event loop.

Parameters:

Name Type Description Default
serial_number int

of the device.

 required
channel BleCommandChannel

the commands of the link; a readout holds its lock while it collects.

 required
firmware_version Callable[[], int | None]

the firmware of the device, None while it is not known.

 required
device_type Callable[[], DeviceType | None]

the type of the device, None while it is not known.

 required
Source code in src/naneos/ble/partector/readout.py 
 25
 26
 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
 52
 53
 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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class BleDiagnosticsReader:
    """Reads the diagnostics of one device. Must be used on the connection's event loop.

    Args:
        serial_number: of the device.
        channel: the commands of the link; a readout holds its lock while it collects.
        firmware_version: the firmware of the device, None while it is not known.
        device_type: the type of the device, None while it is not known.
    """

    def __init__(
        self,
        serial_number: int,
        channel: BleCommandChannel,
        firmware_version: Callable[[], int | None],
        device_type: Callable[[], DeviceType | None],
    ) -> None:
        self._serial_number = serial_number
        self._channel = channel
        self._firmware_version = firmware_version
        self._device_type = device_type

        # The data is held back during a UI curve sweep.
        self.hold_points_until = 0.0

        # A readout in flight: the packets it waits for, and the future that
        # gets them once the last packet is in.
        self._kind: str | None = None  # "ui_curve" / "pulse_form"
        self._packets: list[bytes] = []
        self._future: asyncio.Future[list[bytes]] | None = None

    @property
    def holding_points(self) -> bool:
        """True while a sweep disturbs the measurement: its data points are not published."""
        return time.time() < self.hold_points_until

    async def read_ui_curve(self, timeout: float | None = None) -> UiCurve:
        """See PartectorDevice.read_ui_curve()."""
        check_firmware(self._firmware_version(), "UI curve")
        # The sweep ramps the corona voltage: whatever the device measures
        # meanwhile is not air, and it needs an integration time to recover.
        # The integration time is not known over BLE; 16 s is the longest.
        self.hold_points_until = time.time() + UI_COMPUTE_SECONDS + 16 + 2
        try:
            await self._channel.write("UI!")
        except Exception:
            self.hold_points_until = 0.0  # no sweep started: do not hold back good data
            raise
        await asyncio.sleep(UI_COMPUTE_SECONDS)  # without the command lock

        packets = await self._read_packets("UI?", "ui_curve", timeout)
        # 20 packets of 5 points make exactly 100: a different count is a lost or a stray
        # packet, and is left for is_complete to show instead of being cut to size.
        points = sorted(
            point
            for packet in packets
            for point in PartectorBleDiagnosticsPackets.ui_curve_points(packet)
        )
        return UiCurve(
            device_type=self._device_type_or_p2(),
            serial_number=self._serial_number,
            unix_timestamp=int(time.time()),
            voltages=tuple(u for u, _ in points),
            currents=tuple(i for _, i in points),
        )

    async def read_pulse_form(self, timeout: float | None = None) -> PulseForm:
        """See PartectorDevice.read_pulse_form()."""
        check_firmware(self._firmware_version(), "pulse form")
        packets = await self._read_packets("pulse?", "pulse_form", timeout)
        # Every sample goes to its index: a lost packet leaves a gap that is_complete shows,
        # and does not shift the samples behind it.
        samples = {
            index: value
            for packet in packets
            for index, value in PartectorBleDiagnosticsPackets.pulse_form_values(packet)
        }
        return PulseForm(
            device_type=self._device_type_or_p2(),
            serial_number=self._serial_number,
            unix_timestamp=int(time.time()),
            currents=tuple(samples[i] for i in range(PULSE_FORM_VALUES) if i in samples),
        )

    def on_packet(self, data: bytes) -> None:
        """A UI curve or pulse form packet on the aux characteristic (event loop thread)."""
        future = self._future
        expected_ui = self._kind == "ui_curve"
        if future is None or future.done():
            return  # nobody asked: a readout started by a plain write("UI?")
        if PartectorBleDiagnosticsPackets.is_ui_curve(data) != expected_ui:
            return  # the other kind, left over from an earlier readout
        self._packets.append(data)
        if PartectorBleDiagnosticsPackets.is_last(data):
            future.set_result(self._packets)

    def _device_type_or_p2(self) -> DeviceType:
        kind = self._device_type()
        return DeviceType.P2 if kind is None else kind

    async def _read_packets(self, command: str, kind: str, timeout: float | None) -> list[bytes]:
        """Send a command and collect the diagnostics packets that answer it."""
        async with self._channel.lock:
            self._kind = kind
            self._packets = []
            self._future = asyncio.get_running_loop().create_future()
            try:
                await self._channel.write_locked(command)
                return await asyncio.wait_for(
                    asyncio.shield(self._future),
                    timeout or BLE_READOUT_TIMEOUT_SECONDS,
                )
            except TimeoutError:
                raise TimeoutError(
                    f"SN{self._serial_number}: {command!r} answered {len(self._packets)} packets."
                ) from None
            finally:
                self._kind = None
                self._future = None

holding_points property

True while a sweep disturbs the measurement: its data points are not published.

on_packet(data)

A UI curve or pulse form packet on the aux characteristic (event loop thread).

Source code in src/naneos/ble/partector/readout.py 
109
110
111
112
113
114
115
116
117
118
119
def on_packet(self, data: bytes) -> None:
    """A UI curve or pulse form packet on the aux characteristic (event loop thread)."""
    future = self._future
    expected_ui = self._kind == "ui_curve"
    if future is None or future.done():
        return  # nobody asked: a readout started by a plain write("UI?")
    if PartectorBleDiagnosticsPackets.is_ui_curve(data) != expected_ui:
        return  # the other kind, left over from an earlier readout
    self._packets.append(data)
    if PartectorBleDiagnosticsPackets.is_last(data):
        future.set_result(self._packets)

read_pulse_form(timeout=None) async

See PartectorDevice.read_pulse_form().

Source code in src/naneos/ble/partector/readout.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def read_pulse_form(self, timeout: float | None = None) -> PulseForm:
    """See PartectorDevice.read_pulse_form()."""
    check_firmware(self._firmware_version(), "pulse form")
    packets = await self._read_packets("pulse?", "pulse_form", timeout)
    # Every sample goes to its index: a lost packet leaves a gap that is_complete shows,
    # and does not shift the samples behind it.
    samples = {
        index: value
        for packet in packets
        for index, value in PartectorBleDiagnosticsPackets.pulse_form_values(packet)
    }
    return PulseForm(
        device_type=self._device_type_or_p2(),
        serial_number=self._serial_number,
        unix_timestamp=int(time.time()),
        currents=tuple(samples[i] for i in range(PULSE_FORM_VALUES) if i in samples),
    )

read_ui_curve(timeout=None) async

See PartectorDevice.read_ui_curve().

Source code in src/naneos/ble/partector/readout.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def read_ui_curve(self, timeout: float | None = None) -> UiCurve:
    """See PartectorDevice.read_ui_curve()."""
    check_firmware(self._firmware_version(), "UI curve")
    # The sweep ramps the corona voltage: whatever the device measures
    # meanwhile is not air, and it needs an integration time to recover.
    # The integration time is not known over BLE; 16 s is the longest.
    self.hold_points_until = time.time() + UI_COMPUTE_SECONDS + 16 + 2
    try:
        await self._channel.write("UI!")
    except Exception:
        self.hold_points_until = 0.0  # no sweep started: do not hold back good data
        raise
    await asyncio.sleep(UI_COMPUTE_SECONDS)  # without the command lock

    packets = await self._read_packets("UI?", "ui_curve", timeout)
    # 20 packets of 5 points make exactly 100: a different count is a lost or a stray
    # packet, and is left for is_complete to show instead of being cut to size.
    points = sorted(
        point
        for packet in packets
        for point in PartectorBleDiagnosticsPackets.ui_curve_points(packet)
    )
    return UiCurve(
        device_type=self._device_type_or_p2(),
        serial_number=self._serial_number,
        unix_timestamp=int(time.time()),
        voltages=tuple(u for u, _ in points),
        currents=tuple(i for _, i in points),
    )