auto_detection

korobka.utils.auto_detection

Auto-detects a port to which the device is connected.

At the moment, it supports only serial ports.

Functions

get_serial_port(test_command, test_response=None, baudrate=1200, bytesize=serial.EIGHTBITS, parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE, timeout=1, response_size=256)

Scans all available serial ports and returns the one that responds to a specific test command.

:param test_command: The binary command to send. :param test_response: Optional expected substring in the response to confirm a match. :param baudrate: Baud rate to use while testing ports. :param bytesize: Data bits. :param parity: Parity bit setting. :param stopbits: Stop bits. :param timeout: Timeout in seconds. :param response_size: Number of bytes to read from response. :return: The detected serial port name (e.g., 'COM3' or '/dev/ttyUSB0'). :raises: RuntimeError if no matching device is found.

Source code in korobka/utils/auto_detection.py

def get_serial_port(
    test_command: bytes | str,
    test_response: str | None = None,
    baudrate: int = 1200,
    bytesize: int = serial.EIGHTBITS,
    parity: str = serial.PARITY_EVEN,
    stopbits: int = serial.STOPBITS_ONE,
    timeout: int = 1,
    response_size: int = 256,
) -> str:
    """Scans all available serial ports and returns the one that responds to a specific test command.

    :param test_command: The binary command to send.
    :param test_response: Optional expected substring in the response to confirm a match.
    :param baudrate: Baud rate to use while testing ports.
    :param bytesize: Data bits.
    :param parity: Parity bit setting.
    :param stopbits: Stop bits.
    :param timeout: Timeout in seconds.
    :param response_size: Number of bytes to read from response.
    :return: The detected serial port name (e.g., 'COM3' or '/dev/ttyUSB0').
    :raises: RuntimeError if no matching device is found.
    """
    ports = list_ports.comports()
    logger.info("Scanning %d ports for a match...", len(ports))

    for port_info in ports:
        port_name = port_info.device
        try:
            with serial.Serial(
                port=port_name,
                baudrate=baudrate,
                bytesize=bytesize,
                parity=parity,
                stopbits=stopbits,
                timeout=timeout,
            ) as ser:
                ser.reset_input_buffer()
                ser.reset_output_buffer()

                if isinstance(test_command, str):
                    ser.write(f"{test_command}\r\n".encode("ascii"))
                else:
                    ser.write(test_command)

                time.sleep(0.5)
                response = ser.read(response_size).decode(errors="ignore")
                logger.info("Testing port %s: response = %s", port_name, response)

                if test_response is None or test_response in response:
                    logger.info("Device matched on port %s", port_name)
                    return port_name
        except Exception as e:
            logger.error("Failed to test port %s: %s", port_name, str(e))

    raise RuntimeError("No matching serial device found.")