1) cat > /tmp/ox02f10_live_1080p60.py <<'PY'
#!/usr/bin/env python3

import gi
import numpy as np
import time

gi.require_version("Gst", "1.0")
from gi.repository import Gst, GLib

Gst.init(None)

W = 1920
H = 1080
FPS = 60

Y_SIZE = W * H
UV_SIZE = Y_SIZE // 2
NV12_SIZE = Y_SIZE + UV_SIZE

FRAME_DURATION = Gst.SECOND // FPS

frame_count = 0
start_time = None

# Constant neutral chroma for grayscale NV12.
UV = bytes([128]) * UV_SIZE


capture_desc = (
    "v4l2src device=/dev/video3 io-mode=mmap ! "
    "video/x-bayer,"
    "width=1920,"
    "height=1080,"
    "framerate=60/1,"
    "format=rggb12 ! "
    "appsink name=cam "
    "emit-signals=true "
    "sync=false "
    "max-buffers=1 "
    "drop=true"
)


display_desc = (
    "appsrc name=src "
    "is-live=true "
    "format=time "
    "do-timestamp=false "
    "block=false "
    "caps=video/x-raw,"
    "format=NV12,"
    "width=1920,"
    "height=1080,"
    "framerate=60/1 ! "
    "queue max-size-buffers=2 leaky=downstream ! "
    "kmssink "
    "driver-name=tidss "
    "connector-id=40 "
    "plane-id=41 "
    "force-modesetting=true "
    "sync=false"
)


capture = Gst.parse_launch(capture_desc)
display = Gst.parse_launch(display_desc)

appsink = capture.get_by_name("cam")
appsrc = display.get_by_name("src")


def on_sample(sink):
    global frame_count, start_time

    sample = sink.emit("pull-sample")
    if sample is None:
        return Gst.FlowReturn.ERROR

    buf = sample.get_buffer()

    ok, info = buf.map(Gst.MapFlags.READ)
    if not ok:
        print("ERROR: RAW buffer map failed")
        return Gst.FlowReturn.ERROR

    try:
        raw = np.frombuffer(
            info.data,
            dtype="<u2",
            count=W * H
        ).reshape(H, W)

        #
        # OX02F10 RAW12:
        # bits 11:0 contain pixel data.
        #
        raw12 = raw & 0x0fff

        #
        # Fast RAW12 -> 8-bit grayscale.
        #
        # >>3 rather than >>4 gives a little more
        # brightness for the current exposure.
        #
        y = np.minimum(raw12 >> 3, 255).astype(np.uint8)

    finally:
        buf.unmap(info)

    #
    # Construct NV12:
    #
    # Y  = actual camera grayscale
    # UV = neutral chroma (128,128)
    #
    payload = y.tobytes() + UV

    outbuf = Gst.Buffer.new_allocate(None, NV12_SIZE, None)
    outbuf.fill(0, payload)

    outbuf.pts = frame_count * FRAME_DURATION
    outbuf.dts = outbuf.pts
    outbuf.duration = FRAME_DURATION

    ret = appsrc.emit("push-buffer", outbuf)

    frame_count += 1

    if start_time is None:
        start_time = time.monotonic()

    if frame_count % 120 == 0:
        elapsed = time.monotonic() - start_time
        fps = frame_count / elapsed

        print(
            "Frames: %d   Average processing FPS: %.2f"
            % (frame_count, fps),
            flush=True
        )

    return ret


appsink.connect("new-sample", on_sample)


def bus_message(bus, message, loop):
    t = message.type

    if t == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("GStreamer ERROR:", err)
        print(debug)
        loop.quit()

    elif t == Gst.MessageType.EOS:
        print("EOS")
        loop.quit()


loop = GLib.MainLoop()

for pipe in (capture, display):
    bus = pipe.get_bus()
    bus.add_signal_watch()
    bus.connect("message", bus_message, loop)


display.set_state(Gst.State.PLAYING)
capture.set_state(Gst.State.PLAYING)

print("")
print("OX02F10 LIVE CAMERA")
print("===================")
print("Input  : RAW12 RGGB 1920x1080 @ 60 fps")
print("Output : NV12       1920x1080 @ 60 fps")
print("Mode   : Full-resolution grayscale")
print("")
print("Move the camera: image should move on HDMI.")
print("Press Ctrl+C to stop.")
print("")

try:
    loop.run()

except KeyboardInterrupt:
    print("\nStopping...")


capture.set_state(Gst.State.NULL)

appsrc.emit("end-of-stream")

display.set_state(Gst.State.NULL)

print("Stopped.")
PY

chmod +x /tmp/ox02f10_live_1080p60.py