Latest-value iterator for a subscribed global scene camera.
Use it as a context manager or call
close to unsubscribe. The stream
keeps only the newest frame, so a slow consumer drops intermediate images
instead of accumulating latency. Iteration also skips frames older than
max_age.
Parameters:
-
robot
('Robot')
–
Owning client connection. The name is historical; cameras are
global scene resources and are not owned by a robot command address.
-
camera_id
(str)
–
-
max_age
(float, default:
0.25
)
–
Maximum frame age in seconds accepted by iterator mode.
Source code in sim2bot/client.py
| class CameraStream:
"""Latest-value iterator for a subscribed global scene camera.
Use it as a context manager or call
[`close`][sim2bot.client.CameraStream.close] to unsubscribe. The stream
keeps only the newest frame, so a slow consumer drops intermediate images
instead of accumulating latency. Iteration also skips frames older than
``max_age``.
Args:
robot: Owning client connection. The name is historical; cameras are
global scene resources and are not owned by a robot command address.
camera_id: ID returned by [`Robot.cameras`][sim2bot.client.Robot.cameras].
max_age: Maximum frame age in seconds accepted by iterator mode.
"""
def __init__(self, robot: "Robot", camera_id: str, max_age: float = 0.25) -> None:
self._robot = robot
self.camera_id = camera_id
self._slot = _LatestSlot()
self._last_seq = 0
self._max_age = max_age
self._closed = False
def _deliver(self, frame: CameraFrame) -> None:
self._slot.put(frame)
def read(self, timeout: float = 5.0) -> Optional[CameraFrame]:
"""Wait for a frame newer than the previous read.
Args:
timeout: Maximum blocking time in seconds.
Returns:
The next [`CameraFrame`][sim2bot.client.CameraFrame], or ``None`` when
the timeout expires.
Notes:
``read()`` returns the next fresh slot value even if it is older than
``max_age``. Iterator mode performs the age filter automatically.
"""
frame, self._last_seq = self._slot.get(self._last_seq, timeout)
return frame
def __iter__(self) -> Iterator[CameraFrame]:
return self
def __next__(self) -> CameraFrame:
while not self._closed:
frame = self.read()
if frame is None:
continue
# Skip a frame that's already stale (a fresher one is on the way).
if frame.age > self._max_age:
continue
return frame
raise StopIteration
def close(self) -> None:
"""Unsubscribe this camera and make the operation idempotently closed.
Returns:
Calling the method again after closure has no effect.
"""
if self._closed:
return
self._closed = True
self._robot._remove_camera(self.camera_id)
def __enter__(self) -> "CameraStream":
return self
def __exit__(self, *exc: Any) -> None:
self.close()
|