#!/usr/bin/env python3 """Read the g2g timecode twice out of one screenshot -- the real pattern drawn by g2g_pattern.py at the monitor's top-left, and the viewer's rendering of it inside the latenz_app window. Their difference is the end-to-end latency of the whole loop (compose -> capture -> encode -> network -> decode -> present). Usage: g2g_measure.py [samples] [interval_s] Both copies are located by the pattern's saturated-green frame, which is what makes this robust: the first version computed where the viewer's copy *should* be from the window rect (aspect-fit 16:9 inside the content box) and refined with a brightest-patch search. That is wrong twice over -- the app bar means the video does not start at the window's top edge, and the brightest patch near the guess is window chrome, not the pattern. It reported a timecode that never advanced, which reads as a frozen stream rather than as a mislocated crop. """ import io import os import subprocess import sys import time from PIL import Image try: from Xlib import X, display as xdisplay except ImportError: # python3-xlib absent: fall back to full-root `import` X = xdisplay = None BITS = 12 BLOCK = 96 FRAME = 16 PW = BITS * BLOCK + 2 * FRAME PH = BLOCK + 2 * FRAME ASPECT = PW / PH # The measurement box's X display. Not a constant: this rig has run on :1 and # on :0, and a wrong value fails as "pattern not found" rather than as a # display error, which reads like the viewer is not up. DISPLAY = os.environ.get("G2G_DISPLAY", os.environ.get("DISPLAY", ":0")) def _import_grab(box=None): cmd = ["import", "-display", DISPLAY, "-window", "root", "-silent"] if box: x, y, w, h = box cmd += ["-crop", f"{w}x{h}+{x}+{y}", "+repage"] cmd.append("png:-") png = subprocess.run(cmd, capture_output=True, check=True).stdout return Image.open(io.BytesIO(png)).convert("RGB") class Grabber: """Reads a rectangle of the root window. Uses XGetImage on just that rectangle when python3-xlib is available. This is not an optimisation, it is a correctness fix: grabbing the whole 5120x1440 root every sample stalls the compositor for ~200 ms, which beam-host's capture then inherits. Measured -- the viewer's freezeCount sat at 0 for the first 24 s of a session and then climbed by one every two seconds for exactly as long as the measurement ran, and stopped when it stopped. The instrument was injecting the stalls whose consequences it was reading back as latency. """ def __init__(self): self.dpy = None if xdisplay is not None: try: self.dpy = xdisplay.Display(DISPLAY) self.root = self.dpy.screen().root except Exception: self.dpy = None def full(self): """Whole root, once, to locate the patterns.""" if self.dpy is None: return _import_grab() g = self.root.get_geometry() return self.rect(0, 0, g.width, g.height) def rect(self, x, y, w, h): if self.dpy is None: return _import_grab((x, y, w, h)) raw = self.root.get_image(x, y, w, h, X.ZPixmap, 0xFFFFFFFF) # ZPixmap on a 24/32-bit visual is BGRX per pixel. return Image.frombytes("RGB", (w, h), raw.data, "raw", "BGRX") def green_boxes(img, min_w=120): """Bounding boxes of the green-framed patterns, left to right. The frame is hollow, so a green run of the pattern's full width exists only on its top and bottom bands -- the sides are 16px runs that fall below min_w. Collect the wide runs, then pair bands that share an x-extent into one box. (Merging only vertically-adjacent runs, the obvious first cut, yields two 1184x16 slivers per pattern and every one of them fails the aspect filter, so nothing is ever found.) """ w, h = img.size px = img.load() runs = [] # (y, x0, x1) for y in range(h): x = 0 while x < w: r, g, b = px[x, y] if g > 110 and r < g - 60 and b < g - 60: x0 = x while x < w: r, g, b = px[x, y] if not (g > 110 and r < g - 60 and b < g - 60): break x += 1 if x - x0 >= min_w: runs.append((y, x0, x)) else: x += 1 groups = [] # [x0, y0, x1, y1] for y, x0, x1 in runs: for gx in groups: # Same pattern if the runs line up horizontally within 10% of width. if abs(gx[0] - x0) < 0.1 * (x1 - x0) and abs(gx[2] - x1) < 0.1 * (x1 - x0): gx[0] = min(gx[0], x0) gx[2] = max(gx[2], x1) gx[3] = y break else: groups.append([x0, y, x1, y]) # Keep only boxes shaped like the pattern; the aspect is 9.25:1, generous # tolerance because the viewer's copy goes through a 2560->1920 downscale # and then the window's own scaling. out = [] for x0, y0, x1, y1 in groups: bw, bh = x1 - x0, y1 - y0 + 1 if bh < 8: continue if 0.6 * ASPECT < bw / bh < 1.6 * ASPECT: out.append((x0, y0, bw, bh)) out.sort(key=lambda b: (b[1], b[0])) return out def decode(img, x, y, bw, bh): """Read BITS blocks from a pattern whose green frame's top-left is x,y.""" frame = bh * FRAME / PH block_w = (bw - 2 * frame) / BITS block_h = bh - 2 * frame g = img.convert("L") bits = 0 for i in range(BITS): cx = int(x + frame + (i + 0.5) * block_w) cy = int(y + frame + 0.5 * block_h) if not (0 <= cx < g.width and 0 <= cy < g.height): return None # Average a small patch to survive codec ringing. rad = max(2, int(block_w / 6)) patch = g.crop((cx - rad, cy - rad, cx + rad, cy + rad)) v = sum(patch.getdata()) / (patch.width * patch.height) bits = (bits << 1) | (1 if v > 127 else 0) return bits def main(): n = int(sys.argv[1]) if len(sys.argv) > 1 else 20 interval = float(sys.argv[2]) if len(sys.argv) > 2 else 0.35 grabber = Grabber() samples = [] # Locate both patterns once on a full-root grab, then read only the # bounding box that contains them (typically ~1/18 of the root's pixels). img = grabber.full() boxes = green_boxes(img) if len(boxes) < 2: print(f"found {len(boxes)} pattern(s): {boxes}") print("need two: the real pattern and the viewer's copy of it. Check " "g2g_pattern.py is running, the host is capturing that monitor, " "and the viewer window is not covering the pattern itself.") return src_box = min(boxes, key=lambda b: b[0] + b[1]) dst_box = max((b for b in boxes if b is not src_box), key=lambda b: b[2]) print(f"source {src_box} copy {dst_box} " f"scale {dst_box[2] / src_box[2]:.3f}") bx0 = min(src_box[0], dst_box[0]) by0 = min(src_box[1], dst_box[1]) bx1 = max(src_box[0] + src_box[2], dst_box[0] + dst_box[2]) by1 = max(src_box[1] + src_box[3], dst_box[1] + dst_box[3]) bw, bh = bx1 - bx0, by1 - by0 print(f"reading {bw}x{bh}+{bx0}+{by0} per sample " f"({100.0 * bw * bh / (img.width * img.height):.1f}% of root)") sb = (src_box[0] - bx0, src_box[1] - by0, src_box[2], src_box[3]) db = (dst_box[0] - bx0, dst_box[1] - by0, dst_box[2], dst_box[3]) for _ in range(n): crop = grabber.rect(bx0, by0, bw, bh) src = decode(crop, *sb) dst = decode(crop, *db) if src is None or dst is None: continue samples.append((src - dst) % (1 << BITS)) time.sleep(interval) if not samples: print("no samples") return plausible = sorted(s for s in samples if 0 < s < 1000) print(f"raw samples ({len(samples)}): {sorted(samples)}") if plausible: k = len(plausible) print(f"end-to-end latency: n={k} p50 {plausible[k // 2]} ms " f"min {plausible[0]} ms p90 {plausible[int(0.9 * (k - 1))]} ms " f"max {plausible[-1]} ms") else: print("no plausible samples -- is the viewer showing a live stream?") if __name__ == "__main__": main()