#!/usr/bin/python3 import os import asyncio from pywayland.client import Display from pywayland.protocol.wayland import ( WlSeat, WlShell, WlShm, ) from protocols.text_input_unstable_v3 import ZwpTextInputManagerV3 import dbussy as dbus from dbussy import \ DBUS, \ Introspection import ravel socket_path = os.environ["XDG_RUNTIME_DIR"] + "/maliit-server" if not os.path.exists(socket_path): os.makedirs(socket_path) my_socket_name = "unix:path=" + socket_path + "/dbus-socket" my_interface_name = "com.meego.inputmethod.uiserver1" client_interface_name = "com.meego.inputmethod.inputcontext1" idle_timeout = 15 # seconds, short value for testing ClientInterface = ravel.def_proxy_interface \ ( ravel.INTERFACE.CLIENT, name = "ClientInterface", introspected = Introspection.Interface ( name = client_interface_name, methods = [ Introspection.Interface.Method ( name = "setLanguage", args = [ Introspection.Interface.Method.Arg ( name = "lang", type = dbus.BasicType(dbus.TYPE.STRING), direction = Introspection.DIRECTION.IN ), ], ), Introspection.Interface.Method ( name = "setRedirectKeys", args = [ Introspection.Interface.Method.Arg ( name = "redi", type = dbus.BasicType(dbus.TYPE.BOOLEAN), direction = Introspection.DIRECTION.IN ), ], ), Introspection.Interface.Method ( name = "updateInputMethodArea", args = [ Introspection.Interface.Method.Arg ( name = "x", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), Introspection.Interface.Method.Arg ( name = "y", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), Introspection.Interface.Method.Arg ( name = "width", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), Introspection.Interface.Method.Arg ( name = "height", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), ], ), Introspection.Interface.Method ( name = "commitString", args = [ Introspection.Interface.Method.Arg ( name = "string", type = dbus.BasicType(dbus.TYPE.STRING), direction = Introspection.DIRECTION.IN ), Introspection.Interface.Method.Arg ( name = "replacementStart", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), Introspection.Interface.Method.Arg ( name = "replacementLength", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), Introspection.Interface.Method.Arg ( name = "cursorPos", type = dbus.BasicType(dbus.TYPE.INT32), direction = Introspection.DIRECTION.IN ), ], ), ], ), is_async = True ) useless_objects_bus_name = "org.maliit.server" useless_objects_iface_name = "org.maliit.Server.Address" @ravel.interface(ravel.INTERFACE.SERVER, name = useless_objects_iface_name) class UselessObjectServer : __slots__ = ("bus",) def __init__(self, bus) : self.bus = bus #end __init__ @ravel.propgetter \ ( name = "address", type = dbus.BasicType(dbus.TYPE.STRING), change_notification = dbus.Introspection.PROP_CHANGE_NOTIFICATION.NEW_VALUE, ) def get_address(self) : return "unix:path=" + socket_path + "/dbus-socket" #end get_address object_created = ravel.def_signal_stub \ ( name = "object_created", in_signature = [], ) object_deleted = ravel.def_signal_stub \ ( name = "object_deleted", in_signature = [], ) #end UselessObjectServer background_tasks = set() def run_in_background(invocation): task = asyncio.create_task(invocation) background_tasks.add(task) task.add_done_callback(background_tasks.discard) @ravel.interface(ravel.INTERFACE.SERVER, name = my_interface_name) class DirectConnectServer : __slots__ = ("bus", "_last_request", "intf", "wayland") def __init__(self, bus, wayland) : self.bus = bus self.wayland = wayland self.intf = ClientInterface \ ( connection = self.bus.connection, dest = "com.meego.inputmethod.inputcontext1", # no D-Bus daemon to care )["/com/meego/inputmethod/inputcontext"] #end __init__ # @ravel.method \ # ( # name = "reset", # in_signature = "", # out_signature = "", # arg_keys = [], # ) # def reset(self) : # print("got reset() message") # #end reset @ravel.method \ ( name = "updateWidgetInformation", in_signature = "a{sv}b", out_signature = "", arg_keys = ["vardict", "boolean"], ) def updateWidgetInformation(self, vardict, boolean) : # print("got updateWidgetInformation() message") # print(vardict) surrounding_text = vardict["surroundingText"][1] cursor_position = vardict["cursorPosition"][1] self.wayland["text_input"].set_surrounding_text(surrounding_text, cursor_position, cursor_position) self.wayland["text_input"].commit() self.wayland["display"].flush() #end updateWidgetInformation # @ravel.method \ # ( # name = "activateContext", # in_signature = "", # out_signature = "", # arg_keys = [], # ) # def activateContext(self) : # print("got activateContext() message") # #end activateContext @ravel.method \ ( name = "appOrientationChanged", in_signature = "i", out_signature = "", arg_keys = ["orientation"], ) def appOrientationChanged(self, orientation) : print("got appOrientationChanged() message") #end appOrientationChanged @ravel.method \ ( name = "showInputMethod", in_signature = "", out_signature = "", arg_keys = [], ) def showInputMethod(self) : # print("got showInputMethod() message, setting input size") self.wayland["text_input"].set_surrounding_text("magictext", 0, 0) self.wayland["display"].flush() self.wayland["text_input"].set_content_type(0, 0) self.wayland["text_input"].enable() self.wayland["text_input"].commit() self.wayland["display"].flush() run_in_background(self.intf.updateInputMethodArea(0, 1360, 1080, 670)) #end showInputMethod @ravel.method \ ( name = "hideInputMethod", in_signature = "", out_signature = "", arg_keys = [], ) def hideInputMethod(self) : # print("got hideInputMethod() message, setting input size") self.wayland["text_input"].disable() self.wayland["text_input"].commit() self.wayland["display"].flush() run_in_background(self.intf.updateInputMethodArea(0, 0, 0, 0)) #end hideInputMethod # @ravel.method \ # ( # name = "setGlobalCorrectionEnabled", # in_signature = "b", # out_signature = "", # arg_keys = ["enabled"], # ) # def setGlobalCorrectionEnabled(self, enabled) : # print("got setGlobalCorrectionEnabled() message") # #end setGlobalCorrectionEnabled # @ravel.method \ # ( # name = "setDetectableAutoRepeat", # in_signature = "b", # out_signature = "", # arg_keys = ["detectable"], # ) # def setDetectableAutoRepeat(self, detectable) : # print("got setDetectableAutoRepeat() message") # #end setDetectableAutoRepeat #end DirectConnectServer def handle_enter(text_input, surface): # print("got enter event for surface ") return 0 def handle_leave(text_input, surface): # print("got leave event for surface ") return 0 def handle_preedit_string(text_input, string, cursor_begin, cursor_end): # print("got preedit str ") return 0 def handle_commit_string(text_input, string): # print("commit str: " + string + " sending to dbus") wayland = text_input.user_data run_in_background(wayland["dbus_server"].intf.commitString(string, 0, 0, 0)) return 0 def handle_delete_surrounding_text(text_input, before_len, after_len): # print("got delete surrounding text") return 0 def handle_done(text_input, serial): # print("got a done event serial " + str(serial)) return 0 def handle_seat_capabilities(wl_seat, capabilities): wayland = wl_seat.user_data if "text_input_manager" not in wayland: raise Exception("text_input_manager protocol is not supported by compositor") wayland["text_input"] = wayland["text_input_manager"].get_text_input(wl_seat) wayland["text_input"].user_data = wayland wayland["text_input"].dispatcher["enter"] = handle_enter wayland["text_input"].dispatcher["leave"] = handle_leave # wayland["text_input"].dispatcher["preedit_string"] = handle_preedit_string wayland["text_input"].dispatcher["commit_string"] = handle_commit_string wayland["text_input"].dispatcher["delete_surrounding_text"] = handle_delete_surrounding_text wayland["text_input"].dispatcher["done"] = handle_done wayland["display"].flush() return 1 def handle_registry_global(wl_registry, id_num, iface_name, version): wayland = wl_registry.user_data if iface_name == "wl_seat": wayland["seat"] = wl_registry.bind(id_num, WlSeat, version) wayland["seat"].user_data = wayland wayland["seat"].dispatcher["capabilities"] = handle_seat_capabilities elif iface_name == "zwp_text_input_manager_v3": wayland["text_input_manager"] = wl_registry.bind(id_num, ZwpTextInputManagerV3, version) wayland["display"].flush() return 1 async def setup_dbus(wayland): listen = ravel.Server(address = my_socket_name) clients = {} @ravel.signal \ ( in_signature = "", bus_keyword = "conn" ) def connection_terminated(conn) : # handler for signal sent by libdbus when client disconnects. print("connection from PID %s terminated\n" % conn.connection.unix_process_id) del clients[id(conn)] #end connection_terminated while True : conn = await listen.await_new_connection(timeout = idle_timeout / 3) if conn != None : print("new connection on DBus") server = DirectConnectServer(conn, wayland) wayland["dbus_server"] = server conn.register \ ( path = "/com/meego/inputmethod/uiserver1", fallback = True, interface = server ) conn.listen_signal \ ( path = "/", fallback = True, interface = DBUS.INTERFACE_LOCAL, name = "Disconnected", func = connection_terminated ) clients[id(conn)] = conn await server.intf.setLanguage('') #end if #end while async def run_dbus_loop(wayland): server = await setup_dbus(wayland) display = Display() display.connect() loop = asyncio.new_event_loop() loop.add_reader(display.get_fd(), lambda: display.dispatch(block=True)) wayland = {} wayland["display"] = display wayland["registry"] = wayland["display"].get_registry() wayland["registry"].user_data = wayland wayland["registry"].dispatcher["global"] = handle_registry_global wayland["display"].flush() bus = ravel.session_bus() bus.attach_asyncio(loop) bus.request_name \ ( bus_name = useless_objects_bus_name, flags = DBUS.NAME_FLAG_DO_NOT_QUEUE ) bus.register \ ( path = "/org/maliit/server", fallback = True, interface = UselessObjectServer(bus) ) loop.create_task(run_dbus_loop(wayland)) loop.run_forever()