robosim.network.channel

channel.py Thread-safe data structure that acts as a channel between networking (outside) and simulator (inside)

 1"""channel.py Thread-safe data structure that acts as a channel between networking (outside) and simulator (inside)"""
 2from dataclasses import dataclass, field
 3from queue import Queue
 4import threading
 5import socket
 6
 7@dataclass
 8class ObjToSimChannel:
 9    """Channel of two queues handling the communication between the main thread (sim) and the tcp handler of a robot"""
10    index: int # the global index of the channel. Should be unique across the simulator.
11    tcp2sim: Queue = field(default_factory=Queue) # user -> simulator. User can send many messages theoretically.
12    sim2tcp: Queue = field(default_factory=lambda: Queue(maxsize=1)) # sim -> user. One message at a time.
13    lock: threading.Lock = field(default_factory=threading.Lock)
14    client: socket.socket | None = None
15
16    def clear(self):
17        """clears the queues of this channel"""
18        with self.lock:
19            self.tcp2sim.queue.clear()
20            self.sim2tcp.queue.clear()
@dataclass
class ObjToSimChannel: 
 8@dataclass
 9class ObjToSimChannel:
10    """Channel of two queues handling the communication between the main thread (sim) and the tcp handler of a robot"""
11    index: int # the global index of the channel. Should be unique across the simulator.
12    tcp2sim: Queue = field(default_factory=Queue) # user -> simulator. User can send many messages theoretically.
13    sim2tcp: Queue = field(default_factory=lambda: Queue(maxsize=1)) # sim -> user. One message at a time.
14    lock: threading.Lock = field(default_factory=threading.Lock)
15    client: socket.socket | None = None
16
17    def clear(self):
18        """clears the queues of this channel"""
19        with self.lock:
20            self.tcp2sim.queue.clear()
21            self.sim2tcp.queue.clear()

Channel of two queues handling the communication between the main thread (sim) and the tcp handler of a robot

ObjToSimChannel( index: int, tcp2sim: queue.Queue = <factory>, sim2tcp: queue.Queue = <factory>, lock: <built-in function allocate_lock> = <factory>, client: socket.socket | None = None)
index: int
tcp2sim: queue.Queue
sim2tcp: queue.Queue
lock: <built-in function allocate_lock>
client: socket.socket | None = None
def clear(self): 
17    def clear(self):
18        """clears the queues of this channel"""
19        with self.lock:
20            self.tcp2sim.queue.clear()
21            self.sim2tcp.queue.clear()

clears the queues of this channel