Skip to content

naneos.partector.scan

Find Partectors on the serial ports of this machine.

ScanPartector

Bases: PartectorBlueprint

Minimal device used to identify what is behind a port. Never streams data.

Source code in src/naneos/partector/scan.py
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
class ScanPartector(PartectorBlueprint):
    """Minimal device used to identify what is behind a port. Never streams data."""

    def _init_print_connection_info(self) -> None:
        pass

    def _init_serial_data_structure(self) -> None:
        """Not used by the scan partector, but mandatory in the partector blueprint."""

    def _serial_wrapper(self, func) -> Any | None:
        """Like the blueprint, but raises instead of logging: a port without a
        Partector behind it must not produce warnings."""
        if not self._connected:
            return None

        excep = "Was not able to fetch the serial number!"

        for _ in range(self.SERIAL_RETRIES):
            try:
                return func()
            except Exception as e:
                excep = f"SN{self._sn} Exception occurred during user function call: {e}"

        raise Exception(excep)

    def _init_get_device_info(self) -> None:
        try:
            if self._sn is None:
                self._sn = self._get_serial_number_secure()
            self._fw = self.get_firmware_version()
            logger.debug(f"Connected to SN{self._sn} on {self._port}")
        except Exception:
            # Every port is scanned, so most of them simply have no Partector.
            pass

    def _set_verbose_freq(self, freq: int = 0) -> None:
        """Only ever used to silence the device while it is identified."""
        self._write_line("X0000!")

scan_for_serial_partector(serial_number, kind=None)

Port of the device with this serial number, or None if it is not plugged in.

kind restricts the search to one family; it accepts a DeviceType or one of the names "P1", "P2", "P2pro".

Source code in src/naneos/partector/scan.py
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def scan_for_serial_partector(
    serial_number: int, kind: DeviceType | str | None = None
) -> str | None:
    """Port of the device with this serial number, or None if it is not plugged in.

    kind restricts the search to one family; it accepts a DeviceType or one of
    the names "P1", "P2", "P2pro".
    """
    if isinstance(kind, str):
        by_name = {name: device_type for device_type, name in DEVICE_KIND_NAMES.items()}
        kind = by_name.get(kind)

    for device in scan_serial_ports():
        if device.serial_number == serial_number and (kind is None or device.kind == kind):
            return device.port

    return None

scan_for_serial_partectors(ports_exclude=None)

Found devices grouped by family: {"P1": {sn: port}, "P2": {...}, "P2pro": {...}}.

Source code in src/naneos/partector/scan.py
88
89
90
91
92
93
def scan_for_serial_partectors(ports_exclude: list[str] | None = None) -> dict[str, dict[int, str]]:
    """Found devices grouped by family: {"P1": {sn: port}, "P2": {...}, "P2pro": {...}}."""
    grouped: dict[str, dict[int, str]] = {name: {} for name in DEVICE_KIND_NAMES.values()}
    for device in scan_serial_ports(ports_exclude):
        grouped[DEVICE_KIND_NAMES[device.kind]][device.serial_number] = device.port
    return grouped

scan_serial_ports(ports_exclude=None)

Identify the Partector behind every candidate serial port, in parallel.

Source code in src/naneos/partector/scan.py
75
76
77
78
79
80
81
82
83
84
85
def scan_serial_ports(ports_exclude: list[str] | None = None) -> list[FoundDevice]:
    """Identify the Partector behind every candidate serial port, in parallel."""
    ports = list_serial_ports(ports_exclude=ports_exclude or [])
    if not ports:
        return []

    with ThreadPoolExecutor(max_workers=len(ports)) as pool:
        found = [device for device in pool.map(_scan_port, ports) if device is not None]

    logger.debug(f"Found devices: {found}")
    return found