diff --git a/MANIFEST.in b/MANIFEST.in index b3bc16ab9..d31491773 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,4 +1,3 @@ include README.md pyproject.toml setup.py SConstruct SConscript -recursive-include site_scons *.py -recursive-include msgq *.py *.pyx *.pxd *.h *.cc +recursive-include msgq *.py *.h *.cc global-exclude *.so *.o *.os *.a *.pyc diff --git a/SConscript b/SConscript index 5a8d52041..ab361a9ca 100644 --- a/SConscript +++ b/SConscript @@ -1,4 +1,4 @@ -Import('env', 'envCython', 'common') +Import('env', 'envPython', 'common') visionipc_dir = Dir('msgq/visionipc') @@ -10,7 +10,7 @@ msgq_objects = env.SharedObject([ 'msgq/msgq.cc', ]) msgq = env.Library('msgq', msgq_objects) -msgq_python = envCython.Program('msgq/ipc_pyx.so', 'msgq/ipc_pyx.pyx', LIBS=envCython["LIBS"]+[msgq]+common) +msgq_python = envPython.Program('msgq/ipc_pyx.so', 'msgq/ipc_module.cc', LIBS=[msgq]+common) # Build Vision IPC vipc_files = ['visionipc.cc', 'visionipc_server.cc', 'visionipc_client.cc'] @@ -24,9 +24,8 @@ vipc_objects = env.SharedObject(vipc_sources) visionipc = env.Library('visionipc', vipc_objects) -vipc_libs = envCython["LIBS"] + [visionipc, msgq] + common -envCython.Program(f'{visionipc_dir.abspath}/visionipc_pyx.so', f'{visionipc_dir.abspath}/visionipc_pyx.pyx', - LIBS=vipc_libs) +envPython.Program(f'{visionipc_dir.abspath}/visionipc_pyx.so', f'{visionipc_dir.abspath}/visionipc_module.cc', + LIBS=[visionipc, msgq]+common) if GetOption('extras'): env.Program('msgq/test_runner', ['msgq/msgq_tests.cc'], LIBS=[msgq]+common) diff --git a/SConstruct b/SConstruct index b7e1a4c76..d98cc5ec7 100644 --- a/SConstruct +++ b/SConstruct @@ -62,21 +62,18 @@ env = Environment( CFLAGS="-std=gnu11", CXXFLAGS="-std=c++1z", CPPPATH=cpppath, - CYTHONCFILESUFFIX=".cpp", - tools=["default", "cython"] ) Export('env', 'arch', 'common') -envCython = env.Clone(LIBS=[]) -envCython["CCFLAGS"] += ["-Wno-#warnings", "-Wno-cpp", "-Wno-shadow", "-Wno-deprecated-declarations"] -envCython["CCFLAGS"].remove('-Werror') +# Python extension modules use the limited API, so one build works on every Python >= 3.11 +envPython = env.Clone(LIBS=[]) if arch == "Darwin": - envCython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"] + envPython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"] else: - envCython["LINKFLAGS"] = ["-pthread", "-shared"] + envPython["LINKFLAGS"] = ["-pthread", "-shared"] -Export('envCython') +Export('envPython') SConscript(['SConscript']) diff --git a/lefthook.yml b/lefthook.yml index 91b18aa52..0ddd157bd 100644 --- a/lefthook.yml +++ b/lefthook.yml @@ -18,12 +18,12 @@ test: ty: run: ty check . codespell: - run: codespell {files} -L ned,stdio,master --builtin clear,rare,informal,usage,code,names,en-GB_to_en-US -S uv.lock,*_pyx.cpp + run: codespell {files} -L ned,stdio,master --builtin clear,rare,informal,usage,code,names,en-GB_to_en-US -S uv.lock files: git ls-tree -r HEAD --name-only cppcheck: - run: cppcheck --error-exitcode=1 --inline-suppr --language=c++ --force --quiet -j4 --check-level=exhaustive $(git ls-files '*.cc' | grep -v -E '(msgq_tests|test_runner)\.cc') + run: cppcheck --error-exitcode=1 --inline-suppr --library=python --language=c++ --force --quiet -j4 --check-level=exhaustive $(git ls-files '*.cc' | grep -v -E '(msgq_tests|test_runner)\.cc') cpplint: - run: cpplint --exclude=msgq/ipc_pyx.cpp --exclude=msgq/visionipc/visionipc_pyx.cpp --recursive --quiet --counting=detailed --linelength=240 --filter=-build,-legal,-readability,-runtime,-whitespace,+build/include_subdir,+build/forward_decl,+build/include_what_you_use,+build/deprecated,+whitespace/comma,+whitespace/line_length,+whitespace/empty_if_body,+whitespace/empty_loop_body,+whitespace/empty_conditional_body,+whitespace/forcolon,+whitespace/parens,+whitespace/semicolon,+whitespace/tab,+readability/braces msgq/ + run: cpplint --recursive --quiet --counting=detailed --linelength=240 --filter=-build,-legal,-readability,-runtime,-whitespace,+build/include_subdir,+build/forward_decl,+build/include_what_you_use,+build/deprecated,+whitespace/comma,+whitespace/line_length,+whitespace/empty_if_body,+whitespace/empty_loop_body,+whitespace/empty_conditional_body,+whitespace/forcolon,+whitespace/parens,+whitespace/semicolon,+whitespace/tab,+readability/braces msgq/ # *** tests *** test_runner: diff --git a/msgq/.gitignore b/msgq/.gitignore deleted file mode 100644 index 6bd751773..000000000 --- a/msgq/.gitignore +++ /dev/null @@ -1 +0,0 @@ -ipc_pyx.cpp diff --git a/msgq/ipc.pxd b/msgq/ipc.pxd deleted file mode 100644 index 3d35af0ae..000000000 --- a/msgq/ipc.pxd +++ /dev/null @@ -1,61 +0,0 @@ -# distutils: language = c++ -#cython: language_level=3 - -from libcpp.string cimport string -from libcpp.vector cimport vector -from libcpp cimport bool - - -cdef extern from "msgq/event.h": - cdef cppclass Event: - @staticmethod - int wait_for_one(vector[Event], int) except + - - Event() - Event(int) - void set() - int clear() - void wait(int) except + - bool peek() - int fd() - - cdef cppclass SocketEventHandle: - @staticmethod - void toggle_fake_events(bool) - @staticmethod - void set_fake_prefix(string) - @staticmethod - string fake_prefix() - - SocketEventHandle(string, string, bool) - bool is_enabled() - void set_enabled(bool) - Event recv_called() - Event recv_ready() - - -cdef extern from "msgq/ipc.h": - cdef cppclass Message: - void init(size_t) - void init(char *, size_t) - void close() - size_t getSize() - char *getData() - - cdef cppclass SubSocket: - SubSocket() nogil - int connect(string, bool, size_t) nogil - Message * receive(bool) nogil - void setTimeout(int) nogil - - cdef cppclass PubSocket: - PubSocket() - int connect(string, size_t) - int sendMessage(Message *) - int send(char *, size_t) - bool all_readers_updated() - - cdef cppclass Poller: - Poller() - void registerSocket(SubSocket *) - vector[SubSocket*] poll(int) nogil diff --git a/msgq/ipc_module.cc b/msgq/ipc_module.cc new file mode 100644 index 000000000..cfd6e8443 --- /dev/null +++ b/msgq/ipc_module.cc @@ -0,0 +1,483 @@ +// Python bindings for msgq, written against the CPython limited API (abi3). +#define PY_SSIZE_T_CLEAN +#define Py_LIMITED_API 0x030B0000 +#include + +#include +#include +#include +#include +#include +#include + +#include "msgq/ipc.h" +#include "msgq/event.h" + +#include "msgq/python_helpers.h" + +static PyObject *IpcError; +static PyObject *MultiplePublishersError; +static PyTypeObject *SubSocketType, *PubSocketType, *PollerType, *EventType, *SocketEventHandleType; + +static PyObject *raise_ipc_error(const std::string *endpoint) { + int err = errno; + std::string suffix = endpoint ? "with " + *endpoint : ""; + PyErr_Format(err == EADDRINUSE ? MultiplePublishersError : IpcError, "Messaging failure %s: %s", suffix.c_str(), strerror(err)); + return nullptr; +} + +// *** SubSocket *** + +struct SubSocketObject { + PyObject_HEAD + SubSocket *socket; +}; + +static PyObject *SubSocket_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + if (!PyArg_ParseTuple(args, ":SubSocket")) return nullptr; + auto self = (SubSocketObject *)alloc_object(type); + if (!self) return nullptr; + self->socket = new SubSocket(); + return (PyObject *)self; +} + +static void SubSocket_dealloc(PyObject *obj) { + auto self = (SubSocketObject *)obj; + if (self->socket) { + release_gil([&] { delete self->socket; }); + } + free_object(obj); +} + +static PyObject *SubSocket_connect(SubSocketObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"endpoint", "conflate", "segment_size", nullptr}; + PyObject *o[3] = {}; + std::string endpoint; + bool conflate = false; + size_t segment_size = 0; + if (!parse_args("connect", args, nargs, kwnames, kwlist, 1, o) || !to_string(o[0], endpoint) || + (o[1] && !to_bool(o[1], conflate)) || (o[2] && !to_size(o[2], segment_size))) { + return nullptr; + } + + int r; + release_gil([&] { r = self->socket->connect(endpoint, conflate, segment_size); }); + if (r != 0) return raise_ipc_error(&endpoint); + Py_RETURN_NONE; +} + +static PyObject *SubSocket_setTimeout(SubSocketObject *self, PyObject *arg) { + int timeout; + if (!to_int(arg, timeout)) return nullptr; + release_gil([&] { self->socket->setTimeout(timeout); }); + Py_RETURN_NONE; +} + +static PyObject *SubSocket_receive(SubSocketObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"non_blocking", nullptr}; + PyObject *o[1] = {}; + bool non_blocking = false; + if (!parse_args("receive", args, nargs, kwnames, kwlist, 0, o) || (o[0] && !to_bool(o[0], non_blocking))) return nullptr; + + Message *msg; + if (non_blocking) { + // a non-blocking receive is too fast to be worth releasing the GIL + msg = self->socket->receive(true); + } else { + release_gil([&] { msg = self->socket->receive(false); }); + } + if (!msg) Py_RETURN_NONE; + PyObject *ret = PyBytes_FromStringAndSize(msg->getData(), msg->getSize()); + delete msg; + return ret; +} + +static PyMethodDef SubSocket_methods[] = { + {"connect", (PyCFunction)(void (*)(void))SubSocket_connect, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"setTimeout", (PyCFunction)SubSocket_setTimeout, METH_O, nullptr}, + {"receive", (PyCFunction)(void (*)(void))SubSocket_receive, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {}, +}; + +static PyType_Slot SubSocket_slots[] = { + {Py_tp_new, (void *)SubSocket_new}, + {Py_tp_dealloc, (void *)SubSocket_dealloc}, + {Py_tp_methods, SubSocket_methods}, + {}, +}; + +// *** PubSocket *** + +struct PubSocketObject { + PyObject_HEAD + PubSocket *socket; +}; + +static PyObject *PubSocket_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + if (!PyArg_ParseTuple(args, ":PubSocket")) return nullptr; + auto self = (PubSocketObject *)alloc_object(type); + if (!self) return nullptr; + self->socket = new PubSocket(); + return (PyObject *)self; +} + +static void PubSocket_dealloc(PyObject *obj) { + delete ((PubSocketObject *)obj)->socket; + free_object(obj); +} + +static PyObject *PubSocket_connect(PubSocketObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"endpoint", "segment_size", nullptr}; + PyObject *o[2] = {}; + std::string endpoint; + size_t segment_size = 0; + if (!parse_args("connect", args, nargs, kwnames, kwlist, 1, o) || !to_string(o[0], endpoint) || + (o[1] && !to_size(o[1], segment_size))) { + return nullptr; + } + + if (self->socket->connect(endpoint, segment_size) != 0) return raise_ipc_error(&endpoint); + Py_RETURN_NONE; +} + +static PyObject *PubSocket_send(PubSocketObject *self, PyObject *data) { + char *buf; + Py_ssize_t len; + int r; + if (PyBytes_Check(data)) { + if (PyBytes_AsStringAndSize(data, &buf, &len) < 0) return nullptr; + r = self->socket->send(buf, len); + } else { + Py_buffer view; + if (PyObject_GetBuffer(data, &view, PyBUF_SIMPLE) < 0) return nullptr; + len = view.len; + r = self->socket->send((char *)view.buf, len); + PyBuffer_Release(&view); + } + if (r != len) return raise_ipc_error(nullptr); + Py_RETURN_NONE; +} + +static PyObject *PubSocket_all_readers_updated(PubSocketObject *self, PyObject *) { + return PyBool_FromLong(self->socket->all_readers_updated()); +} + +static PyObject *PubSocket_wait_for_readers(PubSocketObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"timeout", "interval", nullptr}; + PyObject *o[2] = {}; + double timeout = 1.0, interval = 0.001; + if (!parse_args("wait_for_readers", args, nargs, kwnames, kwlist, 0, o) || (o[0] && !to_double(o[0], timeout)) || + (o[1] && !to_double(o[1], interval))) { + return nullptr; + } + + double deadline = monotonic() + timeout; + while (monotonic() < deadline) { + if (self->socket->all_readers_updated()) Py_RETURN_NONE; + struct timespec ts = {(time_t)interval, (long)((interval - (time_t)interval) * 1e9)}; + release_gil([&] { nanosleep(&ts, nullptr); }); + if (PyErr_CheckSignals() < 0) return nullptr; + } + PyErr_SetString(PyExc_TimeoutError, "subscriber did not connect"); + return nullptr; +} + +static PyMethodDef PubSocket_methods[] = { + {"connect", (PyCFunction)(void (*)(void))PubSocket_connect, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"send", (PyCFunction)PubSocket_send, METH_O, nullptr}, + {"all_readers_updated", (PyCFunction)PubSocket_all_readers_updated, METH_NOARGS, nullptr}, + {"wait_for_readers", (PyCFunction)(void (*)(void))PubSocket_wait_for_readers, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {}, +}; + +static PyType_Slot PubSocket_slots[] = { + {Py_tp_new, (void *)PubSocket_new}, + {Py_tp_dealloc, (void *)PubSocket_dealloc}, + {Py_tp_methods, PubSocket_methods}, + {}, +}; + +// *** Poller *** + +struct PollerObject { + PyObject_HEAD + Poller *poller; + PyObject *sockets; // list of registered SubSockets, keeps them alive and maps results back to them +}; + +static PyObject *Poller_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + if (!PyArg_ParseTuple(args, ":Poller")) return nullptr; + auto self = (PollerObject *)alloc_object(type); + if (!self) return nullptr; + self->sockets = PyList_New(0); + if (!self->sockets) { + Py_DECREF(self); + return nullptr; + } + self->poller = new Poller(); + return (PyObject *)self; +} + +static void Poller_dealloc(PyObject *obj) { + auto self = (PollerObject *)obj; + delete self->poller; + Py_XDECREF(self->sockets); + free_object(obj); +} + +static PyObject *Poller_registerSocket(PollerObject *self, PyObject *socket) { + if (!PyObject_TypeCheck(socket, SubSocketType)) return PyErr_Format(PyExc_TypeError, "socket must be a SubSocket"); + if (PyList_Append(self->sockets, socket) < 0) return nullptr; + self->poller->registerSocket(((SubSocketObject *)socket)->socket); + Py_RETURN_NONE; +} + +static PyObject *Poller_poll(PollerObject *self, PyObject *arg) { + int timeout; + if (!to_int(arg, timeout)) return nullptr; + std::vector ready; + release_gil([&] { ready = self->poller->poll(timeout); }); + + PyObject *ret = PyList_New(0); + Py_ssize_t n = PyList_Size(self->sockets); + for (SubSocket *s : ready) { + for (Py_ssize_t i = 0; ret && i < n; i++) { + PyObject *socket = PyList_GetItem(self->sockets, i); + if (((SubSocketObject *)socket)->socket == s) { + if (PyList_Append(ret, socket) < 0) Py_CLEAR(ret); + break; + } + } + } + return ret; +} + +static PyMethodDef Poller_methods[] = { + {"registerSocket", (PyCFunction)Poller_registerSocket, METH_O, nullptr}, + {"poll", (PyCFunction)Poller_poll, METH_O, nullptr}, + {}, +}; + +static PyType_Slot Poller_slots[] = { + {Py_tp_new, (void *)Poller_new}, + {Py_tp_dealloc, (void *)Poller_dealloc}, + {Py_tp_methods, Poller_methods}, + {}, +}; + +// *** Event *** + +struct EventObject { + PyObject_HEAD + Event event; +}; + +static PyObject *new_event(Event event) { + auto self = (EventObject *)alloc_object(EventType); + if (self) self->event = event; + return (PyObject *)self; +} + +static PyObject *Event_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + if (!PyArg_ParseTuple(args, ":Event")) return nullptr; + return new_event(Event()); +} + +static PyObject *Event_set(EventObject *self, PyObject *) { + return call_cpp([&] { self->event.set(); Py_RETURN_NONE; }); +} + +static PyObject *Event_clear(EventObject *self, PyObject *) { + return call_cpp([&] { return PyLong_FromLong(self->event.clear()); }); +} + +static PyObject *Event_wait(EventObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"timeout", nullptr}; + PyObject *o[1] = {}; + int timeout = -1; + if (!parse_args("wait", args, nargs, kwnames, kwlist, 0, o) || (o[0] && !to_int(o[0], timeout))) return nullptr; + bool ok = without_gil([&] { self->event.wait(timeout); }); + if (!ok) return nullptr; + Py_RETURN_NONE; +} + +static PyObject *Event_peek(EventObject *self, PyObject *) { + return call_cpp([&] { return PyBool_FromLong(self->event.peek()); }); +} + +static PyObject *Event_get_fd(EventObject *self, void *) { return PyLong_FromLong(self->event.fd()); } +static PyObject *Event_get_ptr(EventObject *self, void *) { return PyLong_FromVoidPtr(&self->event); } + +static PyMethodDef Event_methods[] = { + {"set", (PyCFunction)Event_set, METH_NOARGS, nullptr}, + {"clear", (PyCFunction)Event_clear, METH_NOARGS, nullptr}, + {"wait", (PyCFunction)(void (*)(void))Event_wait, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"peek", (PyCFunction)Event_peek, METH_NOARGS, nullptr}, + {}, +}; + +static PyGetSetDef Event_getset[] = { + {"fd", (getter)Event_get_fd, nullptr, nullptr, nullptr}, + {"ptr", (getter)Event_get_ptr, nullptr, nullptr, nullptr}, + {}, +}; + +static PyType_Slot Event_slots[] = { + {Py_tp_new, (void *)Event_new}, + {Py_tp_dealloc, (void *)free_object}, + {Py_tp_methods, Event_methods}, + {Py_tp_getset, Event_getset}, + {}, +}; + +// *** SocketEventHandle *** + +struct SocketEventHandleObject { + PyObject_HEAD + SocketEventHandle *handle; +}; + +static PyObject *SocketEventHandle_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + static const char *kwlist[] = {"endpoint", "identifier", "override", nullptr}; + PyObject *o[3]; + std::string endpoint, identifier; + bool override; + if (!PyArg_ParseTupleAndKeywords(args, kwds, "OOO:SocketEventHandle", (char **)kwlist, &o[0], &o[1], &o[2]) || + !to_string(o[0], endpoint) || !to_string(o[1], identifier) || !to_bool(o[2], override)) { + return nullptr; + } + auto self = (SocketEventHandleObject *)alloc_object(type); + if (!self) return nullptr; + try { + self->handle = new SocketEventHandle(endpoint, identifier, override); + } catch (const std::exception &e) { + Py_DECREF(self); + PyErr_SetString(PyExc_RuntimeError, e.what()); + return nullptr; + } + return (PyObject *)self; +} + +static void SocketEventHandle_dealloc(PyObject *obj) { + delete ((SocketEventHandleObject *)obj)->handle; + free_object(obj); +} + +static PyObject *SocketEventHandle_get_enabled(SocketEventHandleObject *self, void *) { + return PyBool_FromLong(self->handle->is_enabled()); +} + +static int SocketEventHandle_set_enabled(SocketEventHandleObject *self, PyObject *value, void *) { + bool enabled; + if (!value) { + PyErr_SetString(PyExc_AttributeError, "cannot delete enabled"); + return -1; + } + if (!to_bool(value, enabled)) return -1; + self->handle->set_enabled(enabled); + return 0; +} + +static PyObject *SocketEventHandle_get_recv_called(SocketEventHandleObject *self, void *) { return new_event(self->handle->recv_called()); } +static PyObject *SocketEventHandle_get_recv_ready(SocketEventHandleObject *self, void *) { return new_event(self->handle->recv_ready()); } + +static PyGetSetDef SocketEventHandle_getset[] = { + {"enabled", (getter)SocketEventHandle_get_enabled, (setter)SocketEventHandle_set_enabled, nullptr, nullptr}, + {"recv_called_event", (getter)SocketEventHandle_get_recv_called, nullptr, nullptr, nullptr}, + {"recv_ready_event", (getter)SocketEventHandle_get_recv_ready, nullptr, nullptr, nullptr}, + {}, +}; + +static PyType_Slot SocketEventHandle_slots[] = { + {Py_tp_new, (void *)SocketEventHandle_new}, + {Py_tp_dealloc, (void *)SocketEventHandle_dealloc}, + {Py_tp_getset, SocketEventHandle_getset}, + {}, +}; + +// *** module functions *** + +static PyObject *toggle_fake_events(PyObject *, PyObject *arg) { + bool enabled; + if (!to_bool(arg, enabled)) return nullptr; + SocketEventHandle::toggle_fake_events(enabled); + Py_RETURN_NONE; +} + +static PyObject *set_fake_prefix(PyObject *, PyObject *arg) { + std::string prefix; + if (!to_string(arg, prefix)) return nullptr; + SocketEventHandle::set_fake_prefix(prefix); + Py_RETURN_NONE; +} + +static PyObject *get_fake_prefix(PyObject *, PyObject *) { + std::string prefix = SocketEventHandle::fake_prefix(); + return PyBytes_FromStringAndSize(prefix.data(), prefix.size()); +} + +static PyObject *delete_fake_prefix(PyObject *, PyObject *) { + SocketEventHandle::set_fake_prefix(""); + Py_RETURN_NONE; +} + +static PyObject *wait_for_one_event(PyObject *, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"events", "timeout", nullptr}; + PyObject *o[2] = {}; + int timeout = -1; + if (!parse_args("wait_for_one_event", args, nargs, kwnames, kwlist, 1, o) || (o[1] && !to_int(o[1], timeout))) return nullptr; + + std::vector events; + PyObject *iter = PyObject_GetIter(o[0]); + if (!iter) return nullptr; + while (PyObject *item = PyIter_Next(iter)) { + bool ok = PyObject_TypeCheck(item, EventType); + if (ok) events.push_back(((EventObject *)item)->event); + Py_DECREF(item); + if (!ok) { + PyErr_SetString(PyExc_TypeError, "events must be Event objects"); + break; + } + } + Py_DECREF(iter); + if (PyErr_Occurred()) return nullptr; + + int ret; + bool ok = without_gil([&] { ret = Event::wait_for_one(events, timeout); }); + if (!ok) return nullptr; + return PyLong_FromLong(ret); +} + +static PyMethodDef module_methods[] = { + {"toggle_fake_events", (PyCFunction)toggle_fake_events, METH_O, nullptr}, + {"set_fake_prefix", (PyCFunction)set_fake_prefix, METH_O, nullptr}, + {"get_fake_prefix", (PyCFunction)get_fake_prefix, METH_NOARGS, nullptr}, + {"delete_fake_prefix", (PyCFunction)delete_fake_prefix, METH_NOARGS, nullptr}, + {"wait_for_one_event", (PyCFunction)(void (*)(void))wait_for_one_event, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {}, +}; + +static PyModuleDef module_def = {PyModuleDef_HEAD_INIT, "msgq.ipc_pyx", nullptr, -1, module_methods}; + +PyMODINIT_FUNC PyInit_ipc_pyx(void) { + PyObject *m = PyModule_Create(&module_def); + if (!m) return nullptr; + + IpcError = PyErr_NewException("msgq.ipc_pyx.IpcError", nullptr, nullptr); + MultiplePublishersError = PyErr_NewException("msgq.ipc_pyx.MultiplePublishersError", IpcError, nullptr); + if (!IpcError || !MultiplePublishersError || PyModule_AddObjectRef(m, "IpcError", IpcError) < 0 || + PyModule_AddObjectRef(m, "MultiplePublishersError", MultiplePublishersError) < 0) { + Py_DECREF(m); + return nullptr; + } + + if (!add_type(m, "msgq.ipc_pyx.SubSocket", sizeof(SubSocketObject), SubSocket_slots, &SubSocketType) || + !add_type(m, "msgq.ipc_pyx.PubSocket", sizeof(PubSocketObject), PubSocket_slots, &PubSocketType) || + !add_type(m, "msgq.ipc_pyx.Poller", sizeof(PollerObject), Poller_slots, &PollerType) || + !add_type(m, "msgq.ipc_pyx.Event", sizeof(EventObject), Event_slots, &EventType) || + !add_type(m, "msgq.ipc_pyx.SocketEventHandle", sizeof(SocketEventHandleObject), SocketEventHandle_slots, &SocketEventHandleType)) { + Py_DECREF(m); + return nullptr; + } + return m; +} diff --git a/msgq/ipc_pyx.pyx b/msgq/ipc_pyx.pyx deleted file mode 100644 index 446f303f6..000000000 --- a/msgq/ipc_pyx.pyx +++ /dev/null @@ -1,234 +0,0 @@ -# distutils: language = c++ -# cython: c_string_encoding=ascii, language_level=3 - -import time -from libcpp.string cimport string -from libcpp.vector cimport vector -from libcpp cimport bool -from libc cimport errno -from libc.string cimport strerror -from cython.operator import dereference - - -from .ipc cimport SubSocket as cppSubSocket -from .ipc cimport PubSocket as cppPubSocket -from .ipc cimport Poller as cppPoller -from .ipc cimport Message as cppMessage -from .ipc cimport Event as cppEvent, SocketEventHandle as cppSocketEventHandle - - -class IpcError(Exception): - def __init__(self, endpoint=None): - suffix = f"with {endpoint.decode('utf-8')}" if endpoint else "" - message = f"Messaging failure {suffix}: {strerror(errno.errno).decode('utf-8')}" - super().__init__(message) - - -class MultiplePublishersError(IpcError): - pass - - -def toggle_fake_events(bool enabled): - cppSocketEventHandle.toggle_fake_events(enabled) - - -def set_fake_prefix(string prefix): - cppSocketEventHandle.set_fake_prefix(prefix) - - -def get_fake_prefix(): - return cppSocketEventHandle.fake_prefix() - - -def delete_fake_prefix(): - cppSocketEventHandle.set_fake_prefix(b"") - - -def wait_for_one_event(list events, int timeout=-1): - cdef vector[cppEvent] items - for event in events: - items.push_back(dereference(event.ptr)) - return cppEvent.wait_for_one(items, timeout) - - -cdef class Event: - cdef cppEvent event; - - cdef setEvent(self, cppEvent event): - self.event = event - - def set(self): - self.event.set() - - def clear(self): - return self.event.clear() - - def wait(self, int timeout=-1): - self.event.wait(timeout) - - def peek(self): - return self.event.peek() - - @property - def fd(self): - return self.event.fd() - - @property - def ptr(self): - return &self.event - - -cdef class SocketEventHandle: - cdef cppSocketEventHandle * handle; - - def __cinit__(self, string endpoint, string identifier, bool override): - self.handle = new cppSocketEventHandle(endpoint, identifier, override) - - def __dealloc__(self): - del self.handle - - @property - def enabled(self): - return self.handle.is_enabled() - - @enabled.setter - def enabled(self, bool value): - self.handle.set_enabled(value) - - @property - def recv_called_event(self): - e = Event() - e.setEvent(self.handle.recv_called()) - - return e - - @property - def recv_ready_event(self): - e = Event() - e.setEvent(self.handle.recv_ready()) - - return e - - -cdef class Poller: - cdef cppPoller * poller - cdef list sub_sockets - - def __cinit__(self): - self.sub_sockets = [] - self.poller = new cppPoller() - - def __dealloc__(self): - del self.poller - - def registerSocket(self, SubSocket socket): - self.sub_sockets.append(socket) - self.poller.registerSocket(socket.socket) - - def poll(self, timeout): - sockets = [] - cdef int t = timeout - - with nogil: - result = self.poller.poll(t) - - for s in result: - socket = SubSocket() - socket.setPtr(s) - sockets.append(socket) - - return sockets - - -cdef class SubSocket: - cdef cppSubSocket * socket - cdef bool is_owner - - def __cinit__(self): - with nogil: - self.socket = new cppSubSocket() - - self.is_owner = True - - def __dealloc__(self): - if self.is_owner: - with nogil: - del self.socket - - cdef setPtr(self, cppSubSocket * ptr): - if self.is_owner: - with nogil: - del self.socket - - self.is_owner = False - self.socket = ptr - - def connect(self, string endpoint, bool conflate=False, size_t segment_size=0): - cdef int r - with nogil: - r = self.socket.connect(endpoint, conflate, segment_size) - - if r != 0: - if errno.errno == errno.EADDRINUSE: - raise MultiplePublishersError(endpoint) - else: - raise IpcError(endpoint) - - def setTimeout(self, int timeout): - with nogil: - self.socket.setTimeout(timeout) - - def receive(self, bool non_blocking=False): - cdef cppMessage *msg - with nogil: - msg = self.socket.receive(non_blocking) - - if msg == NULL: - return None - else: - sz = msg.getSize() - m = msg.getData()[:sz] - with nogil: - del msg - - return m - - -cdef class PubSocket: - cdef cppPubSocket * socket - - def __cinit__(self): - self.socket = new cppPubSocket() - - def __dealloc__(self): - del self.socket - - def connect(self, string endpoint, size_t segment_size=0): - r = self.socket.connect(endpoint, segment_size) - - if r != 0: - if errno.errno == errno.EADDRINUSE: - raise MultiplePublishersError(endpoint) - else: - raise IpcError(endpoint) - - def send(self, bytes data): - length = len(data) - r = self.socket.send(data, length) - - if r != length: - if errno.errno == errno.EADDRINUSE: - raise MultiplePublishersError - else: - raise IpcError - - def all_readers_updated(self): - return self.socket.all_readers_updated() - - def wait_for_readers(self, double timeout=1.0, double interval=0.001): - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if self.all_readers_updated(): - return - time.sleep(interval) - raise TimeoutError("subscriber did not connect") diff --git a/msgq/python_helpers.h b/msgq/python_helpers.h new file mode 100644 index 000000000..089eacfba --- /dev/null +++ b/msgq/python_helpers.h @@ -0,0 +1,168 @@ +#pragma once + +// Small helpers shared by the limited API (abi3) Python modules. Include after Python.h. + +#include +#include +#include +#include +#include +#include + +inline PyObject *alloc_object(PyTypeObject *type) { + return ((allocfunc)PyType_GetSlot(type, Py_tp_alloc))(type, 0); +} + +inline void free_object(PyObject *obj) { + PyTypeObject *type = Py_TYPE(obj); + ((freefunc)PyType_GetSlot(type, Py_tp_free))(obj); + Py_DECREF(type); +} + +// Adds a heap type named after the last component of its fully qualified name +inline bool add_type(PyObject *module, const char *qualname, int size, PyType_Slot *slots, PyTypeObject **out, + unsigned int flags = Py_TPFLAGS_BASETYPE) { + PyType_Spec spec = {qualname, size, 0, Py_TPFLAGS_DEFAULT | flags, slots}; + PyObject *type = PyType_FromSpec(&spec); + if (!type) return false; + *out = (PyTypeObject *)type; + const char *name = strrchr(qualname, '.'); + return PyModule_AddObjectRef(module, name ? name + 1 : qualname, type) == 0; +} + +// Parses METH_FASTCALL | METH_KEYWORDS arguments into out, which must be zeroed and hold one slot per kwlist entry +inline bool parse_args(const char *fname, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames, + const char *const *kwlist, Py_ssize_t required, PyObject **out) { + Py_ssize_t n = 0; + while (kwlist[n]) n++; + if (nargs > n) { + PyErr_Format(PyExc_TypeError, "%s() takes at most %zd positional arguments (%zd given)", fname, n, nargs); + return false; + } + for (Py_ssize_t i = 0; i < nargs; i++) out[i] = args[i]; + + Py_ssize_t nkw = kwnames ? PyTuple_Size(kwnames) : 0; + for (Py_ssize_t i = 0; i < nkw; i++) { + PyObject *key = PyTuple_GetItem(kwnames, i); + Py_ssize_t j = 0; + while (j < n && PyUnicode_CompareWithASCIIString(key, kwlist[j]) != 0) j++; + if (j == n) { + PyErr_Format(PyExc_TypeError, "%s() got an unexpected keyword argument '%U'", fname, key); + return false; + } + if (out[j]) { + PyErr_Format(PyExc_TypeError, "%s() got multiple values for argument '%s'", fname, kwlist[j]); + return false; + } + out[j] = args[nargs + i]; + } + + for (Py_ssize_t i = 0; i < required; i++) { + if (!out[i]) { + PyErr_Format(PyExc_TypeError, "%s() missing required argument '%s'", fname, kwlist[i]); + return false; + } + } + return true; +} + +inline bool to_string(PyObject *obj, std::string &out) { + const char *data; + Py_ssize_t len; + if (PyBytes_Check(obj)) { + char *buf; + if (PyBytes_AsStringAndSize(obj, &buf, &len) < 0) return false; + data = buf; + } else if (PyUnicode_Check(obj)) { + if (!(data = PyUnicode_AsUTF8AndSize(obj, &len))) return false; + } else if (PyByteArray_Check(obj)) { + data = PyByteArray_AsString(obj); + len = PyByteArray_Size(obj); + } else { + PyErr_SetString(PyExc_TypeError, "expected str or bytes"); + return false; + } + out.assign(data, len); + return true; +} + +inline bool to_bool(PyObject *obj, bool &out) { + int r = PyObject_IsTrue(obj); + if (r < 0) return false; + out = r; + return true; +} + +inline bool to_int(PyObject *obj, int &out) { + long v = PyLong_AsLong(obj); + if (v == -1 && PyErr_Occurred()) return false; + if (v < INT_MIN || v > INT_MAX) { + PyErr_SetString(PyExc_OverflowError, "value too large to convert to int"); + return false; + } + out = (int)v; + return true; +} + +inline bool to_size(PyObject *obj, size_t &out) { + size_t v = PyLong_AsSize_t(obj); + if (v == (size_t)-1 && PyErr_Occurred()) return false; + out = v; + return true; +} + +inline bool to_uint64(PyObject *obj, uint64_t &out) { + unsigned long long v = PyLong_AsUnsignedLongLong(obj); + if (v == (unsigned long long)-1 && PyErr_Occurred()) return false; + out = v; + return true; +} + +inline bool to_double(PyObject *obj, double &out) { + double v = PyFloat_AsDouble(obj); + if (v == -1.0 && PyErr_Occurred()) return false; + out = v; + return true; +} + +inline double monotonic() { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return ts.tv_sec + ts.tv_nsec * 1e-9; +} + +// Calls f, translating C++ exceptions into RuntimeError +template +PyObject *call_cpp(F f) { + try { + return f(); + } catch (const std::exception &e) { + PyErr_SetString(PyExc_RuntimeError, e.what()); + return nullptr; + } +} + +// Calls f with the GIL released +template +void release_gil(F f) { + PyThreadState *state = PyEval_SaveThread(); + f(); + PyEval_RestoreThread(state); +} + +// Calls f with the GIL released, translating C++ exceptions into RuntimeError +template +bool without_gil(F f) { + std::string error; + bool ok = true; + release_gil([&] { + try { + f(); + } catch (const std::exception &e) { + error = e.what(); + ok = false; + } + }); + if (!ok) PyErr_SetString(PyExc_RuntimeError, error.c_str()); + return ok; +} diff --git a/msgq/visionipc/.gitignore b/msgq/visionipc/.gitignore deleted file mode 100644 index fa5a2b9f3..000000000 --- a/msgq/visionipc/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -visionipc_pyx.cpp -*.so diff --git a/msgq/visionipc/visionipc.pxd b/msgq/visionipc/visionipc.pxd deleted file mode 100644 index 2552fe0e4..000000000 --- a/msgq/visionipc/visionipc.pxd +++ /dev/null @@ -1,52 +0,0 @@ -# distutils: language = c++ -#cython: language_level=3 - -from libcpp.string cimport string -from libcpp.vector cimport vector -from libcpp.set cimport set -from libc.stdint cimport uint32_t, uint64_t -from libcpp cimport bool, int - -cdef extern from "msgq/visionipc/visionbuf.h": - ctypedef uint32_t VisionStreamType "VisionStreamType" - - cdef cppclass VisionBuf: - void * addr - int fd - size_t len - size_t width - size_t height - size_t stride - size_t uv_offset - size_t idx - void set_frame_id(uint64_t id) - uint64_t get_frame_id() - -cdef extern from "msgq/visionipc/visionipc.h": - struct VisionIpcBufExtra: - uint32_t frame_id - uint64_t timestamp_sof - uint64_t timestamp_eof - bool valid - -cdef extern from "msgq/visionipc/visionipc_server.h": - string get_endpoint_name(string, VisionStreamType) - - cdef cppclass VisionIpcServer: - VisionIpcServer(string) - void create_buffers(VisionStreamType, size_t, size_t, size_t) - void create_buffers_with_sizes(VisionStreamType, size_t, size_t, size_t, size_t, size_t, size_t) - VisionBuf * get_buffer(VisionStreamType) - void send(VisionBuf *, VisionIpcBufExtra *, bool) - void start_listener() - -cdef extern from "msgq/visionipc/visionipc_client.h": - cdef cppclass VisionIpcClient: - int num_buffers - VisionBuf buffers[1] - VisionIpcClient(string, VisionStreamType, bool) - VisionBuf * recv(VisionIpcBufExtra *, int) nogil - bool connect(bool) - bool is_connected() - @staticmethod - set[VisionStreamType] getAvailableStreams(string, bool) diff --git a/msgq/visionipc/visionipc_module.cc b/msgq/visionipc/visionipc_module.cc new file mode 100644 index 000000000..7ccf09540 --- /dev/null +++ b/msgq/visionipc/visionipc_module.cc @@ -0,0 +1,328 @@ +// Python bindings for visionipc, written against the CPython limited API (abi3). +#define PY_SSIZE_T_CLEAN +#define Py_LIMITED_API 0x030B0000 +#include + +#include +#include +#include + +#include "msgq/visionipc/visionbuf.h" +#include "msgq/visionipc/visionipc.h" +#include "msgq/visionipc/visionipc_client.h" +#include "msgq/visionipc/visionipc_server.h" + +#include "msgq/python_helpers.h" + +static PyTypeObject *VisionBufType, *VisionIpcServerType, *VisionIpcClientType; + +static bool to_stream_type(PyObject *obj, VisionStreamType &out) { + unsigned long v = PyLong_AsUnsignedLong(obj); + if (v == (unsigned long)-1 && PyErr_Occurred()) return false; + if (v > UINT32_MAX) { + PyErr_SetString(PyExc_OverflowError, "stream type must fit in 32 bits"); + return false; + } + out = (VisionStreamType)v; + return true; +} + +static PyObject *py_get_endpoint_name(PyObject *, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"name", "stream", nullptr}; + PyObject *o[2] = {}; + std::string name; + VisionStreamType stream; + if (!parse_args("get_endpoint_name", args, nargs, kwnames, kwlist, 2, o) || !to_string(o[0], name) || !to_stream_type(o[1], stream)) { + return nullptr; + } + std::string endpoint = get_endpoint_name(name, stream); + return PyUnicode_FromStringAndSize(endpoint.data(), endpoint.size()); +} + +// *** VisionBuf *** + +struct VisionBufObject { + PyObject_HEAD + VisionBuf *buf; + PyObject *owner; // the VisionIpcClient that owns buf +}; + +static void VisionBuf_dealloc(PyObject *obj) { + Py_XDECREF(((VisionBufObject *)obj)->owner); + free_object(obj); +} + +static PyObject *VisionBuf_get_data(VisionBufObject *self, void *) { + return PyMemoryView_FromMemory((char *)self->buf->addr, self->buf->len, PyBUF_WRITE); +} +static PyObject *VisionBuf_get_width(VisionBufObject *self, void *) { return PyLong_FromSize_t(self->buf->width); } +static PyObject *VisionBuf_get_height(VisionBufObject *self, void *) { return PyLong_FromSize_t(self->buf->height); } +static PyObject *VisionBuf_get_stride(VisionBufObject *self, void *) { return PyLong_FromSize_t(self->buf->stride); } +static PyObject *VisionBuf_get_uv_offset(VisionBufObject *self, void *) { return PyLong_FromSize_t(self->buf->uv_offset); } +static PyObject *VisionBuf_get_idx(VisionBufObject *self, void *) { return PyLong_FromSize_t(self->buf->idx); } +static PyObject *VisionBuf_get_fd(VisionBufObject *self, void *) { return PyLong_FromLong(self->buf->fd); } +static PyObject *VisionBuf_get_frame_id(VisionBufObject *self, void *) { return PyLong_FromUnsignedLongLong(self->buf->get_frame_id()); } + +static PyGetSetDef VisionBuf_getset[] = { + {"data", (getter)VisionBuf_get_data, nullptr, nullptr, nullptr}, + {"width", (getter)VisionBuf_get_width, nullptr, nullptr, nullptr}, + {"height", (getter)VisionBuf_get_height, nullptr, nullptr, nullptr}, + {"stride", (getter)VisionBuf_get_stride, nullptr, nullptr, nullptr}, + {"uv_offset", (getter)VisionBuf_get_uv_offset, nullptr, nullptr, nullptr}, + {"idx", (getter)VisionBuf_get_idx, nullptr, nullptr, nullptr}, + {"fd", (getter)VisionBuf_get_fd, nullptr, nullptr, nullptr}, + {"frame_id", (getter)VisionBuf_get_frame_id, nullptr, nullptr, nullptr}, + {}, +}; + +static PyType_Slot VisionBuf_slots[] = { + {Py_tp_dealloc, (void *)VisionBuf_dealloc}, + {Py_tp_getset, VisionBuf_getset}, + {}, +}; + +// *** VisionIpcServer *** + +struct VisionIpcServerObject { + PyObject_HEAD + VisionIpcServer *server; +}; + +static PyObject *VisionIpcServer_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + static const char *kwlist[] = {"name", nullptr}; + PyObject *name_obj; + std::string name; + if (!PyArg_ParseTupleAndKeywords(args, kwds, "O:VisionIpcServer", (char **)kwlist, &name_obj) || !to_string(name_obj, name)) return nullptr; + auto self = (VisionIpcServerObject *)alloc_object(type); + if (self) self->server = new VisionIpcServer(name); + return (PyObject *)self; +} + +static void VisionIpcServer_dealloc(PyObject *obj) { + auto self = (VisionIpcServerObject *)obj; + if (self->server) { + // joins the listener thread + release_gil([&] { delete self->server; }); + } + free_object(obj); +} + +static PyObject *VisionIpcServer_create_buffers(VisionIpcServerObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"tp", "num_buffers", "width", "height", nullptr}; + PyObject *o[4] = {}; + VisionStreamType tp; + size_t num_buffers, width, height; + if (!parse_args("create_buffers", args, nargs, kwnames, kwlist, 4, o) || !to_stream_type(o[0], tp) || !to_size(o[1], num_buffers) || + !to_size(o[2], width) || !to_size(o[3], height)) { + return nullptr; + } + self->server->create_buffers(tp, num_buffers, width, height); + Py_RETURN_NONE; +} + +static PyObject *VisionIpcServer_create_buffers_with_sizes(VisionIpcServerObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"tp", "num_buffers", "width", "height", "size", "stride", "uv_offset", nullptr}; + PyObject *o[7] = {}; + VisionStreamType tp; + size_t num_buffers, width, height, size, stride, uv_offset; + if (!parse_args("create_buffers_with_sizes", args, nargs, kwnames, kwlist, 7, o) || !to_stream_type(o[0], tp) || + !to_size(o[1], num_buffers) || !to_size(o[2], width) || !to_size(o[3], height) || !to_size(o[4], size) || + !to_size(o[5], stride) || !to_size(o[6], uv_offset)) { + return nullptr; + } + self->server->create_buffers_with_sizes(tp, num_buffers, width, height, size, stride, uv_offset); + Py_RETURN_NONE; +} + +static PyObject *VisionIpcServer_send(VisionIpcServerObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"tp", "data", "frame_id", "timestamp_sof", "timestamp_eof", nullptr}; + PyObject *o[5] = {}; + VisionStreamType tp; + VisionIpcBufExtra extra = {}; + if (!parse_args("send", args, nargs, kwnames, kwlist, 2, o) || !to_stream_type(o[0], tp) || (o[2] && !to_stream_type(o[2], extra.frame_id)) || + (o[3] && !to_uint64(o[3], extra.timestamp_sof)) || (o[4] && !to_uint64(o[4], extra.timestamp_eof))) { + return nullptr; + } + + Py_buffer data; + if (PyObject_GetBuffer(o[1], &data, PyBUF_SIMPLE) < 0) return nullptr; + VisionBuf *buf = self->server->get_buffer(tp); + if ((size_t)data.len != buf->len) { + PyErr_Format(PyExc_ValueError, "data is %zd bytes, but the buffer is %zu bytes", data.len, buf->len); + PyBuffer_Release(&data); + return nullptr; + } + memcpy(buf->addr, data.buf, data.len); + PyBuffer_Release(&data); + + buf->set_frame_id(extra.frame_id); + self->server->send(buf, &extra, false); + Py_RETURN_NONE; +} + +static PyObject *VisionIpcServer_start_listener(VisionIpcServerObject *self, PyObject *) { + self->server->start_listener(); + Py_RETURN_NONE; +} + +static PyMethodDef VisionIpcServer_methods[] = { + {"create_buffers", (PyCFunction)(void (*)(void))VisionIpcServer_create_buffers, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"create_buffers_with_sizes", (PyCFunction)(void (*)(void))VisionIpcServer_create_buffers_with_sizes, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"send", (PyCFunction)(void (*)(void))VisionIpcServer_send, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"start_listener", (PyCFunction)VisionIpcServer_start_listener, METH_NOARGS, nullptr}, + {}, +}; + +static PyType_Slot VisionIpcServer_slots[] = { + {Py_tp_new, (void *)VisionIpcServer_new}, + {Py_tp_dealloc, (void *)VisionIpcServer_dealloc}, + {Py_tp_methods, VisionIpcServer_methods}, + {}, +}; + +// *** VisionIpcClient *** + +struct VisionIpcClientObject { + PyObject_HEAD + VisionIpcClient *client; + VisionIpcBufExtra extra; +}; + +static PyObject *VisionIpcClient_new(PyTypeObject *type, PyObject *args, PyObject *kwds) { + static const char *kwlist[] = {"name", "stream", "conflate", nullptr}; + PyObject *o[3]; + std::string name; + VisionStreamType stream; + bool conflate; + if (!PyArg_ParseTupleAndKeywords(args, kwds, "OOO:VisionIpcClient", (char **)kwlist, &o[0], &o[1], &o[2]) || + !to_string(o[0], name) || !to_stream_type(o[1], stream) || !to_bool(o[2], conflate)) { + return nullptr; + } + auto self = (VisionIpcClientObject *)alloc_object(type); + if (self) self->client = new VisionIpcClient(name, stream, conflate); + return (PyObject *)self; +} + +static void VisionIpcClient_dealloc(PyObject *obj) { + delete ((VisionIpcClientObject *)obj)->client; + free_object(obj); +} + +// properties of the first buffer, or None before connecting +static PyObject *buffer_prop(VisionIpcClientObject *self, size_t VisionBuf::*field) { + if (!self->client->num_buffers) Py_RETURN_NONE; + return PyLong_FromSize_t(self->client->buffers[0].*field); +} +static PyObject *VisionIpcClient_get_width(VisionIpcClientObject *self, void *) { return buffer_prop(self, &VisionBuf::width); } +static PyObject *VisionIpcClient_get_height(VisionIpcClientObject *self, void *) { return buffer_prop(self, &VisionBuf::height); } +static PyObject *VisionIpcClient_get_stride(VisionIpcClientObject *self, void *) { return buffer_prop(self, &VisionBuf::stride); } +static PyObject *VisionIpcClient_get_uv_offset(VisionIpcClientObject *self, void *) { return buffer_prop(self, &VisionBuf::uv_offset); } +static PyObject *VisionIpcClient_get_buffer_len(VisionIpcClientObject *self, void *) { return buffer_prop(self, &VisionBuf::len); } +static PyObject *VisionIpcClient_get_num_buffers(VisionIpcClientObject *self, void *) { return PyLong_FromLong(self->client->num_buffers); } +static PyObject *VisionIpcClient_get_frame_id(VisionIpcClientObject *self, void *) { return PyLong_FromUnsignedLong(self->extra.frame_id); } +static PyObject *VisionIpcClient_get_timestamp_sof(VisionIpcClientObject *self, void *) { return PyLong_FromUnsignedLongLong(self->extra.timestamp_sof); } +static PyObject *VisionIpcClient_get_timestamp_eof(VisionIpcClientObject *self, void *) { return PyLong_FromUnsignedLongLong(self->extra.timestamp_eof); } +static PyObject *VisionIpcClient_get_valid(VisionIpcClientObject *self, void *) { return PyBool_FromLong(self->extra.valid); } + +static PyObject *VisionIpcClient_recv(VisionIpcClientObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"timeout_ms", nullptr}; + PyObject *o[1] = {}; + int timeout_ms = 100; + if (!parse_args("recv", args, nargs, kwnames, kwlist, 0, o) || (o[0] && !to_int(o[0], timeout_ms))) return nullptr; + + VisionBuf *buf; + release_gil([&] { buf = self->client->recv(&self->extra, timeout_ms); }); + if (!buf) Py_RETURN_NONE; + + auto ret = (VisionBufObject *)alloc_object(VisionBufType); + if (!ret) return nullptr; + ret->buf = buf; + ret->owner = Py_NewRef((PyObject *)self); + return (PyObject *)ret; +} + +static PyObject *VisionIpcClient_connect(VisionIpcClientObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"blocking", nullptr}; + PyObject *o[1] = {}; + bool blocking, ret; + if (!parse_args("connect", args, nargs, kwnames, kwlist, 1, o) || !to_bool(o[0], blocking)) return nullptr; + release_gil([&] { ret = self->client->connect(blocking); }); + return PyBool_FromLong(ret); +} + +static PyObject *VisionIpcClient_is_connected(VisionIpcClientObject *self, PyObject *) { + return PyBool_FromLong(self->client->is_connected()); +} + +static PyObject *VisionIpcClient_available_streams(PyObject *, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames) { + static const char *const kwlist[] = {"name", "block", nullptr}; + PyObject *o[2] = {}; + std::string name; + bool block; + if (!parse_args("available_streams", args, nargs, kwnames, kwlist, 2, o) || !to_string(o[0], name) || !to_bool(o[1], block)) return nullptr; + + std::set streams; + release_gil([&] { streams = VisionIpcClient::getAvailableStreams(name, block); }); + + PyObject *ret = PySet_New(nullptr); + for (VisionStreamType stream : streams) { + PyObject *item = PyLong_FromUnsignedLong(stream); + if (!item || PySet_Add(ret, item) < 0) { + Py_XDECREF(item); + Py_DECREF(ret); + return nullptr; + } + Py_DECREF(item); + } + return ret; +} + +static PyGetSetDef VisionIpcClient_getset[] = { + {"width", (getter)VisionIpcClient_get_width, nullptr, nullptr, nullptr}, + {"height", (getter)VisionIpcClient_get_height, nullptr, nullptr, nullptr}, + {"stride", (getter)VisionIpcClient_get_stride, nullptr, nullptr, nullptr}, + {"uv_offset", (getter)VisionIpcClient_get_uv_offset, nullptr, nullptr, nullptr}, + {"buffer_len", (getter)VisionIpcClient_get_buffer_len, nullptr, nullptr, nullptr}, + {"num_buffers", (getter)VisionIpcClient_get_num_buffers, nullptr, nullptr, nullptr}, + {"frame_id", (getter)VisionIpcClient_get_frame_id, nullptr, nullptr, nullptr}, + {"timestamp_sof", (getter)VisionIpcClient_get_timestamp_sof, nullptr, nullptr, nullptr}, + {"timestamp_eof", (getter)VisionIpcClient_get_timestamp_eof, nullptr, nullptr, nullptr}, + {"valid", (getter)VisionIpcClient_get_valid, nullptr, nullptr, nullptr}, + {}, +}; + +static PyMethodDef VisionIpcClient_methods[] = { + {"recv", (PyCFunction)(void (*)(void))VisionIpcClient_recv, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"connect", (PyCFunction)(void (*)(void))VisionIpcClient_connect, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {"is_connected", (PyCFunction)VisionIpcClient_is_connected, METH_NOARGS, nullptr}, + {"available_streams", (PyCFunction)(void (*)(void))VisionIpcClient_available_streams, METH_FASTCALL | METH_KEYWORDS | METH_STATIC, nullptr}, + {}, +}; + +static PyType_Slot VisionIpcClient_slots[] = { + {Py_tp_new, (void *)VisionIpcClient_new}, + {Py_tp_dealloc, (void *)VisionIpcClient_dealloc}, + {Py_tp_methods, VisionIpcClient_methods}, + {Py_tp_getset, VisionIpcClient_getset}, + {}, +}; + +static PyMethodDef module_methods[] = { + {"get_endpoint_name", (PyCFunction)(void (*)(void))py_get_endpoint_name, METH_FASTCALL | METH_KEYWORDS, nullptr}, + {}, +}; + +static PyModuleDef module_def = {PyModuleDef_HEAD_INIT, "msgq.visionipc.visionipc_pyx", nullptr, -1, module_methods}; + +PyMODINIT_FUNC PyInit_visionipc_pyx(void) { + PyObject *m = PyModule_Create(&module_def); + if (!m) return nullptr; + if (!add_type(m, "msgq.visionipc.visionipc_pyx.VisionBuf", sizeof(VisionBufObject), VisionBuf_slots, &VisionBufType, + Py_TPFLAGS_DISALLOW_INSTANTIATION) || + !add_type(m, "msgq.visionipc.visionipc_pyx.VisionIpcServer", sizeof(VisionIpcServerObject), VisionIpcServer_slots, &VisionIpcServerType) || + !add_type(m, "msgq.visionipc.visionipc_pyx.VisionIpcClient", sizeof(VisionIpcClientObject), VisionIpcClient_slots, &VisionIpcClientType)) { + Py_DECREF(m); + return nullptr; + } + return m; +} diff --git a/msgq/visionipc/visionipc_pyx.pxd b/msgq/visionipc/visionipc_pyx.pxd deleted file mode 100644 index 6e2d5ed05..000000000 --- a/msgq/visionipc/visionipc_pyx.pxd +++ /dev/null @@ -1,10 +0,0 @@ -# distutils: language = c++ -#cython: language_level=3 - -from .visionipc cimport VisionBuf as cppVisionBuf - -cdef class VisionBuf: - cdef cppVisionBuf * buf - - @staticmethod - cdef create(cppVisionBuf*) diff --git a/msgq/visionipc/visionipc_pyx.pyx b/msgq/visionipc/visionipc_pyx.pyx deleted file mode 100644 index a274d4f7a..000000000 --- a/msgq/visionipc/visionipc_pyx.pyx +++ /dev/null @@ -1,163 +0,0 @@ -# distutils: language = c++ -# cython: c_string_encoding=ascii, language_level=3 - -from libc.string cimport memcpy -from libc.stdint cimport uint32_t, uint64_t -from libcpp cimport bool -from libcpp.string cimport string - -from .visionipc cimport VisionIpcServer as cppVisionIpcServer -from .visionipc cimport VisionIpcClient as cppVisionIpcClient -from .visionipc cimport VisionBuf as cppVisionBuf -from .visionipc cimport VisionIpcBufExtra, VisionStreamType -from .visionipc cimport get_endpoint_name as cpp_get_endpoint_name - - -def get_endpoint_name(string name, VisionStreamType stream): - return cpp_get_endpoint_name(name, stream).decode('utf-8') - - -cdef class VisionBuf: - @staticmethod - cdef create(cppVisionBuf * cbuf): - buf = VisionBuf() - buf.buf = cbuf - return buf - - @property - def data(self): - cdef unsigned char[:] data = self.buf.addr - return memoryview(data) - - @property - def width(self): - return self.buf.width - - @property - def height(self): - return self.buf.height - - @property - def stride(self): - return self.buf.stride - - @property - def uv_offset(self): - return self.buf.uv_offset - - @property - def idx(self): - return self.buf.idx - - @property - def fd(self): - return self.buf.fd - - @property - def frame_id(self): - return self.buf.get_frame_id() - - -cdef class VisionIpcServer: - cdef cppVisionIpcServer * server - - def __init__(self, string name): - self.server = new cppVisionIpcServer(name) - - def create_buffers(self, VisionStreamType tp, size_t num_buffers, size_t width, size_t height): - self.server.create_buffers(tp, num_buffers, width, height) - - def create_buffers_with_sizes(self, VisionStreamType tp, size_t num_buffers, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset): - self.server.create_buffers_with_sizes(tp, num_buffers, width, height, size, stride, uv_offset) - - def send(self, VisionStreamType tp, const unsigned char[:] data, uint32_t frame_id=0, uint64_t timestamp_sof=0, uint64_t timestamp_eof=0): - cdef cppVisionBuf * buf = self.server.get_buffer(tp) - - # Populate buffer - assert buf.len == len(data) - memcpy(buf.addr, &data[0], len(data)) - buf.set_frame_id(frame_id) - - cdef VisionIpcBufExtra extra - extra.frame_id = frame_id - extra.timestamp_sof = timestamp_sof - extra.timestamp_eof = timestamp_eof - extra.valid = False - - self.server.send(buf, &extra, False) - - def start_listener(self): - self.server.start_listener() - - def __dealloc__(self): - del self.server - - -cdef class VisionIpcClient: - cdef cppVisionIpcClient * client - cdef VisionIpcBufExtra extra - - def __cinit__(self, string name, VisionStreamType stream, bool conflate): - self.client = new cppVisionIpcClient(name, stream, conflate) - - def __dealloc__(self): - del self.client - - @property - def width(self): - return self.client.buffers[0].width if self.client.num_buffers else None - - @property - def height(self): - return self.client.buffers[0].height if self.client.num_buffers else None - - @property - def stride(self): - return self.client.buffers[0].stride if self.client.num_buffers else None - - @property - def uv_offset(self): - return self.client.buffers[0].uv_offset if self.client.num_buffers else None - - @property - def buffer_len(self): - return self.client.buffers[0].len if self.client.num_buffers else None - - @property - def num_buffers(self): - return self.client.num_buffers - - @property - def frame_id(self): - return self.extra.frame_id - - @property - def timestamp_sof(self): - return self.extra.timestamp_sof - - @property - def timestamp_eof(self): - return self.extra.timestamp_eof - - @property - def valid(self): - return self.extra.valid - - def recv(self, int timeout_ms=100): - cdef cppVisionBuf * buf - # release the GIL, this can block for timeout_ms - with nogil: - buf = self.client.recv(&self.extra, timeout_ms) - if not buf: - return None - return VisionBuf.create(buf) - - def connect(self, bool blocking): - return self.client.connect(blocking) - - def is_connected(self): - return self.client.is_connected() - - @staticmethod - def available_streams(string name, bool block): - return cppVisionIpcClient.getAvailableStreams(name, block) diff --git a/pyproject.toml b/pyproject.toml index e2f4d16ce..76fb9820c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [build-system] -requires = ["setuptools>=77", "Cython>=3.1", "scons"] +requires = ["setuptools>=77", "scons"] build-backend = "setuptools.build_meta" [project] @@ -23,7 +23,6 @@ Repository = "https://github.com/commaai/msgq" [project.optional-dependencies] dev = [ "setuptools", # for distutils - "Cython", "scons", "ruff", "cppcheck", @@ -41,8 +40,8 @@ include = ["msgq", "msgq.*"] namespaces = false [tool.setuptools.package-data] -"msgq" = ["*.pxd", "*.h"] -"msgq.visionipc" = ["*.pxd", "*.h"] +"msgq" = ["*.h"] +"msgq.visionipc" = ["*.h"] [tool.cibuildwheel] build = "cp311-* cp312-* cp313-* cp314-*" @@ -57,9 +56,6 @@ lint.flake8-implicit-str-concat.allow-multiline=false line-length = 160 target-version="py311" -[tool.ty.src] -exclude = ["site_scons/"] - [tool.ty.rules] -# Cython modules are compiled at build time, not available for static analysis +# extension modules are compiled at build time, not available for static analysis unresolved-import = "ignore" diff --git a/setup.py b/setup.py index 45e2c172d..b1d9a8047 100644 --- a/setup.py +++ b/setup.py @@ -19,8 +19,9 @@ def build_extensions(self): setup( ext_modules=[ - Extension("msgq.ipc_pyx", sources=[]), - Extension("msgq.visionipc.visionipc_pyx", sources=[]), + Extension("msgq.ipc_pyx", sources=[], py_limited_api=True), + Extension("msgq.visionipc.visionipc_pyx", sources=[], py_limited_api=True), ], cmdclass={"build_ext": SConsBuildExt}, + options={"bdist_wheel": {"py_limited_api": "cp311"}}, ) diff --git a/site_scons/site_tools/cython.py b/site_scons/site_tools/cython.py deleted file mode 100644 index c29147553..000000000 --- a/site_scons/site_tools/cython.py +++ /dev/null @@ -1,72 +0,0 @@ -import re -import SCons -from SCons.Action import Action -from SCons.Scanner import Scanner - -pyx_from_import_re = re.compile(r'^from\s+(\S+)\s+cimport', re.M) -pyx_import_re = re.compile(r'^cimport\s+(\S+)', re.M) -cdef_import_re = re.compile(r'^cdef extern from\s+.(\S+).:', re.M) - - -def pyx_scan(node, env, path, arg=None): - contents = node.get_text_contents() - - # from cimport ... - matches = pyx_from_import_re.findall(contents) - # cimport - matches += pyx_import_re.findall(contents) - - # Modules can be either .pxd or .pyx files - files = [m.replace('.', '/') + '.pxd' for m in matches] - files += [m.replace('.', '/') + '.pyx' for m in matches] - - # cdef extern from - files += cdef_import_re.findall(contents) - - # Handle relative imports - cur_dir = str(node.get_dir()) - files = [cur_dir + f if f.startswith('/') else f for f in files] - - # Filter out non-existing files (probably system imports) - files = [f for f in files if env.File(f).exists()] - return env.File(files) - - -pyxscanner = Scanner(function=pyx_scan, skeys=['.pyx', '.pxd'], recursive=True) -cythonAction = Action("$CYTHONCOM") - - -def create_builder(env): - try: - cython = env['BUILDERS']['Cython'] - except KeyError: - cython = SCons.Builder.Builder( - action=cythonAction, - emitter={}, - suffix=cython_suffix_emitter, - single_source=1 - ) - env.Append(SCANNERS=pyxscanner) - env['BUILDERS']['Cython'] = cython - return cython - -def cython_suffix_emitter(env, source): - return "$CYTHONCFILESUFFIX" - -def generate(env): - env["CYTHON"] = "cythonize" - env["CYTHONCOM"] = "$CYTHON $CYTHONFLAGS $SOURCE" - env["CYTHONCFILESUFFIX"] = ".cpp" - - c_file, _ = SCons.Tool.createCFileBuilders(env) - - c_file.suffix['.pyx'] = cython_suffix_emitter - c_file.add_action('.pyx', cythonAction) - - c_file.suffix['.py'] = cython_suffix_emitter - c_file.add_action('.py', cythonAction) - - create_builder(env) - -def exists(env): - return True