Skip to content

sim2bot.Robot.connect

connect() -> 'Robot'

Open the configured bridge connection and start reader threads.

If auto_bridge=True, this first reuses a healthy local bridge or starts one. If wait_for_sim=True, the call does not return until a browser announces at least one robot or the configured timeout expires.

Returns:

  • 'Robot' –

    This client instance, enabling Robot(...).connect() chaining.

Raises:

  • TimeoutError –

    If waiting for a browser simulator times out.

  • RuntimeError –

    If an automatically started bridge cannot become healthy.

  • ConnectionClosed –

    If the WebSocket closes while sending its initial hello.

  • OSError –

    If the configured network endpoint cannot be opened.

Source code in sim2bot/client.py 
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
def connect(self) -> "Robot":
    """Open the configured bridge connection and start reader threads.

    If ``auto_bridge=True``, this first reuses a healthy local bridge or starts
    one. If ``wait_for_sim=True``, the call does not return until a browser
    announces at least one robot or the configured timeout expires.

    Returns:
        This client instance, enabling ``Robot(...).connect()`` chaining.

    Raises:
        TimeoutError: If waiting for a browser simulator times out.
        RuntimeError: If an automatically started bridge cannot become healthy.
        ConnectionClosed: If the WebSocket closes while sending its initial hello.
        OSError: If the configured network endpoint cannot be opened.
    """
    if self._auto_bridge:
        from .bridge import ensure_bridge

        info = ensure_bridge(
            self.url,
            udp_port=self._udp_dest[1],
            timeout=self._bridge_timeout,
        )
        self.url = info.url
        self.video_url = self.video_url or info.video_url
    if self.transport == "udp":
        self._udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self._udp_sock.settimeout(1.0)
        # A first datagram registers us with the bridge so telemetry flows
        # back to this socket (the bridge replies to the sender address).
        self._send({"type": "describe"})
        self._spawn(self._control_reader_udp, name="sim2bot-control-udp")
    else:
        self._conn = ws_connect(
            self.url,
            open_timeout=self._connect_timeout,
            additional_headers=self._auth_headers(),
        )
        self._send({"type": "hello", "role": "controller"})
        self._spawn(self._control_reader_ws, name="sim2bot-control")
    self._spawn(self._renew_loop, name="sim2bot-renew")
    if self._wait_for_sim_on_connect:
        self.wait_for_sim(
            timeout=self._wait_for_sim_timeout,
            poll_interval=self._wait_for_sim_poll_interval,
        )
    return self

Practical examples

Examples use objects discovered from the current scene rather than hard-coded robot metadata.

Connect manually

from sim2bot import Robot

sim = Robot(auto_bridge=True)
sim.connect()
try:
    print(sim.wait_for_sim(timeout=15))
finally:
    sim.close()

Connect with a context manager

from sim2bot import Robot

with Robot(auto_bridge=True, wait_for_sim=True) as sim:
    arm = sim.describe()[0]
    print(arm.name)

Guidance

Usage tip

Prefer with Robot(...) as sim; call connect() directly when lifecycle ownership belongs to a larger application.

Common error

Calling command methods before connect() raises RuntimeError: not connected.

See also