136 lines
5.3 KiB
Python
136 lines
5.3 KiB
Python
"""Generic painted X11/clipboard sources for an isolated Xvfb session."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import gi
|
|
gi.require_version("Gtk", "3.0")
|
|
gi.require_version("Gdk", "3.0")
|
|
gi.require_version("GdkPixbuf", "2.0")
|
|
from gi.repository import Gdk, GdkPixbuf, GLib, Gtk
|
|
from Xlib import X, Xatom, display
|
|
|
|
|
|
def main():
|
|
Gtk.init([])
|
|
connection = display.Display()
|
|
root = connection.screen().root
|
|
activeAtom = connection.intern_atom("_NET_ACTIVE_WINDOW")
|
|
clipboardAtom = connection.intern_atom("CLIPBOARD")
|
|
pngAtom = connection.intern_atom("image/png")
|
|
pngPath = Path(__file__).resolve().parents[1] / "internal/ocr/testdata/painted-text.png"
|
|
pendingCapture = None
|
|
|
|
def make_window(parent, x, y, width, height, color):
|
|
window = parent.create_window(x, y, width, height, 0,
|
|
connection.screen().root_depth,
|
|
X.InputOutput, X.CopyFromParent,
|
|
background_pixel=color)
|
|
window.set_wm_name("OCR fixture")
|
|
window.set_wm_class("example", "example")
|
|
window.map()
|
|
return window
|
|
|
|
window = make_window(root, 40, 30, 300, 200, 0x336699)
|
|
child = make_window(window, 10, 15, 80, 50, 0xABCDEF)
|
|
other = make_window(root, 350, 30, 100, 100, 0x992244)
|
|
windows = dict(window=window, child=child, other=other)
|
|
clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
|
|
painted = GdkPixbuf.Pixbuf.new_from_file(str(pngPath))
|
|
clipboard.set_image(painted)
|
|
paintedWindow = make_window(root, 40, 270, 640, 220, 0xFFFFFF)
|
|
windows["painted"] = paintedWindow
|
|
# The private fixture has an Xvfb 24-bit TrueColor root with 32-bit little
|
|
# endian pixels. Paint the checked-in image without adding another library.
|
|
data = painted.get_pixels()
|
|
stride = painted.get_rowstride()
|
|
nativeData = bytearray(640 * 220 * 4)
|
|
for y in range(220):
|
|
for x in range(640):
|
|
source = y * stride + x * 3
|
|
destination = (y * 640 + x) * 4
|
|
nativeData[destination:destination + 3] = data[source:source + 3][::-1]
|
|
gc = paintedWindow.create_gc()
|
|
for y in range(0, 220, 32):
|
|
height = min(32, 220 - y)
|
|
paintedWindow.put_image(gc, 0, y, 640, height, X.ZPixmap, 24, 0,
|
|
bytes(nativeData[y * 640 * 4:(y + height) * 640 * 4]))
|
|
gc.free()
|
|
|
|
def activate(target):
|
|
root.change_property(activeAtom, Xatom.WINDOW, 32, [target.id])
|
|
target.set_input_focus(X.RevertToParent, X.CurrentTime)
|
|
|
|
activate(window)
|
|
connection.sync()
|
|
print(json.dumps({key: value.id for key, value in windows.items()}), flush=True)
|
|
loop = GLib.MainLoop()
|
|
|
|
def command_ready(_source, condition):
|
|
nonlocal pendingCapture
|
|
if condition & GLib.IO_HUP:
|
|
loop.quit()
|
|
return False
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
loop.quit()
|
|
return False
|
|
command = json.loads(line)
|
|
target = windows.get(command.get("window", "window"), window)
|
|
action = command["action"]
|
|
if action == "activate":
|
|
activate(target)
|
|
elif action == "move":
|
|
target.configure(x=command["x"], y=command["y"])
|
|
elif action == "resize":
|
|
target.configure(width=command["width"], height=command["height"])
|
|
elif action == "hide":
|
|
target.unmap()
|
|
elif action == "destroy":
|
|
target.destroy()
|
|
elif action == "overlay":
|
|
windows["overlay"] = make_window(root, 40, 30, 30, 30, 0xAA1100)
|
|
elif action == "clipboard-text":
|
|
clipboard.set_text("Generic clipboard text", -1)
|
|
elif action == "clipboard-image":
|
|
clipboard.set_image(painted)
|
|
elif action == "clipboard-block":
|
|
other.set_selection_owner(clipboardAtom, X.CurrentTime)
|
|
elif action == "clipboard-release":
|
|
if pendingCapture is None:
|
|
raise RuntimeError("no pending image transfer")
|
|
from Xlib.protocol import event
|
|
request = pendingCapture
|
|
request.requestor.change_property(request.property, pngAtom, 8, pngPath.read_bytes())
|
|
request.requestor.send_event(event.SelectionNotify(
|
|
time=request.time, requestor=request.requestor.id,
|
|
selection=request.selection, target=request.target,
|
|
property=request.property))
|
|
pendingCapture = None
|
|
elif action != "state":
|
|
raise ValueError("unknown fixture action")
|
|
connection.sync()
|
|
focus = connection.get_input_focus().focus
|
|
children = root.query_tree().children
|
|
print(json.dumps({"focus": focus.id, "windows": len(children)}), flush=True)
|
|
return True
|
|
|
|
GLib.io_add_watch(sys.stdin, GLib.IO_IN | GLib.IO_HUP, command_ready)
|
|
|
|
def x_ready(_source, _condition):
|
|
nonlocal pendingCapture
|
|
while connection.pending_events():
|
|
request = connection.next_event()
|
|
if request.type == X.SelectionRequest and request.target == pngAtom:
|
|
pendingCapture = request
|
|
print(json.dumps({"capture-request": 1}), flush=True)
|
|
return True
|
|
|
|
GLib.io_add_watch(connection.fileno(), GLib.IO_IN, x_ready)
|
|
loop.run()
|
|
connection.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|