|
|
@ -1,6 +1,5 @@ |
|
|
|
#!/usr/bin/env python3 |
|
|
|
#!/usr/bin/env python3 |
|
|
|
import capnp |
|
|
|
import capnp |
|
|
|
import dataclasses |
|
|
|
|
|
|
|
import os |
|
|
|
import os |
|
|
|
import time |
|
|
|
import time |
|
|
|
from typing import Any |
|
|
|
from typing import Any |
|
|
@ -61,13 +60,21 @@ def can_comm_callbacks(logcan: messaging.SubSocket, sendcan: messaging.PubSocket |
|
|
|
return can_recv, can_send |
|
|
|
return can_recv, can_send |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_FIELDS = '__dataclass_fields__' |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def is_dataclass_instance(obj): |
|
|
|
|
|
|
|
"""Returns True if obj is an instance of a dataclass.""" |
|
|
|
|
|
|
|
return hasattr(obj, _FIELDS) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def asdictref(obj) -> dict[str, Any]: |
|
|
|
def asdictref(obj) -> dict[str, Any]: |
|
|
|
"""Note that the resulting dict will contain references to the struct field values""" |
|
|
|
"""Note that the resulting dict will contain references to the struct field values""" |
|
|
|
if not dataclasses._is_dataclass_instance(obj): # type: ignore[attr-defined] |
|
|
|
if not is_dataclass_instance(obj): # type: ignore[attr-defined] |
|
|
|
raise TypeError("asdictref() should be called on dataclass instances") |
|
|
|
raise TypeError("asdictref() should be called on dataclass instances") |
|
|
|
|
|
|
|
|
|
|
|
def _asdictref_inner(obj) -> dict[str, Any] | Any: |
|
|
|
def _asdictref_inner(obj) -> dict[str, Any] | Any: |
|
|
|
if dataclasses._is_dataclass_instance(obj): # type: ignore[attr-defined] |
|
|
|
if is_dataclass_instance(obj): # type: ignore[attr-defined] |
|
|
|
ret = {} |
|
|
|
ret = {} |
|
|
|
for field in obj.__dataclass_fields__: |
|
|
|
for field in obj.__dataclass_fields__: |
|
|
|
ret[field] = _asdictref_inner(getattr(obj, field)) |
|
|
|
ret[field] = _asdictref_inner(getattr(obj, field)) |
|
|
|