Add Dex1 grippers and wrist cameras; name end effectors after the hardware

9051116 verified about 1 month ago

2.16 kB

#!/usr/bin/env python3
import sys
from pathlib import Path
import time
# Add sim module to path
sys.path.insert(0, str(Path(__file__).parent))

import yaml
from sim.simulator_factory import SimulatorFactory, init_channel
from sim.model_config import select_end_effector

def main(n_envs=1, use_async_envs: bool = False, 
             publish_images=True, camera_port=5555, cameras=None, **kwargs):
    
    # Load config
    config_path = Path(__file__).parent / "config.yaml"
    with open(config_path) as f:
        config = yaml.safe_load(f)
    config = select_end_effector(config, kwargs.get("end_effector"))
    
    # Override config with default values
    enable_offscreen = publish_images or config.get("ENABLE_OFFSCREEN", False)

    # Configure cameras if requested
    camera_configs = {}
    if enable_offscreen:
        camera_list = config["CAMERAS"] if cameras is None else cameras
        for cam_name in camera_list:
            camera_configs[cam_name] = {"height": 480, "width": 640}
        print(f"📷 Cameras: {', '.join(camera_list)} → ZMQ port {camera_port}")
    
    print("="*60)
    
    # Initialize DDS channel
    init_channel(config=config)
    
    # Create simulator
    sim = SimulatorFactory.create_simulator(
        config=config,
        env_name="default",
        onscreen=kwargs.get("onscreen", config.get("ENABLE_ONSCREEN", True)),
        offscreen=enable_offscreen,
        camera_configs=camera_configs,
    )
    
    # Start simulator
    print("\nSimulator running. Press Ctrl+C to exit.")
    if enable_offscreen and publish_images:
        print(f"Camera images publishing on tcp://localhost:{camera_port}")
    try:
        if publish_images:
            sim.start_image_publish_subprocess(
                start_method="spawn",
                camera_port=camera_port,
            )
            time.sleep(1)
        sim.start()
    except KeyboardInterrupt:
        print("+++++Simulator interrupted by user.")
    except Exception as e:
        print(f"++++error in simulator: {e} ++++")
    finally:
        print("++++closing simulator ++++")
        sim.close()

if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	import sys
	from pathlib import Path
	import time
	# Add sim module to path
	sys.path.insert(0, str(Path(__file__).parent))

	import yaml
	from sim.simulator_factory import SimulatorFactory, init_channel
	from sim.model_config import select_end_effector

	def main(n_envs=1, use_async_envs: bool = False,
	publish_images=True, camera_port=5555, cameras=None, **kwargs):

	# Load config
	config_path = Path(__file__).parent / "config.yaml"
	with open(config_path) as f:
	config = yaml.safe_load(f)
	config = select_end_effector(config, kwargs.get("end_effector"))

	# Override config with default values
	enable_offscreen = publish_images or config.get("ENABLE_OFFSCREEN", False)

	# Configure cameras if requested
	camera_configs = {}
	if enable_offscreen:
	camera_list = config["CAMERAS"] if cameras is None else cameras
	for cam_name in camera_list:
	camera_configs[cam_name] = {"height": 480, "width": 640}
	print(f"📷 Cameras: {', '.join(camera_list)} → ZMQ port {camera_port}")

	print("="*60)

	# Initialize DDS channel
	init_channel(config=config)

	# Create simulator
	sim = SimulatorFactory.create_simulator(
	config=config,
	env_name="default",
	onscreen=kwargs.get("onscreen", config.get("ENABLE_ONSCREEN", True)),
	offscreen=enable_offscreen,
	camera_configs=camera_configs,
	)

	# Start simulator
	print("\nSimulator running. Press Ctrl+C to exit.")
	if enable_offscreen and publish_images:
	print(f"Camera images publishing on tcp://localhost:{camera_port}")
	try:
	if publish_images:
	sim.start_image_publish_subprocess(
	start_method="spawn",
	camera_port=camera_port,
	)
	time.sleep(1)
	sim.start()
	except KeyboardInterrupt:
	print("+++++Simulator interrupted by user.")
	except Exception as e:
	print(f"++++error in simulator: {e} ++++")
	finally:
	print("++++closing simulator ++++")
	sim.close()

	if __name__ == "__main__":
	main()