Skip to content

naneos.usb.partector.scan

Find Partectors on the serial ports of this machine.

list_serial_ports(ports_exclude=None)

Serial ports that look like a Partector and can be opened, excluding ports_exclude.

Source code in src/naneos/usb/partector/scan.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
def list_serial_ports(ports_exclude: list[str] | None = None) -> list[str]:
    """Serial ports that look like a Partector and can be opened, excluding ports_exclude."""
    exclude = ports_exclude or []
    candidates = [
        port.device
        for port in list_ports.comports()
        if port.device not in exclude
        and (
            (port.pid == _PARTECTOR_PID and port.vid == _PARTECTOR_VID)
            or (port.serial_number and "dosemet" in port.serial_number.lower())
        )
    ]
    return [port for port in candidates if _can_open(port)]

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 device family.

Source code in src/naneos/usb/partector/scan.py 
58
59
60
61
62
63
64
65
66
67
def scan_for_serial_partector(serial_number: int, kind: DeviceType | 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 device family.
    """
    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_serial_ports(ports_exclude=None)

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

Source code in src/naneos/usb/partector/scan.py 
45
46
47
48
49
50
51
52
53
54
55
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