Skip to content

naneos.device

The device API shared by every transport.

A Partector on USB and a Partector on a BLE link are used the same way:

device.write("X0001!")            # a command without an answer
fields = device.query("f?")       # a command with an answer -> ["422"]
device.set_sample_rate(10)        # USB only, see NotSupportedError
device.set_sample_rate(None)      # back to the default of the device
curve = device.read_ui_curve()    # the two diagnostics, firmware 418 or newer
form = device.read_pulse_form()

NotSupportedError

Bases: Exception

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

Source code in src/naneos/device.py
19
20
class NotSupportedError(Exception):
    """The device or the transport it is reached over cannot do this."""

PartectorDevice

Bases: ABC

One Partector, whatever it is connected with.

Source code in src/naneos/device.py
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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
65
66
67
68
69
70
71
72
73
74
@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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@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
76
77
78
79
80
81
82
83
84
85
86
87
@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
57
58
59
60
61
62
63
@abstractmethod
def write(self, command: str) -> None:
    """Send a command that has no answer, for example "X0001!".

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