|
3 | 3 | import threading |
4 | 4 | from dataclasses import dataclass, field |
5 | 5 | from time import sleep |
| 6 | +from typing import Any |
6 | 7 |
|
7 | 8 | import numpy as np |
8 | 9 | from rcs._core.common import Pose |
@@ -93,11 +94,10 @@ def __init__(self, config: "QuestConfig", sim: Sim | None = None): |
93 | 94 | self._step_env = False |
94 | 95 | self._set_frame = {key: Pose() for key in self.controller_names} |
95 | 96 | self._video_stream_manager = None |
96 | | - self._video_streamers: dict[str, object] = {} |
97 | | - # if self.config.simulation: |
98 | | - # self._publisher = MujocoPublisher(self.sim.model, self.sim.data, self.config.mq3_addr, visible_geoms_groups=list(range(1, 3))) |
| 97 | + self._video_streamers: dict[str, Any] = {} |
99 | 98 | if not self.config.simulation: |
100 | 99 | self._publisher = FakeSimPublisher(FakeSimScene(), self.config.mq3_addr) |
| 100 | + # Not working code for digital twin: |
101 | 101 | # robot_cfg = default_sim_robot_cfg("fr3_empty_world") |
102 | 102 | # sim_cfg = SimConfig() |
103 | 103 | # sim_cfg.async_control = True |
@@ -200,9 +200,11 @@ def set_camera(self, observation: dict) -> None: |
200 | 200 | if camera_name not in self._video_streamers: |
201 | 201 | height, width = frame.shape[:2] |
202 | 202 | stream_topic = self._get_stream_topic_name(camera_name) |
| 203 | + assert self._video_stream_manager is not None |
203 | 204 | self._video_streamers[camera_name] = self._video_stream_manager.create_streamer( |
204 | 205 | stream_topic, width, height |
205 | 206 | ) |
| 207 | + assert self._video_streamers[camera_name] is not None |
206 | 208 | self._video_streamers[camera_name].update_cv_image(frame[:, :, ::-1]) |
207 | 209 |
|
208 | 210 | @staticmethod |
|
0 commit comments