robolib.utils.utils
utils.py - generic utilities
1"""utils.py - generic utilities""" 2from __future__ import annotations 3from pathlib import Path 4from dataclasses import dataclass, field 5from collections import OrderedDict 6import threading 7import zlib 8 9from loggez import make_logger 10import numpy as np 11 12logger = make_logger("ROBOSIM", exists_ok=True) 13 14Rot3x3 = np.ndarray 15Pose4x4 = np.ndarray 16Point2D = np.ndarray 17Point3D = np.ndarray 18Point6D = np.ndarray 19Mat4x4 = np.ndarray 20Mat3x3 = np.ndarray 21 22def fmt(arr: np.ndarray, precision: int=2) -> str: 23 """formats a numpy array for logging as tuple""" 24 return "(" + ", ".join(f"{float(x):.{precision}g}" for x in arr) + ")" 25 26def get_project_root() -> Path: 27 """The project root (this file lives at src/robolib/utils/utils.py, so root is 3 levels up)""" 28 return Path(__file__).parents[3] 29 30@dataclass 31class FPVData: 32 """The first perso view camera data. Latest frame (as bytes), a lock (for concurrency) and a way to compress it""" 33 frame: bytes 34 frame_shape: tuple[int, int, int] 35 lock: threading.Lock = field(default_factory=threading.Lock) 36 frame_id: int = 0 # used for de-dup on client side 37 _frame_compressed: bytes | None = None 38 39 @property 40 def frame_compressed(self) -> bytes: 41 """compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.""" 42 if self._frame_compressed is None: 43 self._frame_compressed = zlib.compress(self.frame, level=1) 44 return self._frame_compressed 45 46# generic utils below 47 48def make_arr(*data): 49 """makes a numpy array from variable data, e.g. make_arr(1,2,3) <=> np.array([1,2,3], "float32")""" 50 return np.float32(data) 51 52class FixedSizeDict(OrderedDict): 53 """An dict with a fixed size. Useful for caching purposes.""" 54 def __init__(self, *args, maxlen: int = 0, **kwargs): 55 self._maxlen = maxlen 56 super().__init__(*args, **kwargs) 57 58 def __setitem__(self, key, value): 59 existed_before_add = key in self 60 super().__setitem__(key, value) 61 if not existed_before_add and len(self) > self._maxlen: 62 self.popitem(False) 63 64def ecs_data_as_np(x: list | str | dict | np.ndarray | None, dtype: str) -> np.ndarray | None: 65 """converts a list to a f32/int32 np array for microecs compatibility""" 66 if x is None: 67 return np.array([None], "object") 68 res = np.array(x, dtype=dtype) 69 res = res[None] if res.dtype == "object" and len(res.shape) == 0 else res 70 return res
logger =
<loggez.loggez.LoggezLogger object>
Rot3x3 =
<class 'numpy.ndarray'>
Pose4x4 =
<class 'numpy.ndarray'>
Point2D =
<class 'numpy.ndarray'>
Point3D =
<class 'numpy.ndarray'>
Point6D =
<class 'numpy.ndarray'>
Mat4x4 =
<class 'numpy.ndarray'>
Mat3x3 =
<class 'numpy.ndarray'>
def
fmt(arr: numpy.ndarray, precision: int = 2) -> str:
23def fmt(arr: np.ndarray, precision: int=2) -> str: 24 """formats a numpy array for logging as tuple""" 25 return "(" + ", ".join(f"{float(x):.{precision}g}" for x in arr) + ")"
formats a numpy array for logging as tuple
def
get_project_root() -> pathlib.Path:
27def get_project_root() -> Path: 28 """The project root (this file lives at src/robolib/utils/utils.py, so root is 3 levels up)""" 29 return Path(__file__).parents[3]
The project root (this file lives at src/robolib/utils/utils.py, so root is 3 levels up)
@dataclass
class
FPVData:
31@dataclass 32class FPVData: 33 """The first perso view camera data. Latest frame (as bytes), a lock (for concurrency) and a way to compress it""" 34 frame: bytes 35 frame_shape: tuple[int, int, int] 36 lock: threading.Lock = field(default_factory=threading.Lock) 37 frame_id: int = 0 # used for de-dup on client side 38 _frame_compressed: bytes | None = None 39 40 @property 41 def frame_compressed(self) -> bytes: 42 """compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.""" 43 if self._frame_compressed is None: 44 self._frame_compressed = zlib.compress(self.frame, level=1) 45 return self._frame_compressed
The first perso view camera data. Latest frame (as bytes), a lock (for concurrency) and a way to compress it
FPVData( frame: bytes, frame_shape: tuple[int, int, int], lock: <built-in function allocate_lock> = <factory>, frame_id: int = 0, _frame_compressed: bytes | None = None)
frame_compressed: bytes
40 @property 41 def frame_compressed(self) -> bytes: 42 """compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.""" 43 if self._frame_compressed is None: 44 self._frame_compressed = zlib.compress(self.frame, level=1) 45 return self._frame_compressed
compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.
def
make_arr(*data):
49def make_arr(*data): 50 """makes a numpy array from variable data, e.g. make_arr(1,2,3) <=> np.array([1,2,3], "float32")""" 51 return np.float32(data)
makes a numpy array from variable data, e.g. make_arr(1,2,3) <=> np.array([1,2,3], "float32")
class
FixedSizeDict(collections.OrderedDict):
53class FixedSizeDict(OrderedDict): 54 """An dict with a fixed size. Useful for caching purposes.""" 55 def __init__(self, *args, maxlen: int = 0, **kwargs): 56 self._maxlen = maxlen 57 super().__init__(*args, **kwargs) 58 59 def __setitem__(self, key, value): 60 existed_before_add = key in self 61 super().__setitem__(key, value) 62 if not existed_before_add and len(self) > self._maxlen: 63 self.popitem(False)
An dict with a fixed size. Useful for caching purposes.
def
ecs_data_as_np( x: list | str | dict | numpy.ndarray | None, dtype: str) -> numpy.ndarray | None:
65def ecs_data_as_np(x: list | str | dict | np.ndarray | None, dtype: str) -> np.ndarray | None: 66 """converts a list to a f32/int32 np array for microecs compatibility""" 67 if x is None: 68 return np.array([None], "object") 69 res = np.array(x, dtype=dtype) 70 res = res[None] if res.dtype == "object" and len(res.shape) == 0 else res 71 return res
converts a list to a f32/int32 np array for microecs compatibility