summaryrefslogtreecommitdiff
path: root/py/tests/cam.py
blob: 957eb117951b33ea55103ee8e2a8488f047a6e20 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
#!/usr/bin/python3

import sys
import selectors
import pykms
import argparse
import time

parser = argparse.ArgumentParser()
parser.add_argument("width", type=int)
parser.add_argument("height", type=int)
parser.add_argument("fourcc", type=str, nargs="?", default="YUVY")
args = parser.parse_args()

w = args.width
h = args.height
fmt = pykms.fourcc_to_pixelformat(args.fourcc)

print("Capturing in {}x{}".format(w, h))

card = pykms.Card()
res = pykms.ResourceManager(card)
conn = res.reserve_connector()
crtc = res.reserve_crtc(conn)
plane = res.reserve_overlay_plane(crtc, fmt)

mode = conn.get_default_mode()
modeb = mode.to_blob(card)

req = pykms.AtomicReq(card)
req.add(conn, "CRTC_ID", crtc.id)
req.add(crtc, {"ACTIVE": 1,
        "MODE_ID": modeb.id})
req.commit_sync(allow_modeset = True)

NUM_BUFS = 5

fbs = []
for i in range(NUM_BUFS):
    fb = pykms.DumbFramebuffer(card, w, h, fmt)
    fbs.append(fb)

vidpath = pykms.VideoDevice.get_capture_devices()[0]

vid = pykms.VideoDevice(vidpath)
cap = vid.capture_streamer
cap.set_port(0)
cap.set_format(fmt, w, h)
cap.set_queue_size(NUM_BUFS)

for fb in fbs:
    cap.queue(fb)

cap.stream_on()


def readvid(conn, mask):
    fb = cap.dequeue()

    if card.has_atomic:
        plane.set_props({
            "FB_ID": fb.id,
            "CRTC_ID": crtc.id,
            "SRC_W": fb.width << 16,
            "SRC_H": fb.height << 16,
            "CRTC_W": fb.width,
            "CRTC_H": fb.height,
        })
    else:
        crtc.set_plane(plane, fb, 0, 0, fb.width, fb.height,
            0, 0, fb.width, fb.height)

    cap.queue(fb)

def readkey(conn, mask):
    #print("KEY EVENT");
    sys.stdin.readline()
    exit(0)

sel = selectors.DefaultSelector()
sel.register(cap.fd, selectors.EVENT_READ, readvid)
sel.register(sys.stdin, selectors.EVENT_READ, readkey)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)