style: fix ruff SIM105/format lints in unitree_g1 scripts

Use contextlib.suppress(zmq.Again) instead of try/except/pass in the
onboard state publisher and teleop client, plus ruff-format/isort
autofixes across the unitree_g1 helpers and dataset_reader imports.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Martino Russi
2026-07-24 12:08:15 +02:00
parent 39fe436248
commit a517594722
6 changed files with 29 additions and 19 deletions
+4 -1
View File
@@ -137,7 +137,10 @@ class ImageServer:
last_err = e last_err = e
logger.warning( logger.warning(
"Camera %s open attempt %d/%d failed: %s", "Camera %s open attempt %d/%d failed: %s",
name, attempt, self.open_attempts, e, name,
attempt,
self.open_attempts,
e,
) )
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
camera.disconnect() camera.disconnect()
+1 -1
View File
@@ -19,9 +19,9 @@ from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from pathlib import Path from pathlib import Path
import datasets
import torch import torch
import datasets
from lerobot.configs import ( from lerobot.configs import (
DEFAULT_DEPTH_UNIT, DEFAULT_DEPTH_UNIT,
DEPTH_METER_UNIT, DEPTH_METER_UNIT,
@@ -44,6 +44,7 @@ Examples (on the robot):
""" """
import argparse import argparse
import contextlib
import json import json
import logging import logging
import os import os
@@ -177,8 +178,8 @@ def main() -> None:
logger.warning("E-STOP ('e'): going passive NOW.") logger.warning("E-STOP ('e'): going passive NOW.")
try: try:
robot._shutdown_event.set() # stop the 50Hz controller loop publishing robot._shutdown_event.set() # stop the 50Hz controller loop publishing
time.sleep(0.05) # let it finish its current cycle time.sleep(0.05) # let it finish its current cycle
robot._send_zero_torque() # motors limp; nothing overwrites it now robot._send_zero_torque() # motors limp; nothing overwrites it now
except Exception as e: # noqa: BLE001 except Exception as e: # noqa: BLE001
logger.warning("E-stop zero-torque failed: %s", e) logger.warning("E-stop zero-torque failed: %s", e)
os._exit(0) # immediate hard exit, no slow cleanup os._exit(0) # immediate hard exit, no slow cleanup
@@ -218,10 +219,8 @@ def main() -> None:
state["groot.rpy.roll"] = float(orient[0]) state["groot.rpy.roll"] = float(orient[0])
state["groot.rpy.pitch"] = float(orient[1]) state["groot.rpy.pitch"] = float(orient[1])
state["groot.rpy.yaw"] = float(orient[2]) state["groot.rpy.yaw"] = float(orient[2])
try: with contextlib.suppress(zmq.Again):
state_sock.send_json(state, zmq.NOBLOCK) state_sock.send_json(state, zmq.NOBLOCK)
except zmq.Again:
pass
time.sleep(max(0.0, period - (time.time() - t0))) time.sleep(max(0.0, period - (time.time() - t0)))
threading.Thread(target=publish_state, daemon=True).start() threading.Thread(target=publish_state, daemon=True).start()
@@ -151,11 +151,7 @@ def parse_camera_specs(spec: str, default_width: int, default_height: int) -> di
# a pixel format. Real device-path tail segments (e.g. "1.0-video-index0") # a pixel format. Real device-path tail segments (e.g. "1.0-video-index0")
# won't match these strict patterns. # won't match these strict patterns.
fourcc = None fourcc = None
if ( if len(tokens) >= 3 and wh_re.fullmatch(tokens[-2]) and fourcc_re.fullmatch(tokens[-1]):
len(tokens) >= 3
and wh_re.fullmatch(tokens[-2])
and fourcc_re.fullmatch(tokens[-1])
):
fourcc = tokens.pop().upper() fourcc = tokens.pop().upper()
width, height = default_width, default_height width, height = default_width, default_height
if len(tokens) >= 2 and wh_re.fullmatch(tokens[-1]): if len(tokens) >= 2 and wh_re.fullmatch(tokens[-1]):
@@ -32,6 +32,7 @@ Example (on the laptop):
""" """
import argparse import argparse
import contextlib
import logging import logging
import time import time
@@ -106,10 +107,8 @@ def main() -> None:
rc.lx = rc.ly = rc.rx = rc.ry = 0.0 rc.lx = rc.ly = rc.rx = rc.ry = 0.0
rc.button = [0] * 16 rc.button = [0] * 16
action = teleop.get_action() action = teleop.get_action()
try: with contextlib.suppress(zmq.Again):
sock.send_json(_to_jsonable(action), zmq.NOBLOCK) sock.send_json(_to_jsonable(action), zmq.NOBLOCK)
except zmq.Again:
pass
n += 1 n += 1
if n % 60 == 0: if n % 60 == 0:
axes = { axes = {
+17 -4
View File
@@ -70,8 +70,22 @@ else:
# teleoperator's RemoteController so the onboard path reads the physical Unitree remote # teleoperator's RemoteController so the onboard path reads the physical Unitree remote
# identically. # identically.
_REMOTE_BUTTON_MAP: list[str] = [ _REMOTE_BUTTON_MAP: list[str] = [
"RB", "LB", "start", "back", "RT", "LT", "", "", "RB",
"A", "B", "X", "Y", "up", "right", "down", "left", "LB",
"start",
"back",
"RT",
"LT",
"",
"",
"A",
"B",
"X",
"Y",
"up",
"right",
"down",
"left",
] ]
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -299,8 +313,7 @@ class UnitreeG1(Robot):
actual_hz = loop_count / (time.time() - last_log_time) actual_hz = loop_count / (time.time() - last_log_time)
eff = {k: round(float(controller_input.get(k, 0.0)), 3) for k in REMOTE_AXES} eff = {k: round(float(controller_input.get(k, 0.0)), 3) for k in REMOTE_AXES}
logger.info( logger.info(
f"Controller {actual_hz:.1f}Hz | eff_axes={eff} " f"Controller {actual_hz:.1f}Hz | eff_axes={eff} wireless={'ACTIVE' if wl else 'idle'}"
f"wireless={'ACTIVE' if wl else 'idle'}"
) )
loop_count = 0 loop_count = 0
last_log_time = time.time() last_log_time = time.time()