micropython: Add display data channel module.

This commit is contained in:
iabdalkader 2023-09-24 16:26:50 +02:00
parent 43a4e5519e
commit 9471f94cd1

View File

@ -0,0 +1,263 @@
/*
* This file is part of the OpenMV project.
*
* Copyright (c) 2013-2021 Ibrahim Abdelkader <iabdalkader@openmv.io>
* Copyright (c) 2013-2021 Kwabena W. Agyeman <kwagyeman@openmv.io>
*
* This work is licensed under the MIT license, see the file LICENSE for details.
*
* Display data channel module (such as CEC/DDC).
*/
#include "omv_boardconfig.h"
#if OMV_DISPLAY_CEC_ENABLE || OMV_DISPLAY_DDC_ENABLE
#include "py/obj.h"
#include "py/objarray.h"
#include "py/runtime.h"
#include "mphal.h"
#include "extmod/machine_i2c.h"
#include "omv_gpio.h"
#include "fb_alloc.h"
#include "py_display.h"
#include "cec.h"
typedef struct _py_display_data_obj_t {
mp_obj_base_t base;
#if OMV_DISPLAY_CEC_ENABLE
bool cec_enabled;
uint8_t dst_addr;
mp_obj_t cec_frame;
mp_obj_t cec_callback;
#endif
#if OMV_DISPLAY_DDC_ENABLE
bool ddc_enabled;
uint8_t ddc_addr;
mp_obj_t ddc_bus;
#endif
} py_display_data_obj_t;
#if OMV_DISPLAY_CEC_ENABLE
static void cec_extint_callback(mp_obj_t self_in) {
py_display_data_obj_t *self = MP_OBJ_TO_PTR(self_in);
if (self->cec_callback != mp_const_none) {
uint8_t src_addr = 0;
mp_obj_array_t *frame = MP_OBJ_TO_PTR(self->cec_frame);
if (cec_receive_frame(frame->items, &frame->len, self->dst_addr, &src_addr) != -1) {
mp_call_function_2(self->cec_callback, MP_OBJ_NEW_SMALL_INT(src_addr), self->cec_frame);
}
}
}
STATIC mp_obj_t py_cec_send_frame(uint n_args, const mp_obj_t *pos_args, mp_map_t *kw_args) {
enum { ARG_dst_addr, ARG_src_addr, ARG_data };
static const mp_arg_t allowed_args[] = {
{ MP_QSTR_dst_addr, MP_ARG_REQUIRED | MP_ARG_INT },
{ MP_QSTR_src_addr, MP_ARG_REQUIRED | MP_ARG_INT },
{ MP_QSTR_data, MP_ARG_REQUIRED | MP_ARG_OBJ },
};
// Parse args.
py_display_data_obj_t *self = MP_OBJ_TO_PTR(pos_args[0]);
mp_arg_val_t args[MP_ARRAY_SIZE(allowed_args)];
mp_arg_parse_all(n_args - 1, pos_args + 1, kw_args, MP_ARRAY_SIZE(allowed_args), allowed_args, args);
mp_buffer_info_t bufinfo;
mp_get_buffer_raise(args[ARG_data].u_obj, &bufinfo, MP_BUFFER_READ);
// Disable interrupt while transmitting a frame.
if (self->cec_callback) {
omv_gpio_irq_enable(OMV_CEC_PIN, false);
}
if (!cec_send_frame(args[ARG_dst_addr].u_int, args[ARG_src_addr].u_int, bufinfo.len, bufinfo.buf)) {
mp_raise_msg(&mp_type_OSError, MP_ERROR_TEXT("Failed to send frame."));
}
if (self->cec_callback) {
omv_gpio_irq_enable(OMV_CEC_PIN, true);
}
return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_KW(py_cec_send_frame_obj, 4, py_cec_send_frame);
STATIC mp_obj_t py_cec_receive_frame(uint n_args, const mp_obj_t *pos_args, mp_map_t *kw_args) {
enum { ARG_dst_addr, ARG_timeout };
static const mp_arg_t allowed_args[] = {
{ MP_QSTR_dst_addr, MP_ARG_REQUIRED | MP_ARG_INT },
{ MP_QSTR_timeout, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 1000 } },
};
// Parse args.
mp_arg_val_t args[MP_ARRAY_SIZE(allowed_args)];
mp_arg_parse_all(n_args - 1, pos_args + 1, kw_args, MP_ARRAY_SIZE(allowed_args), allowed_args, args);
mp_uint_t start = mp_hal_ticks_ms();
while (omv_gpio_read(OMV_CEC_PIN)) {
if ((mp_hal_ticks_ms() - start) > args[ARG_timeout].u_int) {
mp_raise_msg(&mp_type_OSError, MP_ERROR_TEXT("Receive timeout!"));
}
}
uint8_t src_addr = 0;
mp_obj_array_t *frame = MP_OBJ_TO_PTR(mp_obj_new_bytearray_by_ref(16, m_new(byte, 16)));
if (cec_receive_frame(frame->items, &frame->len, args[ARG_dst_addr].u_int, &src_addr) != 0) {
mp_raise_msg(&mp_type_OSError, MP_ERROR_TEXT("Failed to receive frame."));
}
return mp_obj_new_tuple(2, (mp_obj_t []) { MP_OBJ_NEW_SMALL_INT(src_addr), frame });
}
STATIC MP_DEFINE_CONST_FUN_OBJ_KW(py_cec_receive_frame_obj, 2, py_cec_receive_frame);
STATIC mp_obj_t py_cec_frame_callback(mp_obj_t self_in, mp_obj_t cb, mp_obj_t dst_addr) {
py_display_data_obj_t *self = MP_OBJ_TO_PTR(self_in);
self->cec_callback = cb;
if (cb == mp_const_none) {
omv_gpio_irq_enable(OMV_CEC_PIN, false);
self->cec_frame = mp_const_none;
} else {
self->dst_addr = mp_obj_get_int(dst_addr);
self->cec_frame = mp_obj_new_bytearray_by_ref(0, m_new(byte, 16));
omv_gpio_config(OMV_CEC_PIN, OMV_GPIO_MODE_IT_FALL, OMV_GPIO_PULL_UP, OMV_GPIO_SPEED_LOW, -1);
omv_gpio_irq_register(OMV_CEC_PIN, cec_extint_callback, self_in);
omv_gpio_irq_enable(OMV_CEC_PIN, true);
}
return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_3(py_cec_frame_callback_obj, py_cec_frame_callback);
#endif // OMV_DISPLAY_CEC_ENABLE
#if OMV_DISPLAY_DDC_ENABLE
static bool ddc_checksum(uint8_t *data, int long_count) {
uint32_t *data32 = (uint32_t *) data;
uint32_t sum = 0;
for (int i = 0; i < long_count; i++) {
sum = __USADA8(data32[i], 0, sum);
}
return !(sum & 0xFF);
}
STATIC mp_obj_t py_ddc_display_id(mp_obj_t self_in) {
py_display_data_obj_t *self = MP_OBJ_TO_PTR(self_in);
if (mp_machine_soft_i2c_transfer(self->ddc_bus, self->ddc_addr, 1, &((mp_machine_i2c_buf_t) {
.len = 1, .buf = (uint8_t []) {0x00} // addr
}), MP_MACHINE_I2C_FLAG_STOP) == 1) {
fb_alloc_mark();
uint8_t *data = fb_alloc(128, FB_ALLOC_NO_HINT);
if (mp_machine_soft_i2c_transfer(self->ddc_bus, self->ddc_addr, 1, &((mp_machine_i2c_buf_t) {
.len = 128, .buf = data
}), MP_MACHINE_I2C_FLAG_READ | MP_MACHINE_I2C_FLAG_STOP) == 0) {
uint32_t *data32 = (uint32_t *) data;
if ((data32[0] == 0xFFFFFF00) && (data32[1] == 0x00FFFFFF) && ddc_checksum(data, 32)
&& (mp_machine_soft_i2c_transfer(self->ddc_bus, self->ddc_addr, 1, &((mp_machine_i2c_buf_t) {
.len = 1, .buf = (uint8_t []) {0x80} // addr
}), MP_MACHINE_I2C_FLAG_STOP) == 1)) {
int extensions = data[126];
int extensions_byte_size = extensions * 128;
int total_data_byte_size = extensions_byte_size + 128;
uint8_t *data2 = fb_alloc(total_data_byte_size, FB_ALLOC_NO_HINT), *data2_ext = data2 + 128;
memcpy(data2, data, 128);
if ((mp_machine_soft_i2c_transfer(self->ddc_bus, self->ddc_addr, 1, &((mp_machine_i2c_buf_t) {
.len = extensions_byte_size, .buf = data2_ext
}), MP_MACHINE_I2C_FLAG_READ | MP_MACHINE_I2C_FLAG_STOP) == 0)
&& ddc_checksum(data2_ext, extensions_byte_size / 4)) {
mp_obj_t result = mp_obj_new_bytes(data2, total_data_byte_size);
fb_alloc_free_till_mark();
return result;
}
}
}
}
mp_raise_msg(&mp_type_OSError, MP_ERROR_TEXT("Failed to get display id data!"));
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(py_ddc_display_id_obj, py_ddc_display_id);
#endif // OMV_DISPLAY_DDC_ENABLE
mp_obj_t py_display_data_make_new(const mp_obj_type_t *type, size_t n_args, size_t n_kw, const mp_obj_t *all_args) {
enum { ARG_cec, ARG_ddc, ARG_ddc_addr };
static const mp_arg_t allowed_args[] = {
{ MP_QSTR_cec, MP_ARG_KW_ONLY | MP_ARG_BOOL, {.u_bool = false} },
{ MP_QSTR_ddc, MP_ARG_KW_ONLY | MP_ARG_BOOL, {.u_bool = false} },
{ MP_QSTR_ddc_addr, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0x50 } },
};
// Parse args.
mp_arg_val_t args[MP_ARRAY_SIZE(allowed_args)];
mp_arg_parse_all_kw_array(n_args, n_kw, all_args, MP_ARRAY_SIZE(allowed_args), allowed_args, args);
py_display_data_obj_t *self = m_new_obj_with_finaliser(py_display_data_obj_t);
self->base.type = &py_display_data_type;
#if OMV_DISPLAY_CEC_ENABLE
self->cec_enabled = args[ARG_cec].u_bool;
self->dst_addr = 0;
self->cec_frame = mp_const_none;
self->cec_callback = mp_const_none;
if (self->cec_enabled) {
cec_init();
}
#endif // OMV_DISPLAY_CEC_ENABLE
#if OMV_DISPLAY_DDC_ENABLE
self->ddc_enabled = args[ARG_ddc].u_bool;
self->ddc_addr = args[ARG_ddc_addr].u_int;
self->ddc_bus = mp_const_none;
if (self->ddc_enabled) {
self->ddc_bus = MP_OBJ_TYPE_GET_SLOT(
&mp_machine_soft_i2c_type, make_new) (&mp_machine_soft_i2c_type, 2, 1, (const mp_obj_t []) {
(mp_obj_t) OMV_DDC_SCL_PIN, (mp_obj_t) OMV_DDC_SDA_PIN,
MP_OBJ_NEW_QSTR(MP_QSTR_freq),
MP_OBJ_NEW_SMALL_INT(100000)
});
}
#endif // OMV_DISPLAY_DDC_ENABLE
return MP_OBJ_FROM_PTR(self);
}
STATIC mp_obj_t py_display_data_deinit(mp_obj_t self_in) {
py_display_data_obj_t *self = MP_OBJ_TO_PTR(self_in);
#if OMV_DISPLAY_DDC_ENABLE
if (self->ddc_enabled) {
// Deinit bus.
}
#endif // OMV_DISPLAY_DDC_ENABLE
#if OMV_DISPLAY_CEC_ENABLE
if (self->cec_enabled) {
cec_deinit();
}
#endif // OMV_DISPLAY_CEC_ENABLE
return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(py_display_data_deinit_obj, py_display_data_deinit);
STATIC const mp_rom_map_elem_t py_display_data_locals_dict_table[] = {
{ MP_ROM_QSTR(MP_QSTR___name__), MP_ROM_QSTR(MP_QSTR_display_data) },
{ MP_ROM_QSTR(MP_QSTR___del__), MP_ROM_PTR(&py_display_data_deinit_obj) },
#if OMV_DISPLAY_DDC_ENABLE
{ MP_ROM_QSTR(MP_QSTR_display_id), MP_ROM_PTR(&py_ddc_display_id_obj) },
#endif
#if OMV_DISPLAY_CEC_ENABLE
{ MP_ROM_QSTR(MP_QSTR_send_frame), MP_ROM_PTR(&py_cec_send_frame_obj) },
{ MP_ROM_QSTR(MP_QSTR_receive_frame), MP_ROM_PTR(&py_cec_receive_frame_obj) },
{ MP_ROM_QSTR(MP_QSTR_frame_callback), MP_ROM_PTR(&py_cec_frame_callback_obj) },
#endif
};
MP_DEFINE_CONST_DICT(py_display_data_locals_dict, py_display_data_locals_dict_table);
MP_DEFINE_CONST_OBJ_TYPE(
py_display_data_type,
MP_QSTR_DisplayData,
MP_TYPE_FLAG_NONE,
make_new, py_display_data_make_new,
locals_dict, &py_display_data_locals_dict
);
#endif // OMV_DISPLAY_DATA