多重处理和实时数据访问问题

Question

我是多处理新手。 我正在从父脚本启动一个 child_process 。 child_process 是一个持续发送实时数据的传感器。在子脚本中，我将该数据发送到管道上，在需要时可以从父脚本中获取数据。 我意识到我的数据有延迟。我试图获取的数据不是来自传感器的实时数据，而是之前记录的数据。

还有其他方法可以处理实时数据吗？

子脚本中的函数：

我添加了 time.sleep（0.1，它稍微改善了问题，但不够准确）

def _send_data(self):
        time.sleep(1)
        offsets = self.calculate_offset()
        
        try:
            while 1:
               
                if self._conn:
                    zeroed_data = self.zero_data(offsets,self.get_data())
                    status = {
                        "Status": self._status,
                        "Timestamp": sR.millis(),
                        "Temperature": self._temperature,
                        }
                    status.update(zeroed_data)
                self._conn.send(status)
                time.sleep(0.1)

        except KeyboardInterrupt:
            # ctrl-C abort handling
            if self.conn:
                self.conn.close() 
            print('stopped')



def main_bota(child_conn,port):
    print('bota started')
    bota_sensor_1 = None
   
    try:
        bota_sensor_1 = BotaSerialSensor(port,child_conn);
        bota_sensor_1.run()
       
    except BotaSerialSensorError as expt:
        print('bota failed: ' + expt.message)
    
    except Exception as e:
        print(f"An unexpected error occured: {e}")
    
    finally:
        if bota_sensor_1 is not None:
            bota_sensor_1.close()
        sys.exit(1)
        

父脚本中的函数：

def robot_command(selected_tool, path):
    """send commands to robot according to the path.

    Args:
        selected_tool (np.array): selected tool matrix (4x4)
        path (list): List of transformation matrices defining the robots path.
    """

    worldToToolStart = sR.getWorldToTool(s, BUFFER_SIZE, selected_tool)
    totNumbPos = len(path)
    totNumb = 0
   
    for i in range(totNumbPos):
        totNumb += 1
        toolToNewTarget = path[i]
        worldToNewTarget = np.matmul(worldToToolStart, toolToNewTarget)
        worldToFlangeCommand = np.matmul(
            worldToNewTarget, sR.homInv(selectedTool))

        # Formulate the command based on robot mode
        if robotMode == 'SmartServo':
            poseStringForRobot = sR.make_cmd(
                "MovePTPHomRowWiseMinChangeRT", worldToFlangeCommand)

        else:
            poseStringForRobot = sR.make_cmd(
                "MovePTPHomRowWiseMinChange", worldToFlangeCommand)


        # Send command to robot
        sR.command(s, BUFFER_SIZE, poseStringForRobot)
      
        # get sensor data
        sensordata = parent_conn.recv()
       
        # Check status of Bota
        #status_bota(sensordata['Status'])
       
        # get robot data dict and combine with sensor data dict
        robotdata = fetch_robot_data()
        combined_data = {**sensordata, **robotdata}
        
        #append combined_data to global data
        data.append(combined_data)
        print(i)
        print(robotdata["timestamp"])
        print(sensordata["Timestamp"])
        # Wait until position is reached (can be adjusted...)
        sR.delay(delayBetweenPoses)

连接传感器和断开连接的功能

def connect_bota(port):
    global p,parent_conn
    parent_conn, child_conn = Pipe()
    p = Process(target=main_bota, args = (child_conn,port))
    p.start()
    print('process bota started')


def disconnect_bota():
    p.terminate()
    p.join()
    print('process bota stopped')



def main():
    config()
    name = generate_experiment_name()
    print(name)
    display_config()
    connect_to_robot()
    connect_bota(port='COM3')
    
    # generate selected path
    path = generate_path(selectedPath_str)

    # (go to supportRobot file to adapt initialization parameters (velocities, damping, ...))
    sR.initRobot(s, BUFFER_SIZE, robotMode)
    print('initRobot')

    # Delay for 3s
    delay(2)

    # Send command to robot
    print('send command')
    robot_command(selectedTool, path)

    # End of operations
    print('STOP in 2s')
    delay(2)
    disconnect_bota()
    s.close()
    print("STOPPED")
    

Answer 1

我使用多处理的 Manger() 而不是 Pipe()

发件人

`import time 
 def send_data(shared_dict, lock):
    count = 0
    while True:
        with lock:
           big_dict ={
               'time': time. Time(),
               'data': count}
           shared_dict["sensordata"] = big_dict
        count += 1
        time.Sleep(0)`

    
    

接收器

`from multiprocessing import Process, Manager, Lock
import time 
from sender import send_data

if __name__ == "__main__":
    manager = Manager()
    shared_dict = manager.dict()
    lock = Lock()

    p = Process(target = send_data, args =(shared_dict, lock))
    p.start()
    
    try:
        for _ in range(20):
            time.sleep(5)
            with lock:
                sensor = shared_dict.get("sensordata", "No data yet")
                timesent = sensor["time"]
                data = sensor["data"]
                print(f"sent: {timesent} receiv: {time.time()} data: {data}")
         
        p.terminate()
    
    except KeyboardInterrupt:
        p.terminate()
        print("Ctrl + C")`