diff --git a/tools/camerastream/compressed_vipc.py b/tools/camerastream/compressed_vipc.py index cab11493f2..0e524c5a04 100755 --- a/tools/camerastream/compressed_vipc.py +++ b/tools/camerastream/compressed_vipc.py @@ -13,8 +13,16 @@ from cereal.visionipc import VisionIpcServer, VisionStreamType W, H = 1928, 1208 V4L2_BUF_FLAG_KEYFRAME = 8 -def decoder(addr, sock_name, vipc_server, vst, nvidia): - print("start decoder for %s" % sock_name) +ENCODE_SOCKETS = { + VisionStreamType.VISION_STREAM_ROAD: "roadEncodeData", + VisionStreamType.VISION_STREAM_WIDE_ROAD: "wideRoadEncodeData", + VisionStreamType.VISION_STREAM_DRIVER: "driverEncodeData", +} + +def decoder(addr, vipc_server, vst, nvidia, debug=False): + sock_name = ENCODE_SOCKETS[vst] + if debug: + print("start decoder for %s" % sock_name) if nvidia: os.environ["NV_LOW_LATENCY"] = "3" # both bLowLatency and CUVID_PKT_ENDOFPICTURE sys.path += os.environ["LD_LIBRARY_PATH"].split(":") @@ -40,11 +48,12 @@ def decoder(addr, sock_name, vipc_server, vst, nvidia): msgs = messaging.drain_sock(sock, wait_for_one=True) for evt in msgs: evta = getattr(evt, evt.which()) - if evta.idx.encodeId != 0 and evta.idx.encodeId != (last_idx+1): + if debug and evta.idx.encodeId != 0 and evta.idx.encodeId != (last_idx+1): print("DROP PACKET!") last_idx = evta.idx.encodeId if not seen_iframe and not (evta.idx.flags & V4L2_BUF_FLAG_KEYFRAME): - print("waiting for iframe") + if debug: + print("waiting for iframe") continue time_q.append(time.monotonic()) network_latency = (int(time.time()*1e9) - evta.unixTimestampNanos)/1e6 @@ -62,14 +71,16 @@ def decoder(addr, sock_name, vipc_server, vst, nvidia): if nvidia: rawSurface = nvDec.DecodeSurfaceFromPacket(np.frombuffer(evta.data, dtype=np.uint8)) if rawSurface.Empty(): - print("DROP SURFACE") + if debug: + print("DROP SURFACE") continue convSurface = conv_yuv.Execute(rawSurface, cc1) nvDwn_yuv.DownloadSingleSurface(convSurface, img_yuv) else: frames = codec.decode(av.packet.Packet(evta.data)) if len(frames) == 0: - print("DROP SURFACE") + if debug: + print("DROP SURFACE") continue assert len(frames) == 1 img_yuv = frames[0].to_ndarray(format=av.video.format.VideoFormat('yuv420p')).flatten() @@ -83,34 +94,45 @@ def decoder(addr, sock_name, vipc_server, vst, nvidia): pc_latency = (time.monotonic()-time_q[0])*1000 time_q = time_q[1:] - print("%2d %4d %.3f %.3f roll %6.2f ms latency %6.2f ms + %6.2f ms + %6.2f ms = %6.2f ms" % (len(msgs), evta.idx.encodeId, evt.logMonoTime/1e9, evta.idx.timestampEof/1e6, frame_latency, process_latency, network_latency, pc_latency, process_latency+network_latency+pc_latency ), len(evta.data), sock_name) + if debug: + print("%2d %4d %.3f %.3f roll %6.2f ms latency %6.2f ms + %6.2f ms + %6.2f ms = %6.2f ms" % (len(msgs), evta.idx.encodeId, evt.logMonoTime/1e9, evta.idx.timestampEof/1e6, frame_latency, process_latency, network_latency, pc_latency, process_latency+network_latency+pc_latency ), len(evta.data), sock_name) + +class CompressedVipc: + def __init__(self, addr, vision_streams, nvidia=False, debug=False): + self.vipc_server = VisionIpcServer("camerad") + for vst in vision_streams: + self.vipc_server.create_buffers(vst, 4, False, W, H) + self.vipc_server.start_listener() -def main(addr, cams, nvidia=False): - vipc_server = VisionIpcServer("camerad") - for vst in cams.values(): - vipc_server.create_buffers(vst, 4, False, W, H) - vipc_server.start_listener() + self.procs = [] + for vst in vision_streams: + p = multiprocessing.Process(target=decoder, args=(addr, self.vipc_server, vst, nvidia, debug)) + p.start() + self.procs.append(p) - procs = [] - for k, v in cams.items(): - p = multiprocessing.Process(target=decoder, args=(addr, k, vipc_server, v, nvidia)) - p.start() - procs.append(p) + def join(self): + for p in self.procs: + p.join() - for p in procs: - p.join() + def kill(self): + for p in self.procs: + p.terminate() + self.join() if __name__ == "__main__": parser = argparse.ArgumentParser(description="Decode video streams and broadcast on VisionIPC") parser.add_argument("addr", help="Address of comma three") parser.add_argument("--nvidia", action="store_true", help="Use nvidia instead of ffmpeg") parser.add_argument("--cams", default="0,1,2", help="Cameras to decode") + parser.add_argument("--silent", action="store_true", help="Suppress debug output") args = parser.parse_args() - all_cams = [ - ("roadEncodeData", VisionStreamType.VISION_STREAM_ROAD), - ("wideRoadEncodeData", VisionStreamType.VISION_STREAM_WIDE_ROAD), - ("driverEncodeData", VisionStreamType.VISION_STREAM_DRIVER), + vision_streams = [ + VisionStreamType.VISION_STREAM_ROAD, + VisionStreamType.VISION_STREAM_WIDE_ROAD, + VisionStreamType.VISION_STREAM_DRIVER, ] - cams = dict([all_cams[int(x)] for x in args.cams.split(",")]) - main(args.addr, cams, args.nvidia) + + vsts = [vision_streams[int(x)] for x in args.cams.split(",")] + cvipc = CompressedVipc(args.addr, vsts, args.nvidia, debug=(not args.silent)) + cvipc.join()