circuitpython/shared-bindings/usb/core/__init__.c

// This file is part of the CircuitPython project: https://circuitpython.org
//
// SPDX-FileCopyrightText: Copyright (c) 2022 Scott Shawcroft for Adafruit Industries
//
// SPDX-License-Identifier: MIT

#include <stdarg.h>
#include <string.h>

#include "py/obj.h"
#include "py/objexcept.h"
#include "py/misc.h"
#include "py/mphal.h"
#include "py/runtime.h"

#include "shared-bindings/usb/core/__init__.h"
#include "shared-bindings/usb/core/Device.h"

//| """USB Core
//|
//| This is a subset of the PyUSB core module.
//| """
//|

//| class USBError(OSError):
//|     """Catchall exception for USB related errors."""
//|
//|     ...
//|
MP_DEFINE_USB_CORE_EXCEPTION(USBError, OSError)
NORETURN void mp_raise_usb_core_USBError(mp_rom_error_text_t fmt, ...) {
    mp_obj_t exception;
    if (fmt == NULL) {
        exception = mp_obj_new_exception(&mp_type_usb_core_USBError);
    } else {
        va_list argptr;
        va_start(argptr, fmt);
        exception = mp_obj_new_exception_msg_vlist(&mp_type_usb_core_USBError, fmt, argptr);
        va_end(argptr);
    }
    nlr_raise(exception);
}

//| class USBTimeoutError(USBError):
//|     """Raised when a USB transfer times out."""
//|
//|     ...
//|
MP_DEFINE_USB_CORE_EXCEPTION(USBTimeoutError, usb_core_USBError)
NORETURN void mp_raise_usb_core_USBTimeoutError(void) {
    mp_raise_type(&mp_type_usb_core_USBTimeoutError);
}


//| def find(
//|     find_all: bool = False, *, idVendor: Optional[int] = None, idProduct: Optional[int] = None
//| ) -> Device:
//|     """Find the first device that matches the given requirements or, if
//|     find_all is True, return a generator of all matching devices.
//|
//|     Returns None if no device matches.
//|     """
//|
typedef struct {
    mp_obj_base_t base;
    mp_int_t vid;
    mp_int_t pid;
    mp_int_t next_index;
} usb_core_devices_obj_t;

// This is an internal iterator type to use with find.
static mp_obj_t _next_device(usb_core_devices_obj_t *iter) {
    // Brute force check all possible device numbers for one that matches.
    usb_core_device_obj_t temp_device;
    for (size_t i = iter->next_index; i < 256; i++) {
        if (!common_hal_usb_core_device_construct(&temp_device, i)) {
            continue;
        }
        if (iter->vid < 0x10000 && iter->vid != common_hal_usb_core_device_get_idVendor(&temp_device)) {
            continue;
        }
        if (iter->pid < 0x10000 && iter->pid != common_hal_usb_core_device_get_idProduct(&temp_device)) {
            continue;
        }

        // We passed the filters. Now make a properly allocated object to
        // return to the user.
        usb_core_device_obj_t *self = mp_obj_malloc(usb_core_device_obj_t, &usb_core_device_type);

        common_hal_usb_core_device_construct(self, i);
        iter->next_index = i + 1;
        return MP_OBJ_FROM_PTR(self);
    }
    // Iter is done.
    iter->next_index = 256;
    return mp_const_none;
}

static mp_obj_t usb_core_devices_iternext(mp_obj_t self_in) {
    mp_check_self(mp_obj_is_type(self_in, &usb_core_devices_type));
    usb_core_devices_obj_t *self = MP_OBJ_TO_PTR(self_in);
    mp_obj_t device = _next_device(self);
    if (device != mp_const_none) {
        return device;
    }
    return MP_OBJ_STOP_ITERATION;
}

MP_DEFINE_CONST_OBJ_TYPE(
    usb_core_devices_type,
    MP_QSTR_USBDevices,
    MP_TYPE_FLAG_ITER_IS_ITERNEXT,
    iter, usb_core_devices_iternext
    );

static mp_obj_t usb_core_find(size_t n_args, const mp_obj_t *pos_args, mp_map_t *kw_args) {
    enum { ARG_find_all, ARG_idVendor, ARG_idProduct };
    static const mp_arg_t allowed_args[] = {
        { MP_QSTR_find_all, MP_ARG_BOOL, {.u_bool = false} },
        { MP_QSTR_idVendor, MP_ARG_OBJ, {.u_obj = mp_const_none} },
        { MP_QSTR_idProduct, MP_ARG_OBJ, {.u_obj = mp_const_none} },
    };
    mp_arg_val_t args[MP_ARRAY_SIZE(allowed_args)];
    mp_arg_parse_all(n_args, pos_args, kw_args, MP_ARRAY_SIZE(allowed_args), allowed_args, args);

    const bool find_all = args[ARG_find_all].u_bool;
    usb_core_devices_obj_t temp_iter;
    temp_iter.base.type = &usb_core_devices_type;
    temp_iter.next_index = 1;
    if (!mp_obj_get_int_maybe(args[ARG_idVendor].u_obj, &temp_iter.vid)) {
        temp_iter.vid = 0x10000;
    }
    if (!mp_obj_get_int_maybe(args[ARG_idProduct].u_obj, &temp_iter.pid)) {
        temp_iter.pid = 0x10000;
    }
    if (find_all) {
        // Copy the temp iter contents to a heap object before we return it.
        // We could do this up front but GCC falsely detects that we may return
        // the stack copy.
        usb_core_devices_obj_t *iter = m_new_obj(usb_core_devices_obj_t);
        memcpy(iter, &temp_iter, sizeof(usb_core_devices_obj_t));
        return MP_OBJ_FROM_PTR(iter);
    }
    return _next_device(&temp_iter);
}
MP_DEFINE_CONST_FUN_OBJ_KW(usb_core_find_obj, 0, usb_core_find);


static mp_rom_map_elem_t usb_core_module_globals_table[] = {
    { MP_ROM_QSTR(MP_QSTR___name__),        MP_OBJ_NEW_QSTR(MP_QSTR_usb_dot_core) },
    // Functions
    { MP_ROM_QSTR(MP_QSTR_find),            MP_OBJ_FROM_PTR(&usb_core_find_obj) },

    // Classes
    { MP_ROM_QSTR(MP_QSTR_Device),          MP_OBJ_FROM_PTR(&usb_core_device_type) },

    // Errors
    { MP_ROM_QSTR(MP_QSTR_USBError),        MP_OBJ_FROM_PTR(&mp_type_usb_core_USBError) },
    { MP_ROM_QSTR(MP_QSTR_USBTimeoutError), MP_OBJ_FROM_PTR(&mp_type_usb_core_USBTimeoutError) },
};

static MP_DEFINE_CONST_DICT(usb_core_module_globals, usb_core_module_globals_table);

void usb_core_exception_print(const mp_print_t *print, mp_obj_t o_in, mp_print_kind_t kind) {
    mp_print_kind_t k = kind & ~PRINT_EXC_SUBCLASS;
    bool is_subclass = kind & PRINT_EXC_SUBCLASS;
    if (!is_subclass && (k == PRINT_EXC)) {
        mp_print_str(print, qstr_str(MP_OBJ_QSTR_VALUE(usb_core_module_globals_table[0].value)));
        mp_print_str(print, ".");
    }
    mp_obj_exception_print(print, o_in, kind);
}

const mp_obj_module_t usb_core_module = {
    .base = { &mp_type_module },
    .globals = (mp_obj_dict_t *)&usb_core_module_globals,
};

MP_REGISTER_MODULE(MP_QSTR_usb_dot_core, usb_core_module);