Bases: PartectorDevice
A Partector on a BLE link, usable from any thread but the BLE manager's.
The link lives on the event loop of PartectorBleManager; every call is
handed over to that loop. The handle stays valid while the manager keeps
the link, also across reconnects: is_connected tells if it is up right now.
Source code in src/naneos/ble/partector/device.py
17
18
19
20
21
22
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 | class BlePartector(PartectorDevice):
"""A Partector on a BLE link, usable from any thread but the BLE manager's.
The link lives on the event loop of PartectorBleManager; every call is
handed over to that loop. The handle stays valid while the manager keeps
the link, also across reconnects: is_connected tells if it is up right now.
"""
# On top of the command's own timeout: waiting for the command in flight
# and for the loop to pick the call up.
HANDOVER_TIMEOUT_SECONDS = 5.0
def __init__(self, connection: PartectorBleConnection, loop: asyncio.AbstractEventLoop) -> None:
self._connection = connection
self._loop = loop
self._loop_thread = threading.get_ident() # created on the loop's thread
@property
def serial_number(self) -> int:
return self._connection.SERIAL_NUMBER
@property
def device_type(self) -> DeviceType | None:
return self._connection.device_type
@property
def firmware_version(self) -> int | None:
return self._connection.firmware_version
@property
def connection_type(self) -> ConnectionType:
return ConnectionType.CONNECTED
@property
def is_connected(self) -> bool:
return self._connection.is_connected
@property
def sample_rate_hz(self) -> float:
return 1
def write(self, command: str) -> None:
self._run(self._connection.write(command), BleCommandChannel.QUERY_TIMEOUT_SECONDS)
def query(self, command: str, timeout: float | None = None) -> list[str]:
timeout = timeout or BleCommandChannel.QUERY_TIMEOUT_SECONDS
return self._run(self._connection.query(command, timeout), timeout)
def set_sample_rate(self, hz: int | None) -> None:
if hz is None:
return # 1 Hz is the default over BLE
raise NotSupportedError(
"The data rate is fixed at 1 Hz over BLE; it can only be changed over USB."
)
def read_ui_curve(self, timeout: float | None = None) -> UiCurve:
"""See PartectorDevice. Over BLE the readout takes about 40 s, default timeout 90 s."""
timeout = timeout or BLE_READOUT_TIMEOUT_SECONDS
return self._run(self._connection.read_ui_curve(timeout), UI_COMPUTE_SECONDS + timeout)
def read_pulse_form(self, timeout: float | None = None) -> PulseForm:
"""See PartectorDevice. Over BLE the readout takes about 50 s, default timeout 90 s."""
timeout = timeout or BLE_READOUT_TIMEOUT_SECONDS
return self._run(self._connection.read_pulse_form(timeout), timeout)
def _run(self, coroutine: Coroutine[Any, Any, T], timeout: float) -> T:
if threading.get_ident() == self._loop_thread:
coroutine.close()
raise RuntimeError("A BlePartector cannot be used from the BLE manager's own thread.")
if self._loop.is_closed():
coroutine.close()
raise ConnectionError(f"SN{self.serial_number}: the BLE manager has stopped.")
future = asyncio.run_coroutine_threadsafe(coroutine, self._loop)
try:
return future.result(timeout + self.HANDOVER_TIMEOUT_SECONDS)
except TimeoutError:
future.cancel()
raise
|
See PartectorDevice. Over BLE the readout takes about 50 s, default timeout 90 s.
Source code in src/naneos/ble/partector/device.py
| def read_pulse_form(self, timeout: float | None = None) -> PulseForm:
"""See PartectorDevice. Over BLE the readout takes about 50 s, default timeout 90 s."""
timeout = timeout or BLE_READOUT_TIMEOUT_SECONDS
return self._run(self._connection.read_pulse_form(timeout), timeout)
|
read_ui_curve(timeout=None)
See PartectorDevice. Over BLE the readout takes about 40 s, default timeout 90 s.
Source code in src/naneos/ble/partector/device.py
| def read_ui_curve(self, timeout: float | None = None) -> UiCurve:
"""See PartectorDevice. Over BLE the readout takes about 40 s, default timeout 90 s."""
timeout = timeout or BLE_READOUT_TIMEOUT_SECONDS
return self._run(self._connection.read_ui_curve(timeout), UI_COMPUTE_SECONDS + timeout)
|