def start_axis_video()

in src/rosbridge/timeseq.py [0:0]

• 70 lines of code
• 11 McCabe index (conditional complexity)

    def start_axis_video(self, timeseq_name, daemon_endpoint, resolution, ipaddr, auth_header, remote_traces_dir, local_link_prefix):
        self.want_axis_video = True

        self.ctx = zmq.Context()
        self.sock = self.ctx.socket(zmq.DEALER)
        logger.info('Connecting to %s...', daemon_endpoint)
        self.sock.connect(daemon_endpoint)
        logger.info('Connected to %s', daemon_endpoint)

        self.add_schema('VideoFrame',
                        ('url', 'string'))
        self.add_channel(timeseq_name, 'VideoFrame')

        def thread_main():
            delta = None
            rpc_count = 0
            while True:
                want = self.want_axis_video
                req = {
                    '__type': 'AxisCameraDaemonReq',
                    'traceSpec': {
                        '__type': 'VideoTraceSpec',
                        'traceDir': path.join(remote_traces_dir, self.trace_name),
                        'traceName': self.trace_name,
                        'timeseqName': timeseq_name,
                    },
                    'cameraConfig': {
                        '__type': 'AxisCameraConfig',
                        'ipaddr': ipaddr,
                        'url': '/axis-cgi/mjpg/video.cgi?compression=30&rotation=0&resolution=' + resolution,
                        'authHeader': auth_header,
                    },
                    'txTime': time.time(),
                    'recordFor': 10.0 if want else 0.0,
                }
                rpc_id = 'rpc%d' % rpc_count
                rpc_count += 1
                tx = ['record', rpc_id, ujson.dumps(req, escape_forward_slashes=False)]
                if 1: logger.info('camera %s < %s', daemon_endpoint, tx)
                self.sock.send_multipart(tx, flags=0, copy=True, track=False)

                rx = self.sock.recv_multipart(flags=0, copy=True, track=False)
                rx_time = time.time()
                if 1: logger.info('%s > %s %s', daemon_endpoint, rx[1], rx[2])
                rpc_id2 = rx[0]
                rpc_err = ujson.loads(rx[1])
                rpc_result = ujson.loads(rx[2])
                if rpc_err:
                    logger.info('%s > error %s', daemon_endpoint, rpc_err)
                    break

                min_delta = rpc_result['txTime'] - rx_time
                max_delta = rpc_result['txTime'] - rpc_result['reqTxTime']
                if delta is None:
                    delta = (min_delta + max_delta) * 0.5
                delta = max(min_delta, min(max_delta, delta))
                if 0: logger.info("timing %0.6f %+0.6f %+0.6f min_delta=%+0.6f max_delta=%+0.6f delta=%+0.6f",
                    rpc_result['reqTxTime'], rpc_result['txTime'], rx_time, min_delta, max_delta, delta)

                rep_times = rpc_result['times']
                rep_samples = rpc_result['samples']
                for ts, sample in zip(rep_times, rep_samples):
                    self.add(ts - delta, timeseq_name, sample)
                rep_chunks = rpc_result['chunks']
                for chunk_ts in rep_chunks:
                    chunk_fn = 'chunk_%s_%.0f.video' % (timeseq_name, chunk_ts)
                    chunk_path = path.join(self.trace_dir, chunk_fn)
                    src_file = path.join(local_link_prefix, self.trace_name, chunk_fn)
                    logger.info('Create symlink(%s, %s)  llp=%s rtd=%s', src_file, chunk_path, local_link_prefix, remote_traces_dir)
                    os.symlink(src_file, chunk_path)

                if not want: break
                time.sleep(0.1)


        thr = threading.Thread(target = thread_main)
        thr.daemon = True
        thr.start()
        self.camera_threads.append(thr)