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..61b54c533 100644 --- a/SConscript +++ b/SConscript @@ -1,4 +1,4 @@ -Import('env', 'envCython', 'common') +Import('env', 'envPython', 'nanobind_lib', '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, nanobind_lib]+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, nanobind_lib]+common) if GetOption('extras'): env.Program('msgq/test_runner', ['msgq/msgq_tests.cc'], LIBS=[msgq]+common) diff --git a/SConstruct b/SConstruct index b7e1a4c76..a46d75153 100644 --- a/SConstruct +++ b/SConstruct @@ -1,8 +1,11 @@ import os import platform import subprocess +import sys import sysconfig +import nanobind + arch = subprocess.check_output(["uname", "-m"], encoding='utf8').rstrip() if platform.system() == "Darwin": arch = "Darwin" @@ -62,21 +65,29 @@ 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') +envPython = env.Clone(LIBS=[]) +envPython["CCFLAGS"] += ["-fvisibility=hidden", "-isystem", nanobind.include_dir()] +envPython["CPPDEFINES"] = ["NB_COMPACT_ASSERTIONS"] +# On Python >= 3.12, build against the limited API so one build works on every newer Python +if sys.version_info >= (3, 12): + envPython["CPPDEFINES"] += [("Py_LIMITED_API", "0x030C0000")] if arch == "Darwin": - envCython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"] + envPython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"] else: - envCython["LINKFLAGS"] = ["-pthread", "-shared"] + envPython["LINKFLAGS"] = ["-pthread", "-shared"] + +# nanobind's runtime, shared by both extension modules +envNanobind = envPython.Clone() +envNanobind["CCFLAGS"] = [f for f in envNanobind["CCFLAGS"] if f != "-Werror" and not f.startswith("-Wshadow")] +envNanobind["CCFLAGS"] += ["-fno-strict-aliasing", "-isystem", f"{nanobind.source_dir()}/../ext/robin_map/include"] +envNanobind["CPPDEFINES"] += ["NB_BUILD"] +nanobind_lib = envNanobind.Library('nanobind', envNanobind.SharedObject('nanobind', f"{nanobind.source_dir()}/nb_combined.cpp")) -Export('envCython') +Export('envPython', 'nanobind_lib') SConscript(['SConscript']) diff --git a/lefthook.yml b/lefthook.yml index 91b18aa52..43d664a3b 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') 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..f7555cb40 --- /dev/null +++ b/msgq/ipc_module.cc @@ -0,0 +1,155 @@ +// Python bindings for msgq +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "msgq/ipc.h" +#include "msgq/event.h" +#include "msgq/nb_helpers.h" + +using namespace nb::literals; + +struct IpcError : std::runtime_error { + using std::runtime_error::runtime_error; +}; +struct MultiplePublishersError : IpcError { + using IpcError::IpcError; +}; + +[[noreturn]] static void throw_ipc_error(const std::string *endpoint = nullptr) { + int err = errno; + std::string msg = "Messaging failure " + (endpoint ? "with " + *endpoint : "") + ": " + strerror(err); + if (err == EADDRINUSE) throw MultiplePublishersError(msg); + throw IpcError(msg); +} + +// Keeps the registered Python SubSockets alive, and lets poll() return them +struct PyPoller { + Poller poller; + std::vector sockets; +}; + +NB_MODULE(ipc_pyx, m) { + auto ipc_error = nb::exception(m, "IpcError"); + nb::exception(m, "MultiplePublishersError", ipc_error); + + nb::class_(m, "Event") + .def(nb::init<>()) + .def("set", &Event::set) + .def("clear", &Event::clear) + .def("wait", &Event::wait, "timeout"_a = -1, nb::call_guard()) + .def("peek", &Event::peek) + .def_prop_ro("fd", &Event::fd) + .def_prop_ro("ptr", [](Event &self) { return (uintptr_t)&self; }); + + nb::class_(m, "SocketEventHandle") + .def("__init__", [](SocketEventHandle *self, Name endpoint, Name identifier, bool override) { + new (self) SocketEventHandle(endpoint.value, identifier.value, override); + }, "endpoint"_a, "identifier"_a, "override"_a) + .def_prop_rw("enabled", &SocketEventHandle::is_enabled, &SocketEventHandle::set_enabled) + .def_prop_ro("recv_called_event", &SocketEventHandle::recv_called) + .def_prop_ro("recv_ready_event", &SocketEventHandle::recv_ready); + + m.def("toggle_fake_events", &SocketEventHandle::toggle_fake_events, "enabled"_a); + m.def("set_fake_prefix", [](Name prefix) { SocketEventHandle::set_fake_prefix(prefix.value); }, "prefix"_a); + m.def("get_fake_prefix", [] { + std::string prefix = SocketEventHandle::fake_prefix(); + return nb::bytes(prefix.data(), prefix.size()); + }); + m.def("delete_fake_prefix", [] { SocketEventHandle::set_fake_prefix(""); }); + m.def("wait_for_one_event", &Event::wait_for_one, "events"_a, "timeout"_a = -1, nb::call_guard()); + + nb::class_(m, "SubSocket") + .def(nb::init<>()) + .def("connect", [](SubSocket &self, Name endpoint, bool conflate, size_t segment_size) { + int r; + { + nb::gil_scoped_release release; + r = self.connect(endpoint.value, conflate, segment_size); + } + if (r != 0) throw_ipc_error(&endpoint.value); + // cppcheck-suppress assignBoolToPointer + }, "endpoint"_a, "conflate"_a = false, "segment_size"_a = 0) + .def("setTimeout", &SubSocket::setTimeout, "timeout"_a, nb::call_guard()) + .def("receive", [](SubSocket &self, bool non_blocking) -> nb::object { + Message *msg; + if (non_blocking) { + // a non-blocking receive is too fast to be worth releasing the GIL + msg = self.receive(true); + } else { + nb::gil_scoped_release release; + msg = self.receive(false); + } + if (!msg) return nb::none(); + nb::bytes ret(msg->getData(), msg->getSize()); + delete msg; + return ret; + // cppcheck-suppress assignBoolToPointer + }, "non_blocking"_a = false); + + nb::class_(m, "PubSocket") + .def(nb::init<>()) + .def("connect", [](PubSocket &self, Name endpoint, size_t segment_size) { + if (self.connect(endpoint.value, segment_size) != 0) throw_ipc_error(&endpoint.value); + }, "endpoint"_a, "segment_size"_a = 0) + .def("send", [](PubSocket &self, nb::handle data) { + int r; + size_t len; + if (nb::isinstance(data)) { + auto bytes = nb::borrow(data); + len = bytes.size(); + r = self.send((char *)bytes.c_str(), len); + } else { + Buffer buf(data); + len = buf.size(); + r = self.send(buf.data(), len); + } + if (r != (int)len) throw_ipc_error(); + }, "data"_a) + .def("all_readers_updated", &PubSocket::all_readers_updated) + .def("wait_for_readers", [](PubSocket &self, double timeout, double interval) { + auto deadline = std::chrono::steady_clock::now() + std::chrono::duration(timeout); + while (std::chrono::steady_clock::now() < deadline) { + if (self.all_readers_updated()) return; + { + nb::gil_scoped_release release; + std::this_thread::sleep_for(std::chrono::duration(interval)); + } + if (PyErr_CheckSignals() < 0) throw nb::python_error(); + } + PyErr_SetString(PyExc_TimeoutError, "subscriber did not connect"); + throw nb::python_error(); + }, "timeout"_a = 1.0, "interval"_a = 0.001); + + nb::class_(m, "Poller") + .def(nb::init<>()) + .def("registerSocket", [](PyPoller &self, nb::object socket) { + self.poller.registerSocket(nb::cast(socket)); + self.sockets.push_back(socket); + }, "socket"_a) + .def("poll", [](PyPoller &self, int timeout) { + std::vector ready; + { + nb::gil_scoped_release release; + ready = self.poller.poll(timeout); + } + nb::list ret; + for (SubSocket *s : ready) { + for (const nb::object &socket : self.sockets) { + if (nb::inst_ptr(socket) == s) { + ret.append(socket); + break; + } + } + } + return ret; + }, "timeout"_a); +} 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/nb_helpers.h b/msgq/nb_helpers.h new file mode 100644 index 000000000..68486a293 --- /dev/null +++ b/msgq/nb_helpers.h @@ -0,0 +1,58 @@ +#pragma once + +// Helpers shared by the nanobind Python modules. + +#include + +#include + +namespace nb = nanobind; + +// msgq accepts endpoint names as either str or bytes +struct Name { + std::string value; +}; + +namespace nanobind::detail { +template <> +struct type_caster { + NB_TYPE_CASTER(Name, const_name("str | bytes")) + + bool from_python(handle src, uint8_t, cleanup_list *) noexcept { + Py_ssize_t len; + if (PyBytes_Check(src.ptr())) { + char *data; + if (PyBytes_AsStringAndSize(src.ptr(), &data, &len) < 0) return false; + value.value.assign(data, len); + return true; + } + if (PyUnicode_Check(src.ptr())) { + const char *data = PyUnicode_AsUTF8AndSize(src.ptr(), &len); + if (!data) { + PyErr_Clear(); + return false; + } + value.value.assign(data, len); + return true; + } + return false; + } +}; +} // namespace nanobind::detail + +// A contiguous view of any object supporting the buffer protocol +class Buffer { + public: + explicit Buffer(nb::handle obj) { + if (PyObject_GetBuffer(obj.ptr(), &view, PyBUF_SIMPLE) < 0) throw nb::python_error(); + } + ~Buffer() { PyBuffer_Release(&view); } + Buffer(const Buffer &) = delete; + Buffer &operator=(const Buffer &) = delete; + + char *data() const { return (char *)view.buf; } + size_t size() const { return view.len; } + + private: + Py_buffer view; +}; 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..62c3101cc --- /dev/null +++ b/msgq/visionipc/visionipc_module.cc @@ -0,0 +1,91 @@ +// Python bindings for visionipc +#include +#include +#include +#include +#include + +#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/nb_helpers.h" + +using namespace nb::literals; + +struct PyVisionIpcClient { + VisionIpcClient client; + VisionIpcBufExtra extra = {}; + + PyVisionIpcClient(const std::string &name, VisionStreamType stream, bool conflate) : client(name, stream, conflate) {} + + // properties of the first buffer, or None before connecting + std::optional buffer_prop(size_t VisionBuf::*field) const { + if (!client.num_buffers) return std::nullopt; + return client.buffers[0].*field; + } +}; + +NB_MODULE(visionipc_pyx, m) { + m.def("get_endpoint_name", [](Name name, VisionStreamType stream) { return get_endpoint_name(name.value, stream); }, "name"_a, "stream"_a); + + // VisionBufs are only created by VisionIpcClient.recv() + nb::class_(m, "VisionBuf") + .def_prop_ro("data", [](VisionBuf &self) { + return nb::steal(PyMemoryView_FromMemory((char *)self.addr, self.len, PyBUF_WRITE)); + }) + .def_ro("width", &VisionBuf::width) + .def_ro("height", &VisionBuf::height) + .def_ro("stride", &VisionBuf::stride) + .def_ro("uv_offset", &VisionBuf::uv_offset) + .def_ro("idx", &VisionBuf::idx) + .def_ro("fd", &VisionBuf::fd) + .def_prop_ro("frame_id", &VisionBuf::get_frame_id); + + nb::class_(m, "VisionIpcServer") + .def("__init__", [](VisionIpcServer *self, Name name) { new (self) VisionIpcServer(name.value); }, "name"_a) + .def("create_buffers", &VisionIpcServer::create_buffers, "tp"_a, "num_buffers"_a, "width"_a, "height"_a) + .def("create_buffers_with_sizes", &VisionIpcServer::create_buffers_with_sizes, + "tp"_a, "num_buffers"_a, "width"_a, "height"_a, "size"_a, "stride"_a, "uv_offset"_a) + .def("send", [](VisionIpcServer &self, VisionStreamType tp, nb::handle data, uint32_t frame_id, uint64_t timestamp_sof, uint64_t timestamp_eof) { + Buffer src(data); + VisionBuf *buf = self.get_buffer(tp); + if (src.size() != buf->len) { + throw nb::value_error(("data is " + std::to_string(src.size()) + " bytes, but the buffer is " + std::to_string(buf->len) + " bytes").c_str()); + } + memcpy(buf->addr, src.data(), src.size()); + buf->set_frame_id(frame_id); + + VisionIpcBufExtra extra = {frame_id, timestamp_sof, timestamp_eof, false}; + self.send(buf, &extra, false); + }, "tp"_a, "data"_a, "frame_id"_a = 0, "timestamp_sof"_a = 0, "timestamp_eof"_a = 0) + .def("start_listener", &VisionIpcServer::start_listener); + + nb::class_(m, "VisionIpcClient") + .def("__init__", [](PyVisionIpcClient *self, Name name, VisionStreamType stream, bool conflate) { + new (self) PyVisionIpcClient(name.value, stream, conflate); + }, "name"_a, "stream"_a, "conflate"_a) + .def_prop_ro("width", [](const PyVisionIpcClient &self) { return self.buffer_prop(&VisionBuf::width); }) + .def_prop_ro("height", [](const PyVisionIpcClient &self) { return self.buffer_prop(&VisionBuf::height); }) + .def_prop_ro("stride", [](const PyVisionIpcClient &self) { return self.buffer_prop(&VisionBuf::stride); }) + .def_prop_ro("uv_offset", [](const PyVisionIpcClient &self) { return self.buffer_prop(&VisionBuf::uv_offset); }) + .def_prop_ro("buffer_len", [](const PyVisionIpcClient &self) { return self.buffer_prop(&VisionBuf::len); }) + .def_prop_ro("num_buffers", [](const PyVisionIpcClient &self) { return self.client.num_buffers; }) + .def_prop_ro("frame_id", [](const PyVisionIpcClient &self) { return self.extra.frame_id; }) + .def_prop_ro("timestamp_sof", [](const PyVisionIpcClient &self) { return self.extra.timestamp_sof; }) + .def_prop_ro("timestamp_eof", [](const PyVisionIpcClient &self) { return self.extra.timestamp_eof; }) + .def_prop_ro("valid", [](const PyVisionIpcClient &self) { return self.extra.valid; }) + // the returned VisionBuf keeps the client alive + .def("recv", [](PyVisionIpcClient &self, int timeout_ms) { return self.client.recv(&self.extra, timeout_ms); }, + "timeout_ms"_a = 100, nb::rv_policy::reference_internal, nb::call_guard()) + .def("connect", [](PyVisionIpcClient &self, bool blocking) { return self.client.connect(blocking); }, + "blocking"_a, nb::call_guard()) + .def("is_connected", [](PyVisionIpcClient &self) { return self.client.is_connected(); }) + .def_static("available_streams", [](Name name, bool block) { return VisionIpcClient::getAvailableStreams(name.value, block); }, + "name"_a, "block"_a, nb::call_guard()); +} 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..fa6f32a3c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [build-system] -requires = ["setuptools>=77", "Cython>=3.1", "scons"] +requires = ["setuptools>=77", "nanobind>=3.1", "scons"] build-backend = "setuptools.build_meta" [project] @@ -23,7 +23,7 @@ Repository = "https://github.com/commaai/msgq" [project.optional-dependencies] dev = [ "setuptools", # for distutils - "Cython", + "nanobind", "scons", "ruff", "cppcheck", @@ -41,13 +41,17 @@ 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-*" test-command = "python -m unittest discover -s msgq" +[tool.cibuildwheel.macos] +# nanobind uses aligned allocation, which needs macOS 10.13+ +environment = { MACOSX_DEPLOYMENT_TARGET = "10.14" } + # https://beta.ruff.rs/docs/configuration/#using-pyprojecttoml [tool.ruff] lint.select = ["E", "F", "W", "PIE", "C4", "ISC", "RUF100", "A"] @@ -57,9 +61,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..1cbad09d5 100644 --- a/setup.py +++ b/setup.py @@ -17,10 +17,14 @@ def build_extensions(self): self.copy_file(str(source), str(target)) +# nanobind supports the stable ABI on Python >= 3.12, so 3.11 gets its own wheel +abi3 = sys.version_info >= (3, 12) + setup( ext_modules=[ - Extension("msgq.ipc_pyx", sources=[]), - Extension("msgq.visionipc.visionipc_pyx", sources=[]), + Extension("msgq.ipc_pyx", sources=[], py_limited_api=abi3), + Extension("msgq.visionipc.visionipc_pyx", sources=[], py_limited_api=abi3), ], cmdclass={"build_ext": SConsBuildExt}, + options={"bdist_wheel": {"py_limited_api": "cp312"}} if abi3 else {}, ) 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