naneos

naneos-devices: talk to naneos particle solutions devices over serial and BLE.

ConnectionType

Bases: StrEnum

How a data point reached us. Ordered by trust: serial > connected.

Source code in src/naneos/data_point.py

class ConnectionType(StrEnum):
    """How a data point reached us. Ordered by trust: serial > connected."""

    SERIAL = "serial"
    CONNECTED = "connected"

DeviceType

Bases: IntEnum

Numeric device type as used by the upload backend. Do not renumber.

Source code in src/naneos/data_point.py

class DeviceType(IntEnum):
    """Numeric device type as used by the upload backend. Do not renumber."""

    P2 = 0
    P1 = 1
    P2PRO = 2
    # 3 is reserved: it was the P2 Pro CS, which the backend still knows.

    @classmethod
    def from_name(cls, name: str) -> "DeviceType | None":
        """The type a device names itself in answer to "name?" (same over USB and BLE)."""
        return _DEVICE_NAMES.get(name)

from_name(name) classmethod

The type a device names itself in answer to "name?" (same over USB and BLE).

Source code in src/naneos/data_point.py

@classmethod
def from_name(cls, name: str) -> "DeviceType | None":
    """The type a device names itself in answer to "name?" (same over USB and BLE)."""
    return _DEVICE_NAMES.get(name)

NaneosDeviceDataPoint dataclass

One measurement of one device. Every field is optional; None means "not reported".

Source code in src/naneos/data_point.py

@dataclass
class NaneosDeviceDataPoint:
    """One measurement of one device. Every field is optional; None means "not reported"."""

    # mandatory
    unix_timestamp: int | None = None  # ms since epoch
    serial_number: int | None = None
    connection_type: ConnectionType | None = None
    firmware_version: int | None = None
    device_type: DeviceType | None = None  # None until the device family is known
    device_status: int | None = None  # bitmask

    # optional
    runtime_min: float | None = None  # minutes since start
    ldsa: float | None = None  # um**2/cm**3
    particle_number_concentration: float | None = None  # particles/cm**3
    average_particle_diameter: float | None = None  # nm
    particle_mass: float | None = None  # ug/m**3
    particle_surface: float | None = None  # um**2/m**3
    diffusion_current: float | None = None  # nA
    diffusion_current_offset: float | None = None  # nA
    diffusion_current_average: float | None = None  # nA
    diffusion_current_stddev: float | None = None  # nA
    diffusion_current_max: float | None = None  # nA
    diffusion_current_delay_on: float | None = None  # centiseconds
    diffusion_current_delay_off: float | None = None  # centiseconds
    corona_voltage: float | None = None  # V
    corona_voltage_onset: float | None = None  # V
    hires_adc1: float | None = None  # instantaneous value electrometer 1
    hires_adc2: float | None = None  # instantaneous value electrometer 2
    electrometer_1_amplitude: float | None = None  # mV
    electrometer_2_amplitude: float | None = None  # mV
    electrometer_1_gain: float | None = None  # mV as in proto_v2.proto; TODO: confirm on a device
    electrometer_2_gain: float | None = None  # mV as in proto_v2.proto; TODO: confirm on a device
    temperature: float | None = None  # Celsius
    relative_humidity: float | None = None  # percent 0-100
    deposition_voltage: float | None = None  # V
    battery_voltage: float | None = None  # V
    flow_from_dp: float | None = None  # l/min
    ambient_pressure: float | None = None  # hPa
    channel_pressure: float | None = None  # hPa
    differential_pressure: float | None = None  # Pa
    pump_voltage: float | None = None  # V
    pump_current: float | None = None  # mA
    pump_pwm: float | None = None  # percent 0-100
    particle_number_10nm: float | None = None  # /cm^3/log(d)
    particle_number_16nm: float | None = None  # /cm^3/log(d)
    particle_number_26nm: float | None = None  # /cm^3/log(d)
    particle_number_43nm: float | None = None  # /cm^3/log(d)
    particle_number_70nm: float | None = None  # /cm^3/log(d)
    particle_number_114nm: float | None = None  # /cm^3/log(d)
    particle_number_185nm: float | None = None  # /cm^3/log(d)
    particle_number_300nm: float | None = None  # /cm^3/log(d)
    sigma_size_dist: float | None = None  # gsd
    steps_inversion: float | None = None  # steps count
    current_dist_0: float | None = None  # mV
    current_dist_1: float | None = None  # mV
    current_dist_2: float | None = None  # mV
    current_dist_3: float | None = None  # mV
    current_dist_4: float | None = None  # mV

    supply_voltage_5V: float | None = None  # V
    positive_voltage_3V3: float | None = None  # V
    negative_voltage_3V3: float | None = None  # V
    usb_cc_voltage: float | None = None  # V

    def to_dict(self, remove_nan: bool = True) -> dict[str, object]:
        """Field values by name; with remove_nan only the fields that are set."""
        values = {name: getattr(self, name) for name in self.__dataclass_fields__}
        if remove_nan:
            return {name: value for name, value in values.items() if value is not None}
        return values

to_dict(remove_nan=True)

Field values by name; with remove_nan only the fields that are set.

Source code in src/naneos/data_point.py

def to_dict(self, remove_nan: bool = True) -> dict[str, object]:
    """Field values by name; with remove_nan only the fields that are set."""
    values = {name: getattr(self, name) for name in self.__dataclass_fields__}
    if remove_nan:
        return {name: value for name, value in values.items() if value is not None}
    return values

NaneosDeviceManager

Bases: Thread

Connects to every Partector on USB and BLE, gathers their data in snapshots and hands each snapshot to the output queue and / or the naneos upload.

Source code in src/naneos/manager.py

class NaneosDeviceManager(threading.Thread):
    """Connects to every Partector on USB and BLE, gathers their data in snapshots
    and hands each snapshot to the output queue and / or the naneos upload."""

    # UI curves and pulse forms waiting for their upload: about 1 KB each, so
    # this is days of them. The snapshots wait in a backlog capped in MB.
    MAX_PENDING_DIAGNOSTICS = 500
    # A readout that comes back short (over BLE a lost packet) or not at all is read again,
    # this many attempts in all, before it is dropped: the backend spaces the entries by index,
    # so an incomplete curve or form would land in the database shifted.
    DIAGNOSTICS_ATTEMPTS = 3
    # A sweep holds the data of its device back for 15 to 30 s: more often than every half hour
    # is too disturbing. Less often than daily is not worth a schedule.
    MIN_DIAGNOSTICS_INTERVAL_HOURS = 0.5
    MAX_DIAGNOSTICS_INTERVAL_HOURS = 24
    # The most rows (all devices) in one request. The backend needs ~2.7 ms per data point, so
    # 2000 rows are ~6 s: well inside the timeout, and 7 devices at 600 s (4200 rows) are not.
    MAX_CHUNK_ROWS = 2000
    # The sender waits this long after a failed upload, growing to the last value.
    RETRY_DELAYS_SECONDS = (5, 10, 20, 40, 60)
    # A running sender is given this long to finish the request it is in.
    SENDER_JOIN_SECONDS = 12

    def __init__(
        self,
        use_serial: bool = True,
        use_ble: bool = True,
        upload_active: bool = True,
        gathering_interval_seconds: int = 30,
        ble_serial_numbers: Iterable[int] | None = None,
        ble_max_links: int = PartectorBleManager.DEFAULT_MAX_LINKS,
        serial_gain_test: bool = True,
        serial_pulse_diagnostics: bool = True,
        sample_rate_hz: int | None = None,
        diagnostics_interval_hours: float | None = 1.0,
        upload_buffer_mb: float = 100,
        ble_p2pro_mode: bool = True,
    ) -> None:
        """
        Args:
            use_serial: connect to Partectors on USB.
            use_ble: connect to Partectors over Bluetooth (data comes from the links only).
            upload_active: upload every snapshot to the naneos IoT service.
            gathering_interval_seconds: snapshot interval, clamped to 10-600 s.
            ble_serial_numbers: only link to these devices over BLE; None means any in reach.
            ble_max_links: upper bound of simultaneous BLE links.
            serial_gain_test: run the electrometer gain test on USB devices. Their data is
                held back for at least 10 s after every connect while it settles.
            serial_pulse_diagnostics: let USB devices report the pulse diagnostics.
            sample_rate_hz: the data rate of the USB devices, see the property.
            diagnostics_interval_hours: read the UI curve and the pulse form of every
                connected device this often, 0.5 to 24 hours, see the property. None
                switches it off.
            upload_buffer_mb: what is kept in RAM while the upload does not work, in MB
                (a P2 at 1 Hz needs about 16 MB per day). The oldest data is dropped
                when it is full. It is lost when the process ends.
            ble_p2pro_mode: put a P2 Pro into size distribution mode after every BLE
                connect, as the USB connect does. Skipped for a device that is connected
                over USB, and for all devices while `sample_rate_hz` is set: USB decides
                their mode. False never touches the mode over BLE.
        """
        if not upload_buffer_mb > 0:
            raise ValueError("upload_buffer_mb must be positive.")
        super().__init__(daemon=True)
        self._use_serial = use_serial
        self._use_ble = use_ble
        self._ble_serial_numbers = frozenset(ble_serial_numbers) if ble_serial_numbers else None
        self._ble_max_links = ble_max_links
        self._ble_p2pro_mode = ble_p2pro_mode
        self._serial_gain_test = serial_gain_test
        self._serial_pulse_diagnostics = serial_pulse_diagnostics
        self._sample_rate_hz = PartectorSerialManager.check_sample_rate(sample_rate_hz)
        self._upload_active = upload_active
        self._next_upload_time = time.time() + gathering_interval_seconds
        self.gathering_interval_seconds = gathering_interval_seconds

        self._out_queue: queue.Queue | None = None
        self._live_queue: queue.Queue | None = None
        self._live_points_dropped = 0

        self.diagnostics_interval_hours = diagnostics_interval_hours
        self._diagnostics_queue: queue.Queue | None = None
        self._diagnostics_requested = threading.Event()
        self._diagnostics_thread: threading.Thread | None = None
        self._last_diagnostics_block: int | None = None
        self._pending_diagnostics: deque[UiCurve | PulseForm] = deque(
            maxlen=self.MAX_PENDING_DIAGNOSTICS
        )
        self._diagnostics_lock = threading.Lock()  # the readout thread appends, the sender pops

        self._stop_event = threading.Event()

        self._manager_serial: PartectorSerialManager | None = None
        self._manager_ble: PartectorBleManager | None = None

        self._data: dict[int, pd.DataFrame] = {}

        # The upload runs on a thread of its own, so a slow or missing network never
        # holds up the gathering: the loop puts snapshots into the backlog, the
        # sender takes them out.
        self._backlog = UploadBacklog(int(upload_buffer_mb * 1_000_000))
        self._sender: threading.Thread | None = None
        self._wake_sender = threading.Event()
        self._chunk_seconds = CHUNK_SECONDS  # halved when a server chokes on a big request
        self._retry_delay = 0
        self._diagnostics_retry_at = 0.0
        self._diagnostics_retry_delay = 0
        self._outage_since: float | None = None
        self._evicted = Eviction(0, 0)  # dropped since the last log line
        self._evicted_logged_at: float | None = None  # monotonic time of the last log line

        self._upload_blocked_devices: list[int | None] = []

    # == Runtime controls ==========================================================================
    @property
    def use_serial(self) -> bool:
        """USB devices on / off. Takes effect within a second, also while running."""
        return self._use_serial

    @use_serial.setter
    def use_serial(self, use: bool) -> None:
        self._use_serial = use

    @property
    def use_ble(self) -> bool:
        """BLE devices on / off. Takes effect within a second, also while running;
        switching off waits for the links to disconnect."""
        return self._use_ble

    @use_ble.setter
    def use_ble(self, use: bool) -> None:
        self._use_ble = use

    @property
    def upload_active(self) -> bool:
        """Upload of the snapshots to the naneos IoT service on / off."""
        return self._upload_active

    @upload_active.setter
    def upload_active(self, active: bool) -> None:
        self._upload_active = active

    @property
    def sample_rate_hz(self) -> int | None:
        """The data rate of every USB device: 1, 10 or 100 Hz, or None for the
        default of each device (1 Hz, size distribution mode on a P2 Pro).
        Takes effect within a second, also while running; BLE stays at 1 Hz.
        The upload to naneos stays at 1 Hz whatever is set here.
        """
        return self._sample_rate_hz

    @sample_rate_hz.setter
    def sample_rate_hz(self, hz: int | None) -> None:
        self._sample_rate_hz = PartectorSerialManager.check_sample_rate(hz)
        if self._manager_serial is not None:
            self._manager_serial.sample_rate_hz = hz

    @property
    def gathering_interval_seconds(self) -> int:
        """Snapshot interval; values are clamped to 10-600 s."""
        return self._gathering_interval_seconds

    @gathering_interval_seconds.setter
    def gathering_interval_seconds(self, interval: int) -> None:
        interval = max(10, min(600, interval))
        logger.info(f"Setting gathering interval to {interval} seconds.")
        self._gathering_interval_seconds = interval

        tmp_next_upload_time = time.time() + self._gathering_interval_seconds
        self._next_upload_time = min(self._next_upload_time, tmp_next_upload_time)

    @property
    def diagnostics_interval_hours(self) -> float | None:
        """How often the UI curve and the pulse form of every connected device are read
        and uploaded, 0.5 to 24 hours, None for never. The readouts happen at the wall clock
        multiples of the interval (with 1 h: on the hour), one device after the other, so a
        change takes effect at the next multiple; request_diagnostics() reads now.

        A UI curve sweep disturbs the measurement: the data points of the device are
        held back for about 15 s (USB) to 30 s (BLE) per readout.
        """
        return self._diagnostics_interval_hours

    @diagnostics_interval_hours.setter
    def diagnostics_interval_hours(self, hours: float | None) -> None:
        if hours is not None and not (
            self.MIN_DIAGNOSTICS_INTERVAL_HOURS <= hours <= self.MAX_DIAGNOSTICS_INTERVAL_HOURS
        ):
            raise ValueError(
                f"The diagnostics interval must be {self.MIN_DIAGNOSTICS_INTERVAL_HOURS:g} to "
                f"{self.MAX_DIAGNOSTICS_INTERVAL_HOURS:g} hours, or None for never."
            )
        self._diagnostics_interval_hours = hours
        self._last_diagnostics_block = None  # the next multiple counts from now

    @property
    def pending_upload_count(self) -> int:
        """Number of snapshots waiting to be uploaded, including retries.

        The request that is being sent right now is not counted.
        """
        return self._backlog.snapshots

    @property
    def pending_upload_seconds(self) -> int:
        """Seconds of data waiting to be uploaded (per chunk, devices in parallel)."""
        return self._backlog.seconds

    @property
    def pending_upload_bytes(self) -> int:
        """RAM the data waiting to be uploaded takes, of upload_buffer_mb."""
        return self._backlog.size_bytes

    @property
    def pending_diagnostics_count(self) -> int:
        """Number of UI curves and pulse forms waiting to be uploaded, including retries."""
        return len(self._pending_diagnostics)

    def register_diagnostics_queue(self, diagnostics_queue: queue.Queue) -> None:
        """Every UiCurve and PulseForm the manager reads is put on this queue, whether
        it was read on the interval or with request_diagnostics(). Only complete ones (100
        U + 100 I values, 200 I values) get here; see DIAGNOSTICS_ATTEMPTS."""
        self._diagnostics_queue = diagnostics_queue

    def unregister_diagnostics_queue(self) -> None:
        self._diagnostics_queue = None

    def request_diagnostics(self) -> None:
        """Read the UI curve and the pulse form of every connected device now, one
        device after the other, on a thread of the manager. Returns at once; the
        results go to the diagnostics queue and the upload like the periodic ones."""
        self._diagnostics_requested.set()

    @property
    def diagnostics_in_progress(self) -> bool:
        """True while the manager is reading the diagnostics of its devices."""
        thread = self._diagnostics_thread
        return thread is not None and thread.is_alive()

    @property
    def seconds_until_next_snapshot(self) -> float:
        """Time until the next snapshot is put on the output queue and uploaded."""
        return max(0, self._next_upload_time - time.time())

    def register_output_queue(self, out_queue: queue.Queue) -> None:
        """Every snapshot (dict[int, pandas.DataFrame]) is put on this queue."""
        self._out_queue = out_queue

    def unregister_output_queue(self) -> None:
        self._out_queue = None

    def register_live_queue(self, live_queue: queue.Queue) -> None:
        """Every NaneosDeviceDataPoint is put on this queue the moment it arrives.

        Independent of the snapshots and the upload, at the rate of the device
        (see set_sample_rate). A device that is connected over USB and BLE
        delivers its USB points only.

        Give the queue a maxsize: when it is full the oldest point is dropped
        to make room (see live_points_dropped), so a consumer that falls
        behind loses old data instead of stalling the devices. An unbounded
        queue grows without limit when nobody reads it.
        """
        self._live_queue = live_queue

    def unregister_live_queue(self) -> None:
        self._live_queue = None

    @property
    def live_points_dropped(self) -> int:
        """Points dropped from the live queue because it was full."""
        return self._live_points_dropped

    def _ble_may_set_p2pro_mode(self, serial_number: int) -> bool:
        """Asked by a BLE link before it switches a P2 Pro into size distribution mode.

        Runs on the BLE event loop. USB decides the mode of a device that is connected over
        USB, and of all devices while a rate is set (a rate means the plain P2 mode).
        """
        if self._sample_rate_hz is not None:
            return False
        serial_manager = self._manager_serial
        return (
            serial_manager is None
            or serial_number not in serial_manager.get_connected_serial_numbers()
        )

    def _on_live_point(self, point: NaneosDeviceDataPoint) -> None:
        """Runs on the serial reader threads and the BLE event loop: never blocks."""
        live_queue = self._live_queue
        if live_queue is None:
            return

        if point.connection_type == ConnectionType.CONNECTED:
            serial_manager = self._manager_serial
            if (
                serial_manager is not None
                and point.serial_number in serial_manager.get_connected_serial_numbers()
            ):
                return  # the USB connection of this device delivers the same measurement

        try:
            live_queue.put_nowait(point)
            return
        except queue.Full:
            pass

        # Drop the oldest point. Another producer may win the freed slot, then this
        # point is the one that is lost; either way nothing blocks.
        self._live_points_dropped += 1
        try:
            live_queue.get_nowait()
            live_queue.put_nowait(point)
        except (queue.Empty, queue.Full):
            pass

    def run(self) -> None:
        self._sender = threading.Thread(target=self._sender_loop, name="naneos-upload", daemon=True)
        self._sender.start()
        self._loop()

        # graceful shutdown in any case
        self._use_serial = False
        self._loop_serial_manager()
        self._use_ble = False
        self._loop_ble_manager()
        self._sender.join(timeout=self.SENDER_JOIN_SECONDS)

    def stop(self) -> None:
        self._stop_event.set()
        self._wake_sender.set()

    def get_devices(self) -> list[PartectorDevice]:
        """One handle per connected device, to write to it, query it and set its rate.

        A device that is reachable both ways is listed with its USB connection,
        like its data: USB is faster and the only way to change the data rate.
        """
        devices: dict[int | None, PartectorDevice] = {}
        for manager in (self._manager_ble, self._manager_serial):  # serial wins
            if manager is not None:
                devices.update({device.serial_number: device for device in manager.get_devices()})
        return list(devices.values())

    def get_device(self, serial_number: int) -> PartectorDevice:
        """The handle of one connected device.

        Raises:
            KeyError: no device with this serial number is connected.
        """
        for device in self.get_devices():
            if device.serial_number == serial_number:
                return device
        raise KeyError(f"SN{serial_number} is not connected.")

    def write(self, serial_number: int, command: str) -> None:
        """Send a command without an answer to a device, see PartectorDevice.write()."""
        self.get_device(serial_number).write(command)

    def query(self, serial_number: int, command: str, timeout: float | None = None) -> list[str]:
        """Send a command to a device and return its answer, see PartectorDevice.query()."""
        return self.get_device(serial_number).query(command, timeout)

    def set_sample_rate(self, serial_number: int, hz: int | None) -> None:
        """Set the data rate of one device on USB, see PartectorDevice.set_sample_rate().
        For all devices at once, and for the ones that connect later, set sample_rate_hz.

        The output queue gets the data at this rate; the upload stays at 1 Hz.
        """
        self.get_device(serial_number).set_sample_rate(hz)

    def read_ui_curve(self, serial_number: int, timeout: float | None = None) -> UiCurve:
        """Read the UI curve of one device, see PartectorDevice.read_ui_curve().
        Blocks for 15 s or more; the result is returned, not uploaded."""
        return self.get_device(serial_number).read_ui_curve(timeout)

    def read_pulse_form(self, serial_number: int, timeout: float | None = None) -> PulseForm:
        """Read the pulse form of one device, see PartectorDevice.read_pulse_form().
        The result is returned, not uploaded."""
        return self.get_device(serial_number).read_pulse_form(timeout)

    # == Diagnostics ===============================================================================
    def _tick_diagnostics(self) -> None:
        """Once a second on the manager thread: start the readouts when they are due."""
        interval = self._diagnostics_interval_hours
        if interval is not None:
            block = int(time.time() // (interval * 3600))
            if self._last_diagnostics_block is None:
                self._last_diagnostics_block = block  # the first readout at the next multiple
            elif block != self._last_diagnostics_block:
                self._last_diagnostics_block = block
                self._diagnostics_requested.set()

        if self._diagnostics_requested.is_set() and not self.diagnostics_in_progress:
            self._diagnostics_requested.clear()
            self._diagnostics_thread = threading.Thread(
                target=self._read_all_diagnostics, name="naneos-diagnostics", daemon=True
            )
            self._diagnostics_thread.start()

    def _read_all_diagnostics(self) -> None:
        """One device after the other: the readouts hold the command lock of their
        device, and the BLE links share one radio."""
        for device in self.get_devices():
            if self._stop_event.is_set():
                return
            # The curve first: a "UI!" interrupts a pulse form that is still streaming.
            for what, read in (
                ("UI curve", device.read_ui_curve),
                ("pulse form", device.read_pulse_form),
            ):
                try:
                    result = self._read_diagnostic(device, what, read)
                except NotSupportedError as e:
                    logger.debug(f"SN{device.serial_number}: no diagnostics: {e}")
                    break
                if result is not None:
                    self._publish_diagnostic(result)

    def _read_diagnostic(
        self, device: PartectorDevice, what: str, read: Callable[[], UiCurve | PulseForm]
    ) -> UiCurve | PulseForm | None:
        """Read one UI curve or pulse form that has all its entries.

        A result with too few or too many entries, or none within the timeout, is read again,
        DIAGNOSTICS_ATTEMPTS attempts in all (a UI curve retry sweeps again). Returns None
        when no attempt was complete, or when the device failed in a way that a retry does
        not help (not connected): only complete results are ever published.

        Raises:
            NotSupportedError: the device has no such diagnostic.
        """
        link = "USB" if device.connection_type == ConnectionType.SERIAL else "BLE"
        who = f"SN{device.serial_number} over {link}"
        attempts = self.DIAGNOSTICS_ATTEMPTS
        problem = ""

        for attempt in range(1, attempts + 1):
            if self._stop_event.is_set():
                return None
            try:
                result = read()
            except NotSupportedError:
                raise
            except (TimeoutError, ValueError) as e:  # the end never came, or a garbled line
                problem = str(e) or type(e).__name__
                logger.warning(f"{who}: {what} attempt {attempt} of {attempts} failed: {problem}")
                continue
            except (ConnectionError, RuntimeError) as e:
                logger.warning(f"{who}: {what} failed: {e}")
                return None
            except Exception as e:
                logger.exception(f"{who}: {what} failed: {e}")
                return None

            if result.is_complete:
                logger.info(
                    f"{who}: read the {what}, {result.entries} (attempt {attempt} of {attempts})"
                )
                return result
            problem = f"{result.entries}, expected {type(result).EXPECTED_ENTRIES}"
            logger.warning(f"{who}: {what} attempt {attempt} of {attempts} incomplete: {problem}")

        logger.error(
            f"{who}: gave up on the {what} after {attempts} attempts ({problem}), not uploaded."
        )
        return None

    def _publish_diagnostic(self, diagnostic: UiCurve | PulseForm) -> None:
        """Hand a UI curve or pulse form to the diagnostics queue and the uploader.
        The upload itself runs on the manager thread with the snapshots."""
        if isinstance(self._diagnostics_queue, queue.Queue):
            self._diagnostics_queue.put(diagnostic)
        if self._upload_active:
            with self._diagnostics_lock:
                self._pending_diagnostics.append(diagnostic)
            self._wake_sender.set()

    def _loop_serial_manager(self) -> None:
        if self._manager_serial is not None and self._manager_serial.is_alive():
            self._upload_blocked_devices = self._manager_serial.get_settling_serial_numbers()
            self._data = add_to_existing_naneos_data(self._data, self._manager_serial.get_data())

        self._manager_serial = self._sync_manager(
            self._manager_serial,
            self._use_serial,
            lambda: PartectorSerialManager(
                self._serial_gain_test,
                self._serial_pulse_diagnostics,
                self._on_live_point,
                self._sample_rate_hz,
            ),
            "serial",
        )

    def _loop_ble_manager(self) -> None:
        if self._manager_ble is not None and self._manager_ble.is_alive():
            self._data = add_to_existing_naneos_data(self._data, self._manager_ble.get_data())

        self._manager_ble = self._sync_manager(
            self._manager_ble,
            self._use_ble,
            lambda: PartectorBleManager(
                self._ble_serial_numbers,
                self._ble_max_links,
                self._on_live_point,
                self._ble_p2pro_mode,
                self._ble_may_set_p2pro_mode,
            ),
            "BLE",
        )

    @staticmethod
    def _sync_manager(
        manager: ManagerT | None, wanted: bool, factory: Callable[[], ManagerT], name: str
    ) -> ManagerT | None:
        """Starts or stops a sub-manager so that it matches the wanted state."""
        if manager is None and wanted:
            logger.info(f"Starting {name} manager...")
            manager = factory()
            manager.start()
        elif manager is not None and not wanted:
            logger.info(f"Stopping {name} manager...")
            manager.stop()
            manager.join()
            manager = None
        return manager

    def _loop(self) -> None:
        self._next_upload_time = time.time() + self._gathering_interval_seconds

        while not self._stop_event.is_set():
            try:
                time.sleep(1)

                self._loop_serial_manager()
                self._loop_ble_manager()
                self._tick_diagnostics()

                # a device that is settling after a gain test start delivers no valid data
                for blocked_sn in self._upload_blocked_devices:
                    if blocked_sn in self._data:
                        del self._data[blocked_sn]

                if time.time() >= self._next_upload_time:
                    self._next_upload_time = time.time() + self._gathering_interval_seconds

                    serial_connected_sns: list[int | None] = []
                    if self._use_serial and self._manager_serial is not None:
                        serial_connected_sns = self._manager_serial.get_connected_serial_numbers()

                    upload_data = sort_and_clean_naneos_data(self._data, serial_connected_sns)
                    self._data = {}

                    self._publish_snapshot(upload_data)

            except Exception as e:
                logger.exception(f"DeviceManager loop exception: {e}")

    def _publish_snapshot(self, snapshot: dict[int, pd.DataFrame]) -> None:
        """Hand a gathered snapshot to the output queue and the uploader.

        The upload never runs here: the snapshot goes into the backlog, and the
        sender thread takes it from there.
        """
        if isinstance(self._out_queue, queue.Queue):
            self._out_queue.put(snapshot)

        if not self._upload_active:
            return

        try:
            evicted = self._backlog.add(prepare_frames(snapshot))
        except Exception as e:  # a frame the upload cannot read must not stop the gathering
            logger.exception(f"Could not keep a snapshot for the upload: {e}")
            return
        self._log_eviction(evicted)
        self._wake_sender.set()

    def _sender_loop(self) -> None:
        """The thread that talks to the network, until the manager stops."""
        while not self._stop_event.is_set():
            try:
                done = self._drain_uploads()
            except Exception as e:
                logger.exception(f"Upload thread exception: {e}")
                done = False

            if done:
                self._retry_delay = 0
                if self._wake_sender.wait(timeout=1.0):
                    self._wake_sender.clear()
            else:
                delays = self.RETRY_DELAYS_SECONDS
                later = [delay for delay in delays if delay > self._retry_delay]
                self._retry_delay = later[0] if later else delays[-1]
                self._stop_event.wait(self._retry_delay)

    def _drain_uploads(self) -> bool:
        """Send what is waiting, oldest data first: the snapshots, then the diagnostics.

        The point timestamps are relative to the moment of sending, so data that
        waited for hours lands at its own time on the server. While the backlog is
        long, requests carry up to CHUNK_SECONDS of data.

        Returns False when a snapshot could not be sent and should be tried again
        later (it stays in the backlog), True otherwise.
        """
        if not self._upload_active:
            return True

        while not self._stop_event.is_set():
            chunk = self._backlog.take(self._chunk_seconds, self.MAX_CHUNK_ROWS)
            if chunk is None:
                break
            outcome = self._try_upload(chunk)
            if outcome in ("retry", "server"):
                self._snapshot_failed(chunk, outcome)
                return False
            self._snapshot_sent(chunk)

        self._drain_diagnostics()
        return True

    def _snapshot_failed(self, chunk: Chunk, outcome: str) -> None:
        self._log_eviction(self._backlog.restore(chunk))
        if self._outage_since is None:
            self._outage_since = time.time()

        smallest = self._gathering_interval_seconds
        if outcome == "server" and chunk.seconds > smallest:
            self._chunk_seconds = max(smallest, self._chunk_seconds // 2)
            logger.warning(
                f"The server refused a request, trying {self._chunk_seconds} s at a time."
            )
        logger.warning(
            f"Upload failed, keeping {self._backlog.snapshots} snapshot(s) "
            f"({self._backlog.size_bytes / 1e6:.1f} MB) for retry."
        )

    def _snapshot_sent(self, chunk: Chunk) -> None:
        if self._chunk_seconds < CHUNK_SECONDS:
            self._chunk_seconds = min(CHUNK_SECONDS, self._chunk_seconds * 2)
        if self._outage_since is not None:
            minutes = (time.time() - self._outage_since) / 60
            logger.info(
                f"Upload works again after {minutes:.0f} min, "
                f"{self._backlog.snapshots} snapshot(s) still to send."
            )
            self._outage_since = None

    def _drain_diagnostics(self) -> None:
        """Send the UI curves and pulse forms. A failing endpoint has a pause of its own,
        so that it does not delay the snapshots."""
        while not self._stop_event.is_set() and time.monotonic() >= self._diagnostics_retry_at:
            with self._diagnostics_lock:
                item = self._pending_diagnostics[0] if self._pending_diagnostics else None
            if item is None:
                return

            outcome = self._try_upload(item)
            if outcome in ("retry", "server"):
                delays = self.RETRY_DELAYS_SECONDS
                later = [d for d in delays if d > self._diagnostics_retry_delay]
                self._diagnostics_retry_delay = later[0] if later else delays[-1]
                self._diagnostics_retry_at = time.monotonic() + self._diagnostics_retry_delay
                logger.warning(
                    f"Upload failed, keeping {len(self._pending_diagnostics)} diagnostic(s) "
                    "for retry."
                )
                return

            self._diagnostics_retry_delay = 0
            with self._diagnostics_lock:
                if self._pending_diagnostics and self._pending_diagnostics[0] is item:
                    self._pending_diagnostics.popleft()

    def _log_eviction(self, evicted: Eviction | None) -> None:
        """Say that a full buffer dropped data, at most once a minute."""
        if evicted is None:
            return
        self._evicted = Eviction(
            self._evicted.snapshots + evicted.snapshots, self._evicted.seconds + evicted.seconds
        )
        now = time.monotonic()
        if self._evicted_logged_at is not None and now - self._evicted_logged_at < 60:
            return
        self._evicted_logged_at = now
        logger.warning(
            f"Upload buffer is full ({self._backlog.size_bytes / 1e6:.0f} MB): dropped the oldest "
            f"{self._evicted.snapshots} snapshot(s), about {self._evicted.seconds / 60:.0f} min "
            "of data."
        )
        self._evicted = Eviction(0, 0)

    @staticmethod
    def _try_upload(item: Chunk | UiCurve | PulseForm) -> str:
        """Returns "ok", "drop" (rejected), "retry" (no answer, the network) or "server"
        (an answer that says try again later, or none in time: the server is too slow for
        a request of this size)."""
        try:
            if isinstance(item, UiCurve | PulseForm):
                response = upload_diagnostic(item)
            else:
                response = send_frames(item.frames)
        except ReadTimeout as e:
            logger.warning(f"Upload timed out, the server did not answer: {e}")
            return "server"
        except Exception as e:
            logger.warning(f"Upload failed: {e}")
            return "retry"

        status, detail = backend_status(response)
        if 200 <= status < 300:
            logger.info(f"Uploaded {NaneosDeviceManager._describe(item)}")
            return "ok"
        if status >= 500 or status in (408, 429):
            logger.warning(f"Upload failed with HTTP {status} {detail}, will retry.")
            return "server"

        # A 4xx will not get better by resending the same payload, and a retry
        # would hold up everything queued behind it.
        what = NaneosDeviceManager._describe(item)
        wrapped = f" (inside HTTP {response.status_code})" if status != response.status_code else ""
        logger.error(f"Upload rejected with HTTP {status}{wrapped} {detail}, dropping {what}.")
        return "drop"

    @staticmethod
    def _describe(item: Chunk | UiCurve | PulseForm) -> str:
        """What a request carried, for the log: a burst of uploads should say what it is."""
        if isinstance(item, UiCurve | PulseForm):
            return f"{type(item).__name__} of SN{item.serial_number} ({item.entries})"
        merged = f", {item.snapshots} snapshots merged" if item.snapshots > 1 else ""
        return f"snapshot: {len(item.frames)} device(s), {item.rows} rows, {item.seconds} s{merged}"

diagnostics_in_progress property

True while the manager is reading the diagnostics of its devices.

diagnostics_interval_hours property writable

How often the UI curve and the pulse form of every connected device are read and uploaded, 0.5 to 24 hours, None for never. The readouts happen at the wall clock multiples of the interval (with 1 h: on the hour), one device after the other, so a change takes effect at the next multiple; request_diagnostics() reads now.

A UI curve sweep disturbs the measurement: the data points of the device are held back for about 15 s (USB) to 30 s (BLE) per readout.

gathering_interval_seconds property writable

Snapshot interval; values are clamped to 10-600 s.

live_points_dropped property

Points dropped from the live queue because it was full.

pending_diagnostics_count property

Number of UI curves and pulse forms waiting to be uploaded, including retries.

pending_upload_bytes property

RAM the data waiting to be uploaded takes, of upload_buffer_mb.

pending_upload_count property

Number of snapshots waiting to be uploaded, including retries.

The request that is being sent right now is not counted.

pending_upload_seconds property

Seconds of data waiting to be uploaded (per chunk, devices in parallel).

sample_rate_hz property writable

The data rate of every USB device: 1, 10 or 100 Hz, or None for the default of each device (1 Hz, size distribution mode on a P2 Pro). Takes effect within a second, also while running; BLE stays at 1 Hz. The upload to naneos stays at 1 Hz whatever is set here.

seconds_until_next_snapshot property

Time until the next snapshot is put on the output queue and uploaded.

upload_active property writable

Upload of the snapshots to the naneos IoT service on / off.

use_ble property writable

BLE devices on / off. Takes effect within a second, also while running; switching off waits for the links to disconnect.

use_serial property writable

USB devices on / off. Takes effect within a second, also while running.

__init__(use_serial=True, use_ble=True, upload_active=True, gathering_interval_seconds=30, ble_serial_numbers=None, ble_max_links=PartectorBleManager.DEFAULT_MAX_LINKS, serial_gain_test=True, serial_pulse_diagnostics=True, sample_rate_hz=None, diagnostics_interval_hours=1.0, upload_buffer_mb=100, ble_p2pro_mode=True)

Parameters:

Name	Type	Description	Default
use_serial	bool	connect to Partectors on USB.	True
use_ble	bool	connect to Partectors over Bluetooth (data comes from the links only).	True
upload_active	bool	upload every snapshot to the naneos IoT service.	True
gathering_interval_seconds	int	snapshot interval, clamped to 10-600 s.	30
ble_serial_numbers	Iterable[int] | None	only link to these devices over BLE; None means any in reach.	None
ble_max_links	int	upper bound of simultaneous BLE links.	DEFAULT_MAX_LINKS
serial_gain_test	bool	run the electrometer gain test on USB devices. Their data is held back for at least 10 s after every connect while it settles.	True
serial_pulse_diagnostics	bool	let USB devices report the pulse diagnostics.	True
sample_rate_hz	int | None	the data rate of the USB devices, see the property.	None
diagnostics_interval_hours	float | None	read the UI curve and the pulse form of every connected device this often, 0.5 to 24 hours, see the property. None switches it off.	1.0
upload_buffer_mb	float	what is kept in RAM while the upload does not work, in MB (a P2 at 1 Hz needs about 16 MB per day). The oldest data is dropped when it is full. It is lost when the process ends.	100
ble_p2pro_mode	bool	put a P2 Pro into size distribution mode after every BLE connect, as the USB connect does. Skipped for a device that is connected over USB, and for all devices while sample_rate_hz is set: USB decides their mode. False never touches the mode over BLE.	True

Source code in src/naneos/manager.py

def __init__(
    self,
    use_serial: bool = True,
    use_ble: bool = True,
    upload_active: bool = True,
    gathering_interval_seconds: int = 30,
    ble_serial_numbers: Iterable[int] | None = None,
    ble_max_links: int = PartectorBleManager.DEFAULT_MAX_LINKS,
    serial_gain_test: bool = True,
    serial_pulse_diagnostics: bool = True,
    sample_rate_hz: int | None = None,
    diagnostics_interval_hours: float | None = 1.0,
    upload_buffer_mb: float = 100,
    ble_p2pro_mode: bool = True,
) -> None:
    """
    Args:
        use_serial: connect to Partectors on USB.
        use_ble: connect to Partectors over Bluetooth (data comes from the links only).
        upload_active: upload every snapshot to the naneos IoT service.
        gathering_interval_seconds: snapshot interval, clamped to 10-600 s.
        ble_serial_numbers: only link to these devices over BLE; None means any in reach.
        ble_max_links: upper bound of simultaneous BLE links.
        serial_gain_test: run the electrometer gain test on USB devices. Their data is
            held back for at least 10 s after every connect while it settles.
        serial_pulse_diagnostics: let USB devices report the pulse diagnostics.
        sample_rate_hz: the data rate of the USB devices, see the property.
        diagnostics_interval_hours: read the UI curve and the pulse form of every
            connected device this often, 0.5 to 24 hours, see the property. None
            switches it off.
        upload_buffer_mb: what is kept in RAM while the upload does not work, in MB
            (a P2 at 1 Hz needs about 16 MB per day). The oldest data is dropped
            when it is full. It is lost when the process ends.
        ble_p2pro_mode: put a P2 Pro into size distribution mode after every BLE
            connect, as the USB connect does. Skipped for a device that is connected
            over USB, and for all devices while `sample_rate_hz` is set: USB decides
            their mode. False never touches the mode over BLE.
    """
    if not upload_buffer_mb > 0:
        raise ValueError("upload_buffer_mb must be positive.")
    super().__init__(daemon=True)
    self._use_serial = use_serial
    self._use_ble = use_ble
    self._ble_serial_numbers = frozenset(ble_serial_numbers) if ble_serial_numbers else None
    self._ble_max_links = ble_max_links
    self._ble_p2pro_mode = ble_p2pro_mode
    self._serial_gain_test = serial_gain_test
    self._serial_pulse_diagnostics = serial_pulse_diagnostics
    self._sample_rate_hz = PartectorSerialManager.check_sample_rate(sample_rate_hz)
    self._upload_active = upload_active
    self._next_upload_time = time.time() + gathering_interval_seconds
    self.gathering_interval_seconds = gathering_interval_seconds

    self._out_queue: queue.Queue | None = None
    self._live_queue: queue.Queue | None = None
    self._live_points_dropped = 0

    self.diagnostics_interval_hours = diagnostics_interval_hours
    self._diagnostics_queue: queue.Queue | None = None
    self._diagnostics_requested = threading.Event()
    self._diagnostics_thread: threading.Thread | None = None
    self._last_diagnostics_block: int | None = None
    self._pending_diagnostics: deque[UiCurve | PulseForm] = deque(
        maxlen=self.MAX_PENDING_DIAGNOSTICS
    )
    self._diagnostics_lock = threading.Lock()  # the readout thread appends, the sender pops

    self._stop_event = threading.Event()

    self._manager_serial: PartectorSerialManager | None = None
    self._manager_ble: PartectorBleManager | None = None

    self._data: dict[int, pd.DataFrame] = {}

    # The upload runs on a thread of its own, so a slow or missing network never
    # holds up the gathering: the loop puts snapshots into the backlog, the
    # sender takes them out.
    self._backlog = UploadBacklog(int(upload_buffer_mb * 1_000_000))
    self._sender: threading.Thread | None = None
    self._wake_sender = threading.Event()
    self._chunk_seconds = CHUNK_SECONDS  # halved when a server chokes on a big request
    self._retry_delay = 0
    self._diagnostics_retry_at = 0.0
    self._diagnostics_retry_delay = 0
    self._outage_since: float | None = None
    self._evicted = Eviction(0, 0)  # dropped since the last log line
    self._evicted_logged_at: float | None = None  # monotonic time of the last log line

    self._upload_blocked_devices: list[int | None] = []

get_device(serial_number)

The handle of one connected device.

Raises:

Type	Description
KeyError	no device with this serial number is connected.

Source code in src/naneos/manager.py

def get_device(self, serial_number: int) -> PartectorDevice:
    """The handle of one connected device.

    Raises:
        KeyError: no device with this serial number is connected.
    """
    for device in self.get_devices():
        if device.serial_number == serial_number:
            return device
    raise KeyError(f"SN{serial_number} is not connected.")

get_devices()

One handle per connected device, to write to it, query it and set its rate.

A device that is reachable both ways is listed with its USB connection, like its data: USB is faster and the only way to change the data rate.

Source code in src/naneos/manager.py

def get_devices(self) -> list[PartectorDevice]:
    """One handle per connected device, to write to it, query it and set its rate.

    A device that is reachable both ways is listed with its USB connection,
    like its data: USB is faster and the only way to change the data rate.
    """
    devices: dict[int | None, PartectorDevice] = {}
    for manager in (self._manager_ble, self._manager_serial):  # serial wins
        if manager is not None:
            devices.update({device.serial_number: device for device in manager.get_devices()})
    return list(devices.values())

query(serial_number, command, timeout=None)

Send a command to a device and return its answer, see PartectorDevice.query().

Source code in src/naneos/manager.py

def query(self, serial_number: int, command: str, timeout: float | None = None) -> list[str]:
    """Send a command to a device and return its answer, see PartectorDevice.query()."""
    return self.get_device(serial_number).query(command, timeout)

read_pulse_form(serial_number, timeout=None)

Read the pulse form of one device, see PartectorDevice.read_pulse_form(). The result is returned, not uploaded.

Source code in src/naneos/manager.py

def read_pulse_form(self, serial_number: int, timeout: float | None = None) -> PulseForm:
    """Read the pulse form of one device, see PartectorDevice.read_pulse_form().
    The result is returned, not uploaded."""
    return self.get_device(serial_number).read_pulse_form(timeout)

read_ui_curve(serial_number, timeout=None)

Read the UI curve of one device, see PartectorDevice.read_ui_curve(). Blocks for 15 s or more; the result is returned, not uploaded.

Source code in src/naneos/manager.py

def read_ui_curve(self, serial_number: int, timeout: float | None = None) -> UiCurve:
    """Read the UI curve of one device, see PartectorDevice.read_ui_curve().
    Blocks for 15 s or more; the result is returned, not uploaded."""
    return self.get_device(serial_number).read_ui_curve(timeout)

register_diagnostics_queue(diagnostics_queue)

Every UiCurve and PulseForm the manager reads is put on this queue, whether it was read on the interval or with request_diagnostics(). Only complete ones (100 U + 100 I values, 200 I values) get here; see DIAGNOSTICS_ATTEMPTS.

Source code in src/naneos/manager.py

def register_diagnostics_queue(self, diagnostics_queue: queue.Queue) -> None:
    """Every UiCurve and PulseForm the manager reads is put on this queue, whether
    it was read on the interval or with request_diagnostics(). Only complete ones (100
    U + 100 I values, 200 I values) get here; see DIAGNOSTICS_ATTEMPTS."""
    self._diagnostics_queue = diagnostics_queue

register_live_queue(live_queue)

Every NaneosDeviceDataPoint is put on this queue the moment it arrives.

Independent of the snapshots and the upload, at the rate of the device (see set_sample_rate). A device that is connected over USB and BLE delivers its USB points only.

Give the queue a maxsize: when it is full the oldest point is dropped to make room (see live_points_dropped), so a consumer that falls behind loses old data instead of stalling the devices. An unbounded queue grows without limit when nobody reads it.

Source code in src/naneos/manager.py

def register_live_queue(self, live_queue: queue.Queue) -> None:
    """Every NaneosDeviceDataPoint is put on this queue the moment it arrives.

    Independent of the snapshots and the upload, at the rate of the device
    (see set_sample_rate). A device that is connected over USB and BLE
    delivers its USB points only.

    Give the queue a maxsize: when it is full the oldest point is dropped
    to make room (see live_points_dropped), so a consumer that falls
    behind loses old data instead of stalling the devices. An unbounded
    queue grows without limit when nobody reads it.
    """
    self._live_queue = live_queue

register_output_queue(out_queue)

Every snapshot (dict[int, pandas.DataFrame]) is put on this queue.

Source code in src/naneos/manager.py

def register_output_queue(self, out_queue: queue.Queue) -> None:
    """Every snapshot (dict[int, pandas.DataFrame]) is put on this queue."""
    self._out_queue = out_queue

request_diagnostics()

Read the UI curve and the pulse form of every connected device now, one device after the other, on a thread of the manager. Returns at once; the results go to the diagnostics queue and the upload like the periodic ones.

Source code in src/naneos/manager.py

def request_diagnostics(self) -> None:
    """Read the UI curve and the pulse form of every connected device now, one
    device after the other, on a thread of the manager. Returns at once; the
    results go to the diagnostics queue and the upload like the periodic ones."""
    self._diagnostics_requested.set()

set_sample_rate(serial_number, hz)

Set the data rate of one device on USB, see PartectorDevice.set_sample_rate(). For all devices at once, and for the ones that connect later, set sample_rate_hz.

The output queue gets the data at this rate; the upload stays at 1 Hz.

Source code in src/naneos/manager.py

def set_sample_rate(self, serial_number: int, hz: int | None) -> None:
    """Set the data rate of one device on USB, see PartectorDevice.set_sample_rate().
    For all devices at once, and for the ones that connect later, set sample_rate_hz.

    The output queue gets the data at this rate; the upload stays at 1 Hz.
    """
    self.get_device(serial_number).set_sample_rate(hz)

write(serial_number, command)

Send a command without an answer to a device, see PartectorDevice.write().

Source code in src/naneos/manager.py

def write(self, serial_number: int, command: str) -> None:
    """Send a command without an answer to a device, see PartectorDevice.write()."""
    self.get_device(serial_number).write(command)

NotSupportedError

Bases: Exception

The device or the transport it is reached over cannot do this.

Source code in src/naneos/device.py

class NotSupportedError(Exception):
    """The device or the transport it is reached over cannot do this."""

PartectorBleManager

Bases: Thread

Connects to the Partectors in reach and collects the data they send over the link.

Only connected devices deliver data. The scanner is used to find devices, to gate connects by signal strength and to refresh stale device handles.

Parameters:

Name	Type	Description	Default
serial_numbers	Iterable[int] | None	Only connect to these devices. None connects to every Partector in reach, first come first served.	None
max_links	int	Upper bound of simultaneous links (including ones that are still retrying). BlueZ handles about seven reliably.	DEFAULT_MAX_LINKS
point_listener	PointListener | None	Called with every data point as it arrives, in addition to get_data(). Runs on the BLE event loop: must be quick and must not block.	None
p2pro_mode	bool	Put a P2 Pro into size distribution mode after every connect, like the USB connect does. False leaves the mode of the device alone.	True
p2pro_mode_guard	Callable[[int], bool] | None	Called with the serial number before that switch, on the BLE thread; returning False skips it for this connect.	None

Source code in src/naneos/ble/partector/manager.py

class PartectorBleManager(threading.Thread):
    """Connects to the Partectors in reach and collects the data they send over the link.

    Only connected devices deliver data. The scanner is used to find devices,
    to gate connects by signal strength and to refresh stale device handles.

    Args:
        serial_numbers: Only connect to these devices. None connects to every
            Partector in reach, first come first served.
        max_links: Upper bound of simultaneous links (including ones that are
            still retrying). BlueZ handles about seven reliably.
        point_listener: Called with every data point as it arrives, in addition to
            get_data(). Runs on the BLE event loop: must be quick and must not block.
        p2pro_mode: Put a P2 Pro into size distribution mode after every connect, like
            the USB connect does. False leaves the mode of the device alone.
        p2pro_mode_guard: Called with the serial number before that switch, on the BLE
            thread; returning False skips it for this connect.
    """

    # How often the manager drains its queues. The queues are bounded and the
    # producers are ~1Hz per device, so polling faster only burns CPU on a
    # Raspberry Pi Zero 2 W without delivering data any sooner.
    LOOP_INTERVAL_SECONDS = 1.0

    # The adapter is considered lost after discovery has been down this long.
    # Discovery failing *is* the adapter check, so no subprocess is needed while
    # running; `bluetoothctl` is only used to decide when it is safe to restart.
    ADAPTER_LOST_AFTER_SECONDS = 30.0
    ADAPTER_CHECK_INTERVAL_SECONDS = 3.0
    # A check that takes this long is logged once. On macOS it does not return until the
    # user has answered the Bluetooth permission dialog, and nothing else would say so.
    ADAPTER_CHECK_SLOW_SECONDS = 10.0
    # While waiting for an adapter the stop event is looked at this often. It is a
    # threading.Event, so it cannot be awaited.
    STOP_POLL_SECONDS = 0.2

    # A link that is down makes its device advertise again, so a scanner that
    # hears no Partector at all for this long while a link waits to reconnect
    # has probably gone deaf, and is restarted. The device may also just be
    # switched off, so the interval doubles per fruitless restart up to the cap.
    SCANNER_SILENCE_SECONDS = 60.0
    SCANNER_RESTART_MAX_INTERVAL_SECONDS = 600.0

    # On a normal stop the links get this long to disconnect gracefully before
    # their tasks are cancelled.
    SHUTDOWN_GRACE_SECONDS = 8.0

    DEFAULT_MAX_LINKS = 7

    def __init__(
        self,
        serial_numbers: Iterable[int] | None = None,
        max_links: int = DEFAULT_MAX_LINKS,
        point_listener: PointListener | None = None,
        p2pro_mode: bool = True,
        p2pro_mode_guard: Callable[[int], bool] | None = None,
    ) -> None:
        super().__init__(daemon=True)
        self._point_listener = point_listener
        self._p2pro_mode = p2pro_mode
        self._p2pro_mode_guard = p2pro_mode_guard
        self._allowed_serials: frozenset[int] | None = (
            frozenset(serial_numbers) if serial_numbers is not None else None
        )
        self._max_links = max(1, max_links)
        self._stop_event = threading.Event()
        # Ends the connection tasks of the current scanner session; only polled,
        # never awaited, so setting it from another thread is fine.
        self._task_stop_event = asyncio.Event()

        self._queue_scanner = PartectorBleScanner.create_scanner_queue()
        self._queue_connection = PartectorBleConnection.create_connection_queue()
        self._links: dict[int, BleLink] = {}  # key: serial number
        self._rejected_for_cap: set[int] = set()
        self._scanner: PartectorBleScanner | None = None
        self._link_down_since: float | None = None
        self._scanner_restart_interval = self.SCANNER_SILENCE_SECONDS
        self._silent_scanner_restarts = 0

        # Raw data points from the links, converted to DataFrames only in
        # get_data(). Building them here would put pandas on the event loop that
        # also services the BLE notifications, which on a Raspberry Pi Zero 2 W is
        # enough to stall the links themselves.
        self._points: dict[int, list[NaneosDeviceDataPoint]] = {}
        self._points_lock = threading.Lock()  # get_data() swaps the buffer under the loop's feet

    # == Public API (any thread) ===================================================================
    def get_data(self) -> dict[int, pd.DataFrame]:
        """Returns the collected data as DataFrames and clears the buffer."""
        # Swap first: the BLE thread keeps appending while we convert. Only the swap
        # is locked, so the conversion never holds up the event loop.
        with self._points_lock:
            points, self._points = self._points, {}

        return {
            serial: df
            for serial, serial_points in points.items()
            if not (df := to_pandas_df(serial_points)).empty
        }

    def stop(self) -> None:
        self._task_stop_event.set()
        self._stop_event.set()

    def run(self) -> None:
        try:
            asyncio.run(self._async_run())
        except RuntimeError as e:
            logger.exception(f"BLEManager loop exited with: {e}")

    def get_connected_serial_numbers(self) -> list[int]:
        """Serial numbers of the devices with a live BLE link."""
        return [sn for sn, _ in self._live_links()]

    def get_devices(self) -> list[PartectorDevice]:
        """Handles to write to and query the devices with a live BLE link."""
        return [link.device for _, link in self._live_links() if link.device is not None]

    def _live_links(self) -> list[tuple[int, BleLink]]:
        # Copy first: the event loop thread changes the dict while we iterate.
        return [(sn, link) for sn, link in list(self._links.items()) if link.is_connected]

    # == Event loop ================================================================================
    async def _async_run(self) -> None:
        self._loop = asyncio.get_running_loop()

        while not self._stop_event.is_set():
            await self._wait_for_bluetooth_adapter()
            if self._stop_event.is_set():
                break

            self._task_stop_event.clear()
            self._scanner = PartectorBleScanner(loop=self._loop, queue=self._queue_scanner)
            adapter_lost = False
            try:
                async with self._scanner:
                    logger.info("Scanner started.")
                    adapter_lost = await self._manager_loop()
            except asyncio.CancelledError:
                logger.info("BLEManager cancelled.")
                break
            finally:
                # A lost adapter cannot disconnect anything gracefully anyway.
                grace = 0.0 if adapter_lost else self.SHUTDOWN_GRACE_SECONDS
                await self._shutdown_links(grace_seconds=grace)
                logger.info("BLEManager cleanup complete.")

    async def _manager_loop(self) -> bool:
        """Runs until stopped. Returns True if it ended because the adapter was lost."""
        discovery_down_since: float | None = None

        while not self._stop_event.is_set():
            try:
                # The scanner reports whether BlueZ discovery is actually running.
                # Probing the adapter with a `bluetoothctl` subprocess instead cost
                # a fork/exec plus a D-Bus round trip against the same bluetoothd
                # that carries the BLE links.
                if self._scanner is not None and not self._scanner.is_discovering:
                    now = time.monotonic()
                    if discovery_down_since is None:
                        discovery_down_since = now
                    elif now - discovery_down_since >= self.ADAPTER_LOST_AFTER_SECONDS:
                        logger.warning("Bluetooth adapter lost. Stopping all connections...")
                        return True
                else:
                    discovery_down_since = None

                await asyncio.sleep(self.LOOP_INTERVAL_SECONDS)

                await self._scanner_queue_routine()
                await self._connection_queue_routine()
                self._forget_finished_links()
                await self._revive_silent_scanner()

            except Exception as e:
                logger.exception(f"Error in manager loop: {e}")

        return False

    async def _revive_silent_scanner(self) -> None:
        """Restarts a scanner that hears nothing although a link waits to reconnect.

        Without advertisements no link can come back: the RSSI gate stays shut and
        BlueZ forgets the devices, so the occasional forced attempt fails with
        "device not found". is_discovering does not catch this, because a passive
        scan that stopped reporting looks exactly like one with nothing in reach.
        """
        now = time.monotonic()
        waiting = [serial for serial, link in self._links.items() if not link.is_connected]
        if self._scanner is None or not waiting:
            self._link_down_since = None
            self._reset_scanner_revival()
            return

        if self._link_down_since is None:
            self._link_down_since = now

        heard = self._scanner.seconds_since_advertisement
        if heard is not None and heard < self.SCANNER_SILENCE_SECONDS:
            self._reset_scanner_revival()  # the scanner hears Partectors, it is fine
            return

        # silent_for starts over with every restart, which paces the restarts.
        silent_for = self._scanner.silent_for
        if min(silent_for, now - self._link_down_since) < self._scanner_restart_interval:
            return

        # Every second fruitless restart tries the other scan mode.
        self._silent_scanner_restarts += 1
        switch_mode = self._silent_scanner_restarts % 2 == 0
        logger.warning(
            f"No Partector advertisement for {heard or silent_for:.0f}s while {waiting} wait to "
            f"reconnect. Restarting the scanner{' in the other scan mode' if switch_mode else ''}."
        )
        await self._scanner.restart(switch_mode=switch_mode)
        self._scanner_restart_interval = min(
            self._scanner_restart_interval * 2, self.SCANNER_RESTART_MAX_INTERVAL_SECONDS
        )

    def _reset_scanner_revival(self) -> None:
        self._scanner_restart_interval = self.SCANNER_SILENCE_SECONDS
        self._silent_scanner_restarts = 0

    async def _shutdown_links(self, grace_seconds: float) -> None:
        """Ends every connection task: first by asking, then by cancelling."""
        self._task_stop_event.set()

        pending = [link.task for link in self._links.values() if not link.task.done()]
        if pending and grace_seconds > 0:
            _, still_pending = await asyncio.wait(pending, timeout=grace_seconds)
            pending = list(still_pending)
            if pending:
                # Normal for a link that is inside a connect attempt to an
                # unreachable device: connect() can take up to its own timeout.
                logger.info(f"{len(pending)} connection task(s) still busy, cancelling.")

        for task in pending:
            task.cancel()
        if pending:
            await asyncio.wait(pending, timeout=2.0)  # let the cancellation propagate

        self._links.clear()

    async def _task_connection(self, device: BLEDevice, serial: int) -> None:
        connection = PartectorBleConnection(
            device=device,
            loop=self._loop,
            serial_number=serial,
            queue=self._queue_connection,
            rssi_provider=lambda: self._get_rssi(device.address),
            device_provider=lambda: self._get_device(device.address),
            point_listener=self._point_listener,
            p2pro_mode=self._p2pro_mode,
            p2pro_mode_guard=self._p2pro_mode_guard,
        )
        link = self._links.get(serial)
        if link is not None:
            link.connection = connection
            link.device = BlePartector(connection, self._loop)

        try:
            async with connection:
                while not self._task_stop_event.is_set():
                    await asyncio.sleep(0.5)

        except asyncio.CancelledError:
            logger.info(f"{serial}: Connection task cancelled.")
        except Exception as e:
            logger.warning(f"{serial}: Connection task failed: {e}")
        finally:
            if link is not None:
                link.connection = None
                link.device = None
            logger.info(f"{serial}: Connection task finished.")

    def _forget_finished_links(self) -> None:
        for serial, link in list(self._links.items()):
            if link.task.done():
                self._links.pop(serial, None)
                logger.info(f"{serial}: Connection task finished and popped.")

    # == Adapter ===================================================================================
    async def _wait_for_bluetooth_adapter(self) -> None:
        """Returns when an adapter is ready, or when stop() was called.

        The check may never return (macOS waits for the Bluetooth permission dialog to be
        answered), so both the check and the pause between two checks give way to stop().
        """
        while not self._stop_event.is_set():
            if await self._check_adapter_unless_stopped():
                logger.info("Bluetooth adapter is available and ready.")
                return
            if self._stop_event.is_set():
                return

            logger.info(
                "Bluetooth adapter not available. "
                f"Retrying in {self.ADAPTER_CHECK_INTERVAL_SECONDS} seconds..."
            )
            await self._sleep_unless_stopped(self.ADAPTER_CHECK_INTERVAL_SECONDS)

    async def _check_adapter_unless_stopped(self) -> bool:
        """True if the adapter is ready. False if it is not, or if stop() came first."""
        check = asyncio.ensure_future(self._adapter_available())
        started = time.monotonic()
        reported = False
        try:
            while not check.done():
                if self._stop_event.is_set():
                    return False
                await asyncio.wait({check}, timeout=self.STOP_POLL_SECONDS)
                if (
                    not reported
                    and not check.done()
                    and time.monotonic() - started >= self.ADAPTER_CHECK_SLOW_SECONDS
                ):
                    reported = True
                    logger.info(
                        "Still waiting for the Bluetooth adapter. On macOS this means the "
                        "Bluetooth permission dialog has not been answered."
                    )
            return check.result()
        finally:
            if not check.done():
                check.cancel()
                await asyncio.wait({check}, timeout=1.0)  # let it clean up, but do not hang on it

    async def _sleep_unless_stopped(self, seconds: float) -> None:
        deadline = time.monotonic() + seconds
        while not self._stop_event.is_set():
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return
            await asyncio.sleep(min(self.STOP_POLL_SECONDS, remaining))

    @staticmethod
    async def _adapter_available() -> bool:
        """Is there a powered Bluetooth adapter to start a scanner session on?"""
        if sys.platform.startswith("linux"):
            return await _bluez_adapter_powered()

        try:
            scanner = BleakScanner()
            await scanner.start()
            await scanner.stop()
            return True
        except Exception as e:
            logger.debug(f"Bluetooth adapter not available: {e}")
            return False

    # == Queue draining ============================================================================
    def _get_device(self, address: str) -> BLEDevice | None:
        """Most recently advertised BLEDevice for an address, or None."""
        if self._scanner is None:
            return None
        return self._scanner.get_device(address)

    def _get_rssi(self, address: str) -> int | None:
        """Most recent RSSI for an address, or None if it is stale / unknown."""
        if self._scanner is None:
            return None
        return self._scanner.get_rssi(address)

    def _buffer_points(self, points: list[NaneosDeviceDataPoint]) -> None:
        """Append data points to the per-device buffer, keeping the newest ones."""
        with self._points_lock:
            for point in points:
                if point.serial_number is None:
                    continue
                buffered = self._points.setdefault(point.serial_number, [])
                buffered.append(point)
                if len(buffered) > MAX_ROWS_PER_DEVICE:
                    del buffered[:-MAX_ROWS_PER_DEVICE]

    async def _scanner_queue_routine(self) -> None:
        """Drain the scanner queue and start a link for every new device that qualifies."""
        seen: dict[int, BLEDevice] = {}

        while not self._queue_scanner.empty():
            try:
                device, serial = self._queue_scanner.get_nowait()
            except asyncio.QueueEmpty:
                break
            seen[serial] = device

        for serial, device in seen.items():
            if serial in self._links:
                continue
            if not self._wants_link(serial):
                continue

            rssi = self._get_rssi(device.address)
            if rssi is not None and rssi < MIN_RSSI_CONNECT_DBM:
                logger.info(
                    f"Ignoring serial={serial} ({device.address}): RSSI {rssi} dBm is below "
                    f"{MIN_RSSI_CONNECT_DBM} dBm."
                )
                continue

            logger.info(
                f"New device detected: serial={serial}, address={device.address}, rssi={rssi}"
            )
            task = self._loop.create_task(self._task_connection(device, serial))
            self._links[serial] = BleLink(task=task)

    def _wants_link(self, serial: int) -> bool:
        """Allow-list and link cap. Logged once per device per decision, not per second."""
        if self._allowed_serials is not None and serial not in self._allowed_serials:
            return False
        if len(self._links) >= self._max_links:
            if serial not in self._rejected_for_cap:
                logger.info(f"Not connecting to serial={serial}: {self._max_links} links in use.")
                self._rejected_for_cap.add(serial)
            return False
        self._rejected_for_cap.discard(serial)
        return True

    async def _connection_queue_routine(self) -> None:
        """Drain the connection queue and record the data points in one batch."""
        batch_data: list[NaneosDeviceDataPoint] = []

        while not self._queue_connection.empty():
            try:
                batch_data.append(self._queue_connection.get_nowait())
            except asyncio.QueueEmpty:
                break

        self._buffer_points(batch_data)

get_connected_serial_numbers()

Serial numbers of the devices with a live BLE link.

Source code in src/naneos/ble/partector/manager.py

def get_connected_serial_numbers(self) -> list[int]:
    """Serial numbers of the devices with a live BLE link."""
    return [sn for sn, _ in self._live_links()]

get_data()

Returns the collected data as DataFrames and clears the buffer.

Source code in src/naneos/ble/partector/manager.py

def get_data(self) -> dict[int, pd.DataFrame]:
    """Returns the collected data as DataFrames and clears the buffer."""
    # Swap first: the BLE thread keeps appending while we convert. Only the swap
    # is locked, so the conversion never holds up the event loop.
    with self._points_lock:
        points, self._points = self._points, {}

    return {
        serial: df
        for serial, serial_points in points.items()
        if not (df := to_pandas_df(serial_points)).empty
    }

get_devices()

Handles to write to and query the devices with a live BLE link.

Source code in src/naneos/ble/partector/manager.py

def get_devices(self) -> list[PartectorDevice]:
    """Handles to write to and query the devices with a live BLE link."""
    return [link.device for _, link in self._live_links() if link.device is not None]

PartectorDevice

Bases: ABC

One Partector, whatever it is connected with.

Source code in src/naneos/device.py

class PartectorDevice(ABC):
    """One Partector, whatever it is connected with."""

    @property
    @abstractmethod
    def serial_number(self) -> int | None:
        """None until the device has told it."""

    @property
    @abstractmethod
    def device_type(self) -> DeviceType | None:
        """None until the device family is known."""

    @property
    @abstractmethod
    def firmware_version(self) -> int | None:
        """None until the device has told it."""

    @property
    @abstractmethod
    def connection_type(self) -> ConnectionType:
        pass

    @property
    @abstractmethod
    def is_connected(self) -> bool:
        pass

    @property
    @abstractmethod
    def sample_rate_hz(self) -> float | None:
        """Data points per second; 0 when the output is off, None when the
        device paces itself (P2 Pro size distribution mode)."""

    @abstractmethod
    def write(self, command: str) -> None:
        """Send a command that has no answer, for example "X0001!".

        Raises:
            ConnectionError: the device is not connected.
        """

    @abstractmethod
    def query(self, command: str, timeout: float | None = None) -> list[str]:
        """Send a command and return the tab separated fields of its answer.

        One command is in flight per device; calls from several threads queue up.

        Raises:
            ConnectionError: the device is not connected.
            TimeoutError: no answer within timeout (default: what suits the transport).
        """

    @abstractmethod
    def set_sample_rate(self, hz: int | None) -> None:
        """Set the data rate to 0 (off), 1, 10 or 100 Hz, or None for the default
        of the device: 1 Hz, or the size distribution mode of a P2 Pro, where the
        device sets the pace itself.

        The upload to naneos is limited to 1 Hz whatever is set here.

        Raises:
            NotSupportedError: over BLE, where the rate is fixed at 1 Hz (None is fine).
            ValueError: for a rate the device does not offer.
        """

    @abstractmethod
    def read_ui_curve(self, timeout: float | None = None) -> UiCurve:
        """Sweep the corona voltage and read the resulting UI curve, 100 points.

        Blocks for the sweep (10 s) plus the readout, so 15 s or more. The
        measurement is disturbed by the sweep: the data points of the device
        are held back until it has settled again.

        Over BLE a lost packet leaves the curve short: check UiCurve.is_complete.
        The manager does that and reads again on its own.

        Args:
            timeout: for the readout that follows the sweep; default 30 s over USB, 90 s over
                BLE (the device sends one packet every 2 s).

        Raises:
            NotSupportedError: a P1, or firmware older than 418.
            TimeoutError: the curve did not arrive complete within the timeout.
            ConnectionError: the device is not connected.
        """

    @abstractmethod
    def read_pulse_form(self, timeout: float | None = None) -> PulseForm:
        """Read the form of the last charging pulse, 200 samples. Does not disturb
        the measurement.

        Over BLE a lost packet leaves the form short: check PulseForm.is_complete.
        The manager does that and reads again on its own.

        Args:
            timeout: for the readout; default 30 s over USB, 90 s over BLE.

        Raises:
            NotSupportedError: a P1, or firmware older than 418.
            TimeoutError: the form did not arrive complete within the timeout.
            ConnectionError: the device is not connected.
        """

    def __repr__(self) -> str:
        kind = self.device_type.name if self.device_type is not None else "?"
        return f"<{type(self).__name__} SN{self.serial_number} {kind} {self.connection_type}>"

device_type abstractmethod property

None until the device family is known.

firmware_version abstractmethod property

None until the device has told it.

sample_rate_hz abstractmethod property

Data points per second; 0 when the output is off, None when the device paces itself (P2 Pro size distribution mode).

serial_number abstractmethod property

None until the device has told it.

query(command, timeout=None) abstractmethod

Send a command and return the tab separated fields of its answer.

One command is in flight per device; calls from several threads queue up.

Raises:

Type	Description
ConnectionError	the device is not connected.
TimeoutError	no answer within timeout (default: what suits the transport).

Source code in src/naneos/device.py

@abstractmethod
def query(self, command: str, timeout: float | None = None) -> list[str]:
    """Send a command and return the tab separated fields of its answer.

    One command is in flight per device; calls from several threads queue up.

    Raises:
        ConnectionError: the device is not connected.
        TimeoutError: no answer within timeout (default: what suits the transport).
    """

read_pulse_form(timeout=None) abstractmethod

Read the form of the last charging pulse, 200 samples. Does not disturb the measurement.

Over BLE a lost packet leaves the form short: check PulseForm.is_complete. The manager does that and reads again on its own.

Parameters:

Name	Type	Description	Default
timeout	float | None	for the readout; default 30 s over USB, 90 s over BLE.	None

Raises:

Type	Description
NotSupportedError	a P1, or firmware older than 418.
TimeoutError	the form did not arrive complete within the timeout.
ConnectionError	the device is not connected.

Source code in src/naneos/device.py

@abstractmethod
def read_pulse_form(self, timeout: float | None = None) -> PulseForm:
    """Read the form of the last charging pulse, 200 samples. Does not disturb
    the measurement.

    Over BLE a lost packet leaves the form short: check PulseForm.is_complete.
    The manager does that and reads again on its own.

    Args:
        timeout: for the readout; default 30 s over USB, 90 s over BLE.

    Raises:
        NotSupportedError: a P1, or firmware older than 418.
        TimeoutError: the form did not arrive complete within the timeout.
        ConnectionError: the device is not connected.
    """

read_ui_curve(timeout=None) abstractmethod

Sweep the corona voltage and read the resulting UI curve, 100 points.

Blocks for the sweep (10 s) plus the readout, so 15 s or more. The measurement is disturbed by the sweep: the data points of the device are held back until it has settled again.

Over BLE a lost packet leaves the curve short: check UiCurve.is_complete. The manager does that and reads again on its own.

Parameters:

Name	Type	Description	Default
timeout	float | None	for the readout that follows the sweep; default 30 s over USB, 90 s over BLE (the device sends one packet every 2 s).	None

Raises:

Type	Description
NotSupportedError	a P1, or firmware older than 418.
TimeoutError	the curve did not arrive complete within the timeout.
ConnectionError	the device is not connected.

Source code in src/naneos/device.py

@abstractmethod
def read_ui_curve(self, timeout: float | None = None) -> UiCurve:
    """Sweep the corona voltage and read the resulting UI curve, 100 points.

    Blocks for the sweep (10 s) plus the readout, so 15 s or more. The
    measurement is disturbed by the sweep: the data points of the device
    are held back until it has settled again.

    Over BLE a lost packet leaves the curve short: check UiCurve.is_complete.
    The manager does that and reads again on its own.

    Args:
        timeout: for the readout that follows the sweep; default 30 s over USB, 90 s over
            BLE (the device sends one packet every 2 s).

    Raises:
        NotSupportedError: a P1, or firmware older than 418.
        TimeoutError: the curve did not arrive complete within the timeout.
        ConnectionError: the device is not connected.
    """

set_sample_rate(hz) abstractmethod

Set the data rate to 0 (off), 1, 10 or 100 Hz, or None for the default of the device: 1 Hz, or the size distribution mode of a P2 Pro, where the device sets the pace itself.

The upload to naneos is limited to 1 Hz whatever is set here.

Raises:

Type	Description
NotSupportedError	over BLE, where the rate is fixed at 1 Hz (None is fine).
ValueError	for a rate the device does not offer.

Source code in src/naneos/device.py

@abstractmethod
def set_sample_rate(self, hz: int | None) -> None:
    """Set the data rate to 0 (off), 1, 10 or 100 Hz, or None for the default
    of the device: 1 Hz, or the size distribution mode of a P2 Pro, where the
    device sets the pace itself.

    The upload to naneos is limited to 1 Hz whatever is set here.

    Raises:
        NotSupportedError: over BLE, where the rate is fixed at 1 Hz (None is fine).
        ValueError: for a rate the device does not offer.
    """

write(command) abstractmethod

Send a command that has no answer, for example "X0001!".

Raises:

Type	Description
ConnectionError	the device is not connected.

Source code in src/naneos/device.py

@abstractmethod
def write(self, command: str) -> None:
    """Send a command that has no answer, for example "X0001!".

    Raises:
        ConnectionError: the device is not connected.
    """

PartectorSerialManager

Bases: Thread

Connects to every Partector on USB, keeps the links alive and collects their data.

Parameters:

Name	Type	Description	Default
gain_test_active	bool	run the electrometer gain test on P2 / P2 Pro. It holds the data of a device back for at least 10 s after every connect.	True
output_pulse_diagnostics	bool	let P2 / P2 Pro append the pulse diagnostics columns.	True
point_listener	PointListener | None	called with every data point as it arrives, in addition to get_data(). Runs on the reader threads: must be quick and must not block.	None
sample_rate_hz	int | None	the data rate of every device, see the property.	None

Source code in src/naneos/usb/partector/manager.py

class PartectorSerialManager(threading.Thread):
    """Connects to every Partector on USB, keeps the links alive and collects their data.

    Args:
        gain_test_active: run the electrometer gain test on P2 / P2 Pro. It holds
            the data of a device back for at least 10 s after every connect.
        output_pulse_diagnostics: let P2 / P2 Pro append the pulse diagnostics columns.
        point_listener: called with every data point as it arrives, in addition to
            get_data(). Runs on the reader threads: must be quick and must not block.
        sample_rate_hz: the data rate of every device, see the property.
    """

    def __init__(
        self,
        gain_test_active: bool = True,
        output_pulse_diagnostics: bool = True,
        point_listener: PointListener | None = None,
        sample_rate_hz: int | None = None,
    ) -> None:
        super().__init__(daemon=True)
        self._point_listener = point_listener
        self._gain_test_active = gain_test_active
        self._output_pulse_diagnostics = output_pulse_diagnostics
        self._sample_rate_hz = self.check_sample_rate(sample_rate_hz)
        self._sample_rate_changed = threading.Event()
        self._stop_event = threading.Event()

        # Written by the manager thread in _fetch_data(), handed over in get_data().
        self._data: dict[int, pd.DataFrame] = {}
        self._data_lock = threading.Lock()

        self._devices: dict[str, UsbPartector] = {}  # key: port

    def get_data(self) -> dict[int, pd.DataFrame]:
        """Returns the data the manager loop collected since the last call.

        The devices are read by the manager thread only (see _fetch_data), so
        this can be called from any thread.
        """
        with self._data_lock:
            data, self._data = self._data, {}
        return data

    def stop(self) -> None:
        self._stop_event.set()

    @property
    def sample_rate_hz(self) -> int | None:
        """The data rate of every device: 1, 10 or 100 Hz, or None for the default
        of each device (1 Hz, size distribution mode on a P2 Pro). Setting it
        changes the connected devices within a second and applies to the ones
        that connect later.
        """
        return self._sample_rate_hz

    @sample_rate_hz.setter
    def sample_rate_hz(self, hz: int | None) -> None:
        self._sample_rate_hz = self.check_sample_rate(hz)
        self._sample_rate_changed.set()  # applied by the manager thread

    @staticmethod
    def check_sample_rate(hz: int | None) -> int | None:
        # The rates a device takes, without 0: that switches one device off.
        rates = sorted(rate for rate in UsbPartector.SAMPLE_RATE_CODES if rate)
        if hz is not None and hz not in rates:
            raise ValueError(f"Sample rate must be one of {rates} Hz, or None for the default.")
        return hz

    def run(self) -> None:
        try:
            self._manager_loop()
        except RuntimeError as e:
            logger.exception(f"SerialManager loop exited with: {e}")

    def get_settling_serial_numbers(self) -> list[int | None]:
        """Serial numbers of devices still warming up after a gain test was started."""
        return [d.serial_number for d in self._all_devices() if d.is_settling]

    def get_connected_serial_numbers(self) -> list[int | None]:
        return [d.serial_number for d in self._all_devices()]

    def get_devices(self) -> list[PartectorDevice]:
        """Handles to write to and query the connected devices and to set their rate."""
        return list(self._all_devices())

    def _all_devices(self) -> list[UsbPartector]:
        """Snapshot of all connected devices, safe to iterate from any thread."""
        return list(self._devices.values())

    def _manager_loop(self) -> None:
        while not self._stop_event.is_set():
            try:
                found = scan_serial_ports(ports_exclude=list(self._devices))

                self._disconnect_unplugged_ports()
                self._apply_sample_rate()  # before the connect: a new device gets the rate itself
                self._connect_to_new_ports(found)

                self._fetch_data()

                time.sleep(1.0)  # Sleep to avoid busy waiting

            except Exception as e:
                logger.exception(f"Error in serial manager loop: {e}")

        self._fetch_data()  # do not lose the last second of data
        self._close_all_ports()

    def _fetch_data(self) -> None:
        """Reads every connected device once. Called from the manager thread only.

        UsbPartector.get_data() is not safe to call concurrently, so
        this must stay the single consumer of the device queues.
        """
        points: list[NaneosDeviceDataPoint] = []
        for device in self._all_devices():
            points.extend(device.get_data())

        if not points:
            return

        with self._data_lock:
            self._data = add_data_points_to_dict(self._data, points)

    def _apply_sample_rate(self) -> None:
        if not self._sample_rate_changed.is_set():
            return
        self._sample_rate_changed.clear()
        for device in self._all_devices():
            try:
                device.set_sample_rate(self._sample_rate_hz)
            except (ConnectionError, NotSupportedError) as e:
                logger.warning(f"SN{device.serial_number}: could not set the sample rate: {e}")

    def _disconnect_unplugged_ports(self) -> None:
        for port, device in list(self._devices.items()):
            if not device.is_connected:
                logger.info(f"Disconnecting SN{device.serial_number} on {port}")
                device.close()
                self._devices.pop(port, None)

    def _connect_to_new_ports(self, found: list[FoundDevice]) -> None:
        for device in found:
            try:
                self._devices[device.port] = self._connect(device)
            except (ConnectionError, TimeoutError) as e:
                # Found again by the next scan.
                logger.warning(f"Could not connect to SN{device.serial_number}: {e}")

    def _connect(self, found: FoundDevice) -> UsbPartector:
        cls = DEVICE_CLASSES[found.kind]
        if issubclass(cls, Partector2Family):
            return cls(
                port=found.port,
                sample_rate_hz=self._sample_rate_hz,
                gain_test_active=self._gain_test_active,
                output_pulse_diagnostics=self._output_pulse_diagnostics,
                point_listener=self._point_listener,
            )
        return cls(
            port=found.port,
            sample_rate_hz=self._sample_rate_hz,
            point_listener=self._point_listener,
        )

    def _close_all_ports(self) -> None:
        for port, device in list(self._devices.items()):
            device.close()
            self._devices.pop(port, None)

sample_rate_hz property writable

The data rate of every device: 1, 10 or 100 Hz, or None for the default of each device (1 Hz, size distribution mode on a P2 Pro). Setting it changes the connected devices within a second and applies to the ones that connect later.

get_data()

Returns the data the manager loop collected since the last call.

The devices are read by the manager thread only (see _fetch_data), so this can be called from any thread.

Source code in src/naneos/usb/partector/manager.py

def get_data(self) -> dict[int, pd.DataFrame]:
    """Returns the data the manager loop collected since the last call.

    The devices are read by the manager thread only (see _fetch_data), so
    this can be called from any thread.
    """
    with self._data_lock:
        data, self._data = self._data, {}
    return data

get_devices()

Handles to write to and query the connected devices and to set their rate.

Source code in src/naneos/usb/partector/manager.py

def get_devices(self) -> list[PartectorDevice]:
    """Handles to write to and query the connected devices and to set their rate."""
    return list(self._all_devices())

get_settling_serial_numbers()

Serial numbers of devices still warming up after a gain test was started.

Source code in src/naneos/usb/partector/manager.py

def get_settling_serial_numbers(self) -> list[int | None]:
    """Serial numbers of devices still warming up after a gain test was started."""
    return [d.serial_number for d in self._all_devices() if d.is_settling]

PulseForm dataclass

The electrometer current along one charging pulse, 200 samples in time order.

Source code in src/naneos/diagnostics.py

@dataclass(frozen=True)
class PulseForm:
    """The electrometer current along one charging pulse, 200 samples in time order."""

    EXPECTED_ENTRIES: ClassVar[str] = f"{PULSE_FORM_VALUES} I values"

    device_type: DeviceType
    serial_number: int
    unix_timestamp: int  # seconds, when the form was read
    currents: tuple[float, ...]  # nA

    @property
    def is_complete(self) -> bool:
        return len(self.currents) == PULSE_FORM_VALUES

    @property
    def entries(self) -> str:
        """What the form holds, for the log: "200 I values"."""
        return f"{len(self.currents)} I values"

entries property

What the form holds, for the log: "200 I values".

UiCurve dataclass

The electrometer current over the corona voltage, 100 points sorted by voltage.

Source code in src/naneos/diagnostics.py

@dataclass(frozen=True)
class UiCurve:
    """The electrometer current over the corona voltage, 100 points sorted by voltage."""

    EXPECTED_ENTRIES: ClassVar[str] = f"{UI_CURVE_POINTS} U + {UI_CURVE_POINTS} I values"

    device_type: DeviceType
    serial_number: int
    unix_timestamp: int  # seconds, when the curve was read
    voltages: tuple[int, ...]  # V
    currents: tuple[float, ...]  # nA

    @property
    def is_complete(self) -> bool:
        return len(self.voltages) == UI_CURVE_POINTS and len(self.currents) == UI_CURVE_POINTS

    @property
    def entries(self) -> str:
        """What the curve holds, for the log: "100 U + 100 I values"."""
        return f"{len(self.voltages)} U + {len(self.currents)} I values"

entries property

What the curve holds, for the log: "100 U + 100 I values".

enable_console_logging(level=logging.INFO, colored=True)

Print naneos log messages of at least level to stderr.

Calling it again replaces the previous console handler, so the output is never duplicated. Returns the "naneos" logger.

Source code in src/naneos/logger.py

def enable_console_logging(level: int = logging.INFO, colored: bool = True) -> logging.Logger:
    """Print naneos log messages of at least `level` to stderr.

    Calling it again replaces the previous console handler, so the output is
    never duplicated. Returns the "naneos" logger.
    """
    root = logging.getLogger(ROOT_LOGGER_NAME)
    for handler in list(root.handlers):
        if isinstance(handler, _NaneosConsoleHandler):
            root.removeHandler(handler)

    handler = _NaneosConsoleHandler()
    handler.setLevel(level)
    handler.setFormatter(CustomFormatter(terminal=colored))
    root.addHandler(handler)
    _lower_level_to(root, level)
    return root

enable_file_logging(path, level=logging.INFO, max_bytes=0, backup_count=0)

Append naneos log messages of at least level to a file.

path may be a directory, in which case naneos-devices.log is created in it. Calling it again replaces the previous file handler. Returns the "naneos" logger.

The file grows without limit unless max_bytes is set: then it is moved to naneos-devices.log.1 (and so on, up to backup_count files) once it would exceed that size. A long-running application should set both.

Source code in src/naneos/logger.py

def enable_file_logging(
    path: str | Path,
    level: int = logging.INFO,
    max_bytes: int = 0,
    backup_count: int = 0,
) -> logging.Logger:
    """Append naneos log messages of at least `level` to a file.

    `path` may be a directory, in which case naneos-devices.log is created in
    it. Calling it again replaces the previous file handler. Returns the
    "naneos" logger.

    The file grows without limit unless `max_bytes` is set: then it is moved to
    naneos-devices.log.1 (and so on, up to `backup_count` files) once it would
    exceed that size. A long-running application should set both.
    """
    path = Path(path).resolve()
    if path.is_dir():
        path = path / DEFAULT_LOG_FILE_NAME
    path.parent.mkdir(parents=True, exist_ok=True)

    root = logging.getLogger(ROOT_LOGGER_NAME)
    for handler in list(root.handlers):
        if isinstance(handler, _NaneosFileHandler):
            handler.close()
            root.removeHandler(handler)

    handler = _NaneosFileHandler(
        str(path), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"
    )
    handler.setLevel(level)
    handler.setFormatter(CustomFormatter(terminal=False))
    root.addHandler(handler)
    _lower_level_to(root, level)
    return root

upload_snapshot(data)

Upload the frames of a snapshot, keyed by device serial number.

Blocks for up to TIMEOUT_SECONDS. Raises requests.RequestException on network problems; HTTP errors are reported by the returned response.

Source code in src/naneos/cloud/upload.py

def upload_snapshot(data: dict[int, pd.DataFrame]) -> requests.Response:
    """Upload the frames of a snapshot, keyed by device serial number.

    Blocks for up to TIMEOUT_SECONDS. Raises requests.RequestException on
    network problems; HTTP errors are reported by the returned response.
    """
    return send_frames(prepare_frames(data))