yarn: forza dualsense adaptive trigger bridge

hosts/yarn/forza-trigger/default.nix

@@ -0,0 +1,196 @@
+{
+  config,
+  lib,
+  pkgs,
+  username,
+  ...
+}:
+# Forza Horizon 4 / 5 → DualSense adaptive trigger bridge.
+#
+# Forza emits a fixed-format UDP telemetry stream ("Data Out") at 60 Hz on a
+# user-configured port. We listen on that port, parse each packet via fdp
+# (nettrom/forza_motorsport, MIT), and drive the PS5 DualSense's adaptive
+# triggers via pydualsense (PyPI, MIT) which talks HID over hidraw.
+#
+# Setup on the user side, once enabled here:
+#   - plug the DualSense in over USB and disable Steam Input for the
+#     controller (Settings → Controller → "PlayStation Configuration Support":
+#     OFF). Bluetooth works too but the udev/hidraw path is more reliable
+#     over USB.
+#   - in Forza, HUD options \u2192 set Data Out: ON, Data Out IP: 127.0.0.1,
+#     Data Out IP Port: 5300, and (FM7 only) Data Out Packet Format: CAR DASH.
+#
+# System-interaction notes:
+#   - With multiple DualSense controllers connected, pydualsense picks one
+#     non-deterministically (`# TODO: implement multiple controllers working`
+#     in pydualsense's source). Forza Horizon is single-player so this is
+#     usually fine. If you need to pin a specific controller, the cleanest
+#     route is monkey-patching `pydualsense.__find_device`.
+#   - The included `dualsensectl` will be overwritten by our BG thread within
+#     ~4 ms; use `systemctl --user stop forza-trigger` first when debugging.
+#   - Hot-plug recovery happens in-process: the daemon polls pydualsense's BG
+#     thread liveness and re-runs `pydualsense.init()` on disconnect. systemd's
+#     `Restart=on-failure` exists only as a crash-recovery safety net.
+let
+  cfg = config.services.forzaTrigger;
+  python = pkgs.python3;
+
+  # CFFI bindings to libhidapi. Upstream is flok/hidapi-cffi published on
+  # PyPI under the name `hidapi-usb`. The shipped hidapi.py picks a libhidapi
+  # soname via ffi.dlopen() — we substitute absolute store paths so the
+  # interpreter inside our wrapped python env can find it without
+  # LD_LIBRARY_PATH gymnastics.
+  hidapi-usb = python.pkgs.buildPythonPackage rec {
+    pname = "hidapi-usb";
+    version = "0.3.2";
+    format = "setuptools";
+
+    # PyPI's project URL slug uses a hyphen (`hidapi-usb`) but the sdist file
+    # itself is PEP-625-normalized to an underscore (`hidapi_usb-…`). Stock
+    # fetchPypi assumes they match — they don't here, so fetch by direct URL.
+    src = pkgs.fetchurl {
+      url = "https://files.pythonhosted.org/packages/55/80/960ae94b615e26a7d1aeebe8e9fefda2f25608bf1016f9aec268b328c35e/hidapi_usb-${version}.tar.gz";
+      hash = "sha256-oxp+2i+qqYd1uwiS2Dh8/PzO62iYQQXpR936MnDIFk0=";
+    };
+
+    propagatedBuildInputs = [ python.pkgs.cffi ];
+
+    postPatch = ''
+      substituteInPlace hidapi.py \
+        --replace-fail "'libhidapi-hidraw.so'," "'${pkgs.hidapi}/lib/libhidapi-hidraw.so'," \
+        --replace-fail "'libhidapi-hidraw.so.0'," "'${pkgs.hidapi}/lib/libhidapi-hidraw.so.0',"
+    '';
+
+    pythonImportsCheck = [ "hidapi" ];
+
+    meta = {
+      description = "CFFI wrapper for hidapi (used by pydualsense)";
+      homepage = "https://github.com/flok/hidapi-cffi";
+      license = lib.licenses.bsd3;
+    };
+  };
+
+  pydualsense = python.pkgs.buildPythonPackage rec {
+    pname = "pydualsense";
+    version = "0.7.5";
+    format = "pyproject";
+
+    src = python.pkgs.fetchPypi {
+      pname = "pydualsense";
+      inherit version;
+      hash = "sha256-YgX8AJE4f8p7geKT3xlCD0Mlh1GcyHpBz4rEIqdwKgs=";
+    };
+
+    nativeBuildInputs = [ python.pkgs.poetry-core ];
+    propagatedBuildInputs = [ hidapi-usb ];
+
+    pythonImportsCheck = [ "pydualsense" ];
+
+    meta = {
+      description = "Control your PS5 DualSense controller from Python";
+      homepage = "https://github.com/flok/pydualsense";
+      license = lib.licenses.mit;
+    };
+  };
+
+  # Single-file Forza UDP packet parser. Pinned to a known-good commit; the
+  # repo is dormant (last commit 2021) but the FH4 packet layout is frozen
+  # and FH5 reuses it byte-for-byte.
+  fdp = python.pkgs.buildPythonPackage {
+    pname = "fdp";
+    version = "0-unstable-2021-05-28";
+    format = "other";
+
+    src = pkgs.fetchurl {
+      url = "https://raw.githubusercontent.com/nettrom/forza_motorsport/61845cb7ff4082211292a51ce3c49edbfd2d6503/fdp.py";
+      hash = "sha256-osFaVF9VaEzU4dp3x6KN6OF7SXsd9ZBwvilU+xTT7mM=";
+    };
+
+    dontUnpack = true;
+
+    installPhase = ''
+      runHook preInstall
+      install -Dm644 $src $out/${python.sitePackages}/fdp.py
+      runHook postInstall
+    '';
+
+    pythonImportsCheck = [ "fdp" ];
+
+    meta = {
+      description = "ForzaDataPacket — Forza Motorsport / Horizon UDP packet parser";
+      homepage = "https://github.com/nettrom/forza_motorsport";
+      license = lib.licenses.mit;
+    };
+  };
+
+  forzaTrigger = pkgs.writers.writePython3Bin "forza-trigger" {
+    libraries = [
+      pydualsense
+      fdp
+    ];
+    # The wrapped binary doesn't need style enforcement — readability of
+    # the source file is what matters, and that lives in forza_trigger.py.
+    doCheck = false;
+  } (builtins.readFile ./forza_trigger.py);
+
+in
+{
+  options.services.forzaTrigger = {
+    enable = lib.mkEnableOption "Forza Horizon → DualSense adaptive trigger bridge";
+
+    user = lib.mkOption {
+      type = lib.types.str;
+      default = username;
+      description = ''
+        User the trigger daemon runs as. Must be the user playing Forza so
+        the DualSense's hidraw uaccess ACL applies.
+      '';
+    };
+
+    port = lib.mkOption {
+      type = lib.types.port;
+      default = 5300;
+      description = ''
+        UDP port the daemon listens on for Forza Data Out packets. Must
+        match the value configured in Forza's HUD options.
+      '';
+    };
+  };
+
+  config = lib.mkIf cfg.enable {
+    # uaccess hands /dev/hidraw* of the connected PS5 DualSense to the
+    # active-seat user via ACL. Steam ships near-identical rules; declaring
+    # them here keeps the module self-contained (and works even if Steam
+    # isn't running).
+    services.udev.extraRules = ''
+      # PS5 DualSense (USB)
+      KERNEL=="hidraw*", ATTRS{idVendor}=="054c", ATTRS{idProduct}=="0ce6", MODE="0660", TAG+="uaccess"
+      # PS5 DualSense Edge (USB)
+      KERNEL=="hidraw*", ATTRS{idVendor}=="054c", ATTRS{idProduct}=="0df2", MODE="0660", TAG+="uaccess"
+      # PS5 DualSense (Bluetooth)
+      KERNEL=="hidraw*", KERNELS=="*054C:0CE6*", MODE="0660", TAG+="uaccess"
+      # PS5 DualSense Edge (Bluetooth)
+      KERNEL=="hidraw*", KERNELS=="*054C:0DF2*", MODE="0660", TAG+="uaccess"
+    '';
+
+    environment.systemPackages = [
+      forzaTrigger
+      # CLI companion for sanity-checking the controller (battery, lightbar,
+      # raw trigger modes, monitor add/remove events).
+      pkgs.dualsensectl
+    ];
+
+    # User-level service so it inherits the seat-bound uaccess ACL on
+    # /dev/hidraw* and dies cleanly when the user logs out.
+    systemd.user.services.forza-trigger = {
+      description = "Forza Horizon → DualSense adaptive trigger bridge";
+      wantedBy = [ "default.target" ];
+      after = [ "graphical-session.target" ];
+      serviceConfig = {
+        ExecStart = "${forzaTrigger}/bin/forza-trigger --host 127.0.0.1 --port ${toString cfg.port}";
+        Restart = "on-failure";
+        RestartSec = 3;
+      };
+    };
+  };
+}