Skip to content

naneos.partector.partector_serial_manager

PartectorSerialManager

Bases: Thread

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

Source code in src/naneos/partector/partector_serial_manager.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class PartectorSerialManager(threading.Thread):
    """Connects to every Partector on USB, keeps the links alive and collects their data."""

    def __init__(self) -> None:
        super().__init__(daemon=True)
        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, PartectorBlueprint] = {}  # 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()

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

    def get_connected_device_strings(self) -> list[str]:
        """Human readable list of connected devices, grouped P1, P2, P2 Pro."""
        devices = self._all_devices()
        strings = []
        for kind, label in DEVICE_LABELS.items():
            strings += [f"SN{d._sn} ({label})" for d in devices if d.device_type == kind]
        return strings

    def get_gain_test_activating_devices(self) -> list[int | None]:
        """Serial numbers of devices still warming up after a gain test was started."""
        now = time.time()
        return [d._sn for d in self._all_devices() if d._wait_with_data_output_until > now]

    def get_connected_addresses(self) -> list[str]:
        return list(self._devices.keys())

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

    def _all_devices(self) -> list[PartectorBlueprint]:
        """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=self.get_connected_addresses())

                self._disconnect_unplugged_ports()
                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.

        PartectorBlueprint.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 _disconnect_unplugged_ports(self) -> None:
        for port, device in list(self._devices.items()):
            if not device._connected:
                logger.info(f"Disconnecting SN{device._sn} on {port}")
                device.close()
                self._devices.pop(port, None)

    def _connect_to_new_ports(self, found: list[FoundDevice]) -> None:
        for device in found:
            self._devices[device.port] = DEVICE_CLASSES[device.kind](port=device.port)

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

get_connected_device_strings()

Human readable list of connected devices, grouped P1, P2, P2 Pro.

Source code in src/naneos/partector/partector_serial_manager.py 
62
63
64
65
66
67
68
def get_connected_device_strings(self) -> list[str]:
    """Human readable list of connected devices, grouped P1, P2, P2 Pro."""
    devices = self._all_devices()
    strings = []
    for kind, label in DEVICE_LABELS.items():
        strings += [f"SN{d._sn} ({label})" for d in devices if d.device_type == kind]
    return strings

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/partector/partector_serial_manager.py 
43
44
45
46
47
48
49
50
51
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_gain_test_activating_devices()

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

Source code in src/naneos/partector/partector_serial_manager.py 
70
71
72
73
def get_gain_test_activating_devices(self) -> list[int | None]:
    """Serial numbers of devices still warming up after a gain test was started."""
    now = time.time()
    return [d._sn for d in self._all_devices() if d._wait_with_data_output_until > now]