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 22@dataclass 23class FPVData: 24 """The first perso view camera data. Latest frame (as bytes), a lock (for concurrency) and a way to compress it""" 25 frame: bytes 26 frame_shape: tuple[int, int, int] 27 lock: threading.Lock = field(default_factory=threading.Lock) 28 frame_id: int = 0 # used for de-dup on client side 29 _frame_compressed: bytes | None = None 30 31 @property 32 def frame_compressed(self) -> bytes: 33 """compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.""" 34 if self._frame_compressed is None: 35 self._frame_compressed = zlib.compress(self.frame, level=1) 36 return self._frame_compressed 37 38class FixedSizeDict(OrderedDict): 39 """An dict with a fixed size. Useful for caching purposes.""" 40 def __init__(self, *args, maxlen: int = 0, **kwargs): 41 self._maxlen = maxlen 42 super().__init__(*args, **kwargs) 43 44 def __setitem__(self, key, value): 45 existed_before_add = key in self 46 super().__setitem__(key, value) 47 if not existed_before_add and len(self) > self._maxlen: 48 self.popitem(False) 49 50def fmt(arr: np.ndarray, precision: int=3) -> str: 51 """formats a numpy array for logging as tuple""" 52 return "(" + ", ".join(f"{float(x):.{precision}g}" for x in arr) + ")" 53 54def get_project_root() -> Path: 55 """The project root (this file lives at src/robolib/utils/utils.py, so root is 3 levels up)""" 56 return Path(__file__).parents[3] 57 58def make_arr(*data): 59 """makes a numpy array from variable data, e.g. make_arr(1,2,3) <=> np.array([1,2,3], "float32")""" 60 return np.float32(data)
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'>
@dataclass
class
FPVData:
23@dataclass 24class FPVData: 25 """The first perso view camera data. Latest frame (as bytes), a lock (for concurrency) and a way to compress it""" 26 frame: bytes 27 frame_shape: tuple[int, int, int] 28 lock: threading.Lock = field(default_factory=threading.Lock) 29 frame_id: int = 0 # used for de-dup on client side 30 _frame_compressed: bytes | None = None 31 32 @property 33 def frame_compressed(self) -> bytes: 34 """compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.""" 35 if self._frame_compressed is None: 36 self._frame_compressed = zlib.compress(self.frame, level=1) 37 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
32 @property 33 def frame_compressed(self) -> bytes: 34 """compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.""" 35 if self._frame_compressed is None: 36 self._frame_compressed = zlib.compress(self.frame, level=1) 37 return self._frame_compressed
compress the frame on demand when the client requests a get_state_with_frame. NOTE: use lock on call.
class
FixedSizeDict(collections.OrderedDict):
39class FixedSizeDict(OrderedDict): 40 """An dict with a fixed size. Useful for caching purposes.""" 41 def __init__(self, *args, maxlen: int = 0, **kwargs): 42 self._maxlen = maxlen 43 super().__init__(*args, **kwargs) 44 45 def __setitem__(self, key, value): 46 existed_before_add = key in self 47 super().__setitem__(key, value) 48 if not existed_before_add and len(self) > self._maxlen: 49 self.popitem(False)
An dict with a fixed size. Useful for caching purposes.
def
fmt(arr: numpy.ndarray, precision: int = 3) -> str:
51def fmt(arr: np.ndarray, precision: int=3) -> str: 52 """formats a numpy array for logging as tuple""" 53 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:
55def get_project_root() -> Path: 56 """The project root (this file lives at src/robolib/utils/utils.py, so root is 3 levels up)""" 57 return Path(__file__).parents[3]
The project root (this file lives at src/robolib/utils/utils.py, so root is 3 levels up)
def
make_arr(*data):
59def make_arr(*data): 60 """makes a numpy array from variable data, e.g. make_arr(1,2,3) <=> np.array([1,2,3], "float32")""" 61 return np.float32(data)
makes a numpy array from variable data, e.g. make_arr(1,2,3) <=> np.array([1,2,3], "float32")