far2l icon indicating copy to clipboard operation
far2l copied to clipboard

[WIP] [PoC] org.freedesktop.portal.GlobalShortcuts tryouts

Open unxed opened this issue 1 month ago • 7 comments

Touch #2040.

Need somebody with Ubuntu 25.10 to test it as my Cinnamon is not ready for it yet

unxed avatar Dec 13 '25 03:12 unxed

Во первых непонятно почему изменены файлы которые вроде не изменены? Во вторых - это работает через ssh? Если нет - то это совершенно не альтернатива Xi.

elfmz avatar Dec 13 '25 06:12 elfmz

@unxed во всех измененных тобой файлах концы строк вместо LF стали CRLF :cry: из-за этого в git сложно понять что именно изменилось по сути. Это какой редактор у тебя так гадит с концами строк?

akruphi avatar Dec 13 '25 06:12 akruphi

Можно запустить локальный фар с этой штукой, а там уже внутри по ssh ходить в удаленный фар через экстеншны. Я, собственно, и для xi такую модель использования считаю предпочтительной, потому как через экстеншны точно не будет проблем с синхронизацией иксового ввода с терминальным.

CRLF гляну, упс, чтото проворонил, не должно такого быть, конечно. UPD: поправил, спс!

unxed avatar Dec 13 '25 07:12 unxed

@unxed расскажи из-за чего у тебя LF самопроизвольно на CRLF сменились - чтоб знать, ибо не хотелось самому на такое в спешке напороться.

akruphi avatar Dec 13 '25 08:12 akruphi

This demo script correctly intercepts keys globally in GNOME 48 in Ubuntu 25.04. But I still can't get it working in tty backend...

import dbus
import dbus.mainloop.glib
from gi.repository import GLib
import sys

loop = None

def on_activated(*args, **kwargs):
    # Accept everything to avoid crashing on signature mismatch
    print("\n--- SIGNAL RECEIVED ---")
    try:
        # Signal signature: (o: session_handle, s: shortcut_id, t: timestamp, a{sv}: options)
        if len(args) >= 2:
            session_handle = args[0]
            shortcut_id = args[1]
            print(f"Session: {session_handle}")
            print(f"ID:      {shortcut_id}")
            
            if shortcut_id == "Test1":
                print(">>> ACTION: Ctrl+Enter detected!")
            elif shortcut_id == "Test2":
                print(">>> ACTION: Ctrl+Space detected!")
        else:
            print(f"Args: {args}")
    except Exception as e:
        print(f"Error parsing signal: {e}")

def on_bind_response(results):
    print("\n--- BindShortcuts Response ---")
    # Documentation states BindShortcuts returns (o), but the actual binding 
    # results arrive here in the dictionary (a{sv})
    print(f"Raw results: {results}")
    
    if 'shortcuts' in results:
        print("Shortcuts info returned by portal:")
        for s in results['shortcuts']:
            print(f"  {s}")
            
    print("\nREADY! Try pressing Ctrl+Enter or Ctrl+Space now.")
    print("If nothing happens, check if the shortcuts were actually enabled in the dialog.")

def on_bind_error(e):
    print(f"Bind Error: {e}")
    sys.exit(1)

def on_session_response(response, results):
    if response != 0:
        print(f"Error creating session: {response}")
        sys.exit(1)
        
    session_handle = results['session_handle']
    print(f"Session Created. Handle: {session_handle}")

    bus = dbus.SessionBus()
    portal = bus.get_object('org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop')
    iface = dbus.Interface(portal, 'org.freedesktop.portal.GlobalShortcuts')

    shortcuts = [
        ('Test1', {
            'description': 'Test One (Ctrl+Enter)', 
            'preferred_trigger': 'Control+Return'
        }),
        ('Test2', {
            'description': 'Test Two (Ctrl+Space)', 
            'preferred_trigger': 'Control+space'
        })
    ]
    
    print("Requesting BindShortcuts...")
    
    # Important: parent_window. An empty string "" is valid for console utilities.
    iface.BindShortcuts(
        session_handle,
        shortcuts,
        "", 
        {'handle_token': 'my_bind_token_v2'},
        reply_handler=on_bind_response,
        error_handler=on_bind_error
    )

    # We omit `path=session_handle` to catch the signal globally, 
    # just in case the emission path differs from the handle.
    bus.add_signal_receiver(
        on_activated,
        dbus_interface='org.freedesktop.portal.GlobalShortcuts',
        signal_name='Activated'
    )

def main():
    global loop
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SessionBus()

    try:
        portal = bus.get_object('org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop')
        iface = dbus.Interface(portal, 'org.freedesktop.portal.GlobalShortcuts')
        
        print("Creating session...")
        request_path = iface.CreateSession(
            {'session_handle_token': 'py_test_sess_v2'}
        )
        print(f"Request object path: {request_path}")

        bus.add_signal_receiver(
            on_session_response,
            dbus_interface='org.freedesktop.portal.Request',
            signal_name='Response',
            path=request_path
        )
        
        loop = GLib.MainLoop()
        loop.run()
        
    except dbus.exceptions.DBusException as e:
        print(f"DBus Exception: {e}")

if __name__ == '__main__':
    main()

unxed avatar Dec 14 '25 01:12 unxed

Хотя никто не мешает сделать обёртку над этим Питон-скриптом 🤷🙂

Тут ещё вот какая штука: там каждое сочетание клавиш нужно вручную подтверждать юзеру. Пока не придумал, как это можно изящно реализовать в UI, у нас же этих сочетаний мильон.

Можно для начала сделать хотя бы несколько самых часто употребляемых, без которых в простых терминалах прям жизни нет. У меня это Ctrl+Enter, например. А у кого ещё какие?

unxed avatar Dec 14 '25 01:12 unxed

В теории вот это

sudo apt install dbus-x11
gsettings set org.gnome.mutter.wayland xwayland-allow-grabs "true"
gsettings set org.gnome.mutter.wayland xwayland-grab-access-rules "['far2l','far2l_ttyx.broker','far2l_gui.so']"

должно было включать xi и под XWayland. Если у нас закомметриовать отключение его под вэйлендом. Но работать оно в GNOME 48 тоже не хочет.

А для схемы с GlobalShortcuts у нас хотя бы есть пример рабочего скрипта, делающего ровно то что нам надо.

unxed avatar Dec 14 '25 04:12 unxed

This demo script correctly intercepts keys globally in GNOME 48

This is an example how it can be used from C code, it works fine. Unfortunately, still no success with far2l tty backend itself.

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <string.h>
#include <string>
#include <thread>
#include <chrono>
#include <sys/wait.h>
#include <atomic>

// Minimal mocking
#define LEFT_CTRL_PRESSED 0x0008

// Python proxy script content (Same as in far2l)
static const char* PYTHON_PROXY_SCRIPT = R"EOF(
import dbus
import dbus.mainloop.glib
from gi.repository import GLib
import sys
import os
import signal
import time

# Flush stdout/stderr immediately for IPC
sys.stdout.reconfigure(line_buffering=True)
sys.stderr.reconfigure(line_buffering=True)

loop = None

def on_activated(session_handle, shortcut_id, timestamp, options):
    print(f"EVENT:{shortcut_id}", flush=True)

def on_bind_response(response, results):
    if response == 0:
        print("LOG:BindShortcuts request acknowledged by portal. Waiting for user dialog...", flush=True)
    else:
        print(f"LOG:BindShortcuts request failed with code {response}", file=sys.stderr, flush=True)

def on_bind_error(e):
    print(f"LOG:Bind Error: {e}", file=sys.stderr, flush=True)
    sys.exit(1)

def on_session_response(response, results):
    if response != 0:
        print(f"LOG:Error creating session response: {response}", file=sys.stderr, flush=True)
        sys.exit(1)
        
    session_handle = results.get('session_handle', 'NO_HANDLE')
    print(f"LOG:Session handle received: {session_handle}", flush=True)
    
    bus = dbus.SessionBus()
    try:
        portal = bus.get_object('org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop')
        iface = dbus.Interface(portal, 'org.freedesktop.portal.GlobalShortcuts')
    except Exception as e:
        print(f"LOG:Failed to get portal interface: {e}", file=sys.stderr, flush=True)
        sys.exit(1)

    shortcuts = [
        ('CtrlEnter', {'description': 'Far2l Ctrl+Enter', 'preferred_trigger': 'Control+Return'}),
        ('CtrlTab', {'description': 'Far2l Ctrl+Tab', 'preferred_trigger': 'Control+Tab'})
    ]
    
    print("LOG:Binding shortcuts...", flush=True)
    
    bind_token = f"test_bind_{os.getpid()}"
    bind_options = {'handle_token': bind_token}
    
    # Try empty parent window first, as per current code
    parent_window = ""
    
    bind_request_path = iface.BindShortcuts(
        session_handle,
        shortcuts,
        parent_window, 
        bind_options
    )
    
    print(f"LOG:BindShortcuts request path: {bind_request_path}", flush=True)
    bus.add_signal_receiver(
        on_bind_response,
        dbus_interface='org.freedesktop.portal.Request',
        signal_name='Response',
        path=bind_request_path
    )

    bus.add_signal_receiver(
        on_activated,
        dbus_interface='org.freedesktop.portal.GlobalShortcuts',
        signal_name='Activated',
        path=session_handle
    )

def main():
    global loop
    print("LOG:Python script started. PID:", os.getpid(), flush=True)
    
    try:
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
        bus = dbus.SessionBus()
        
        portal = bus.get_object('org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop')
        iface = dbus.Interface(portal, 'org.freedesktop.portal.GlobalShortcuts')
        
        print("LOG:Creating session...", flush=True)
        
        session_token = f"test_session_{os.getpid()}"
        request_token = f"test_request_{os.getpid()}"
        create_options = {
            'handle_token': request_token,
            'session_handle_token': session_token
        }
        
        request_path = iface.CreateSession(create_options)
        print(f"LOG:CreateSession request path: {request_path}", flush=True)

        bus.add_signal_receiver(
            on_session_response,
            dbus_interface='org.freedesktop.portal.Request',
            signal_name='Response',
            path=request_path
        )
        
        loop = GLib.MainLoop()
        loop.run()
        
    except Exception as e:
        print(f"LOG:Python Exception: {e}", file=sys.stderr, flush=True)
        sys.exit(1)

if __name__ == '__main__':
    main()
)EOF";

class TestRunner {
    std::atomic<bool> _running{false};
    std::thread _thread;
    int _pipe_fd = -1;
    int _err_pipe_fd = -1;
    pid_t _child_pid = -1;
    std::string _script_path;

public:
    void Start() {
        if (_running) return;
        fprintf(stderr, "[TestRunner] Start() called\n");

        char tmp_path[] = "/tmp/test_wl_proxy_XXXXXX.py";
        int fd = mkstemps(tmp_path, 3);
        if (fd == -1) { perror("mkstemps"); return; }
        
        write(fd, PYTHON_PROXY_SCRIPT, strlen(PYTHON_PROXY_SCRIPT));
        close(fd);
        _script_path = tmp_path;
        fprintf(stderr, "[TestRunner] Script: %s\n", _script_path.c_str());

        int out_pipe[2], err_pipe[2];
        pipe(out_pipe);
        pipe(err_pipe);

        pid_t pid = fork();
        if (pid == 0) {
            // Child
            close(out_pipe[0]); close(err_pipe[0]);
            dup2(out_pipe[1], STDOUT_FILENO);
            dup2(err_pipe[1], STDERR_FILENO);
            close(out_pipe[1]); close(err_pipe[1]);

            // Direct execution, no 'script' wrapper for now
            execlp("python3", "python3", _script_path.c_str(), NULL);
            perror("execlp");
            _exit(1);
        }

        // Parent
        close(out_pipe[1]); close(err_pipe[1]);
        _pipe_fd = out_pipe[0];
        _err_pipe_fd = err_pipe[0];
        _child_pid = pid;
        _running = true;
        
        _thread = std::thread(&TestRunner::WorkerThread, this);
    }

    void Stop() {
        _running = false;
        if (_child_pid > 0) kill(_child_pid, SIGTERM);
        if (_thread.joinable()) _thread.join();
        unlink(_script_path.c_str());
    }

    void WorkerThread() {
        fprintf(stderr, "[TestRunner] Worker monitoring...\n");
        char buf[1024];
        while (_running) {
            fd_set fds;
            FD_ZERO(&fds);
            int max_fd = 0;
            if (_pipe_fd != -1) { FD_SET(_pipe_fd, &fds); max_fd = std::max(max_fd, _pipe_fd); }
            if (_err_pipe_fd != -1) { FD_SET(_err_pipe_fd, &fds); max_fd = std::max(max_fd, _err_pipe_fd); }

            if (max_fd == 0) break;

            struct timeval tv = {1, 0};
            int ret = select(max_fd + 1, &fds, NULL, NULL, &tv);
            if (ret > 0) {
                if (_pipe_fd != -1 && FD_ISSET(_pipe_fd, &fds)) {
                    ssize_t n = read(_pipe_fd, buf, sizeof(buf)-1);
                    if (n > 0) {
                        buf[n] = 0;
                        fprintf(stderr, "[STDOUT] %s", buf);
                    } else {
                        close(_pipe_fd); _pipe_fd = -1;
                    }
                }
                if (_err_pipe_fd != -1 && FD_ISSET(_err_pipe_fd, &fds)) {
                     ssize_t n = read(_err_pipe_fd, buf, sizeof(buf)-1);
                    if (n > 0) {
                        buf[n] = 0;
                        fprintf(stderr, "[STDERR] %s", buf);
                    } else {
                        close(_err_pipe_fd); _err_pipe_fd = -1;
                    }
                }
            }
        }
    }
};

int main() {
    TestRunner runner;
    runner.Start();
    printf("Press Enter to stop...\n");
    getchar();
    runner.Stop();
    return 0;
}

unxed avatar Dec 14 '25 11:12 unxed