Mono preview - Alternate between dot projector and illumination LED

1#!/usr/bin/env python3
2
3import cv2
4import depthai as dai
5
6if 1:  # PoE config
7    fps = 30
8    res = dai.MonoCameraProperties.SensorResolution.THE_400_P
9    poolSize = 24  # default 3, increased to prevent desync
10else:  # USB
11    fps = 30
12    res = dai.MonoCameraProperties.SensorResolution.THE_720_P
13    poolSize = 8  # default 3, increased to prevent desync
14
15# Create pipeline
16pipeline = dai.Pipeline()
17
18# Define sources and outputs
19monoL = pipeline.create(dai.node.MonoCamera)
20monoR = pipeline.create(dai.node.MonoCamera)
21
22monoL.setCamera("left")
23monoL.setResolution(res)
24monoL.setFps(fps)
25monoL.setNumFramesPool(poolSize)
26monoR.setCamera("right")
27monoR.setResolution(res)
28monoR.setFps(fps)
29monoR.setNumFramesPool(poolSize)
30
31xoutDotL = pipeline.create(dai.node.XLinkOut)
32xoutDotR = pipeline.create(dai.node.XLinkOut)
33xoutFloodL = pipeline.create(dai.node.XLinkOut)
34xoutFloodR = pipeline.create(dai.node.XLinkOut)
35
36xoutDotL.setStreamName('dot-left')
37xoutDotR.setStreamName('dot-right')
38xoutFloodL.setStreamName('flood-left')
39xoutFloodR.setStreamName('flood-right')
40streams = ['dot-left', 'dot-right', 'flood-left', 'flood-right']
41
42# Script node for frame routing and IR dot/flood alternate
43script = pipeline.create(dai.node.Script)
44script.setProcessor(dai.ProcessorType.LEON_CSS)
45script.setScript("""
46    dotBright = 0.8
47    floodBright = 0.1
48    LOGGING = False  # Set `True` for latency/timings debugging
49
50    node.warn(f'IR drivers detected: {str(Device.getIrDrivers())}')
51
52    flagDot = False
53    while True:
54        # Wait first for a frame event, received at MIPI start-of-frame
55        event = node.io['event'].get()
56        if LOGGING: tEvent = Clock.now()
57
58        # Immediately reconfigure the IR driver.
59        # Note the logic is inverted, as it applies for next frame
60        Device.setIrLaserDotProjectorIntensity(0 if flagDot else dotBright)
61        Device.setIrFloodLightIntensity(floodBright if flagDot else 0)
62        if LOGGING: tIrSet = Clock.now()
63
64        # Wait for the actual frames (after MIPI capture and ISP proc is done)
65        frameL = node.io['frameL'].get()
66        if LOGGING: tLeft = Clock.now()
67        frameR = node.io['frameR'].get()
68        if LOGGING: tRight = Clock.now()
69
70        if LOGGING:
71            latIR      = (tIrSet - tEvent               ).total_seconds() * 1000
72            latEv      = (tEvent - event.getTimestamp() ).total_seconds() * 1000
73            latProcL   = (tLeft  - event.getTimestamp() ).total_seconds() * 1000
74            diffRecvRL = (tRight - tLeft                ).total_seconds() * 1000
75            node.warn(f'T[ms] latEv:{latEv:5.3f} latIR:{latIR:5.3f} latProcL:{latProcL:6.3f} '
76                    + f' diffRecvRL:{diffRecvRL:5.3f}')
77
78        # Sync checks
79        diffSeq = frameL.getSequenceNum() - event.getSequenceNum()
80        diffTsEv = (frameL.getTimestamp() - event.getTimestamp()).total_seconds() * 1000
81        diffTsRL = (frameR.getTimestamp() - frameL.getTimestamp()).total_seconds() * 1000
82        if diffSeq or diffTsEv or (abs(diffTsRL) > 0.8):
83            node.error(f'frame/event desync! Fr-Ev: {diffSeq} frames,'
84                    + f' {diffTsEv:.3f} ms; R-L: {diffTsRL:.3f} ms')
85
86        # Route the frames to their respective outputs
87        node.io['dotL' if flagDot else 'floodL'].send(frameL)
88        node.io['dotR' if flagDot else 'floodR'].send(frameR)
89
90        flagDot = not flagDot
91""")
92
93# Linking
94monoL.frameEvent.link(script.inputs['event'])
95monoL.out.link(script.inputs['frameL'])
96monoR.out.link(script.inputs['frameR'])
97
98script.outputs['dotL'].link(xoutDotL.input)
99script.outputs['dotR'].link(xoutDotR.input)
100script.outputs['floodL'].link(xoutFloodL.input)
101script.outputs['floodR'].link(xoutFloodR.input)
102
103# Connect to device and start pipeline
104with dai.Device(pipeline) as device:
105    queues = [device.getOutputQueue(name=s, maxSize=4, blocking=False) for s in streams]
106
107    while True:
108        for q in queues:
109            pkt = q.tryGet()
110            if pkt is not None:
111                name = q.getName()
112                frame = pkt.getCvFrame()
113                cv2.imshow(name, frame)
114
115        if cv2.waitKey(5) == ord('q'):
116            break