nixos/hosts/yarn/forza-trigger/default.nix

{
  config,
  lib,
  pkgs,
  username,
  ...
}:
# Forza Horizon 4 / 5 → DualSense adaptive trigger bridge (variant D, "Cosmii"
# port). The daemon listens for Forza's fixed-format UDP "Data Out" telemetry
# stream at 60 Hz, parses each packet via fdp (nettrom/forza_motorsport, MIT),
# and drives the PS5 DualSense's adaptive triggers and lightbar via
# dualsense-controller (PyPI, MIT) over hidraw.
#
# Reference design: cosmii02/ForzaDSXlegacy (variant D in our taxonomy):
#   - Continuous baseline resistance on both triggers (always feels)
#   - Vibration overlay on slip events (both triggers)
#   - RPM-reactive lightbar in-race; car-class color in menus
#   - EWMA smoothing per channel
#   - No body LRA (avoids the "shakes my whole hand" complaint)
# See forza_trigger.py module docstring for the full reference table and
# divergences from upstream.
#
# 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 → 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.
#
# Tuning env vars (read by the daemon at startup; values clamp to [0, 1]):
#   FORZA_L2_INTENSITY=<0..1>   global L2 (brake) feel scale.   default 1.0
#   FORZA_R2_INTENSITY=<0..1>   global R2 (throttle) feel scale. default 1.0
#   FORZA_LIGHTBAR=<0|1>        enable lightbar feedback.        default 1
# Set them in `services.forzaTrigger.environment` (an Environment= block on
# the systemd unit) to override.
let
  cfg = config.services.forzaTrigger;
  pythonPackages = import ./python-packages.nix { inherit lib pkgs; };
  inherit (pythonPackages) dualsense-controller fdp;

  forzaTriggerSrc = ./forza_trigger.py;

  forzaTriggerBin = pkgs.writers.writePython3Bin "forza-trigger" {
    libraries = [
      dualsense-controller
      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 forzaTriggerSrc);

  # Build-time unit tests for the haptic computation. Failure here breaks the
  # NixOS build, so deploys can't ship a daemon whose pure-function logic has
  # regressed. Tests use stdlib unittest + a FakeController; no hardware needed.
  forzaTriggerTests =
    pkgs.runCommand "forza-trigger-tests"
      {
        nativeBuildInputs = [
          (pkgs.python3.withPackages (_: [
            dualsense-controller
            fdp
          ]))
        ];
      }
      ''
        cp ${forzaTriggerSrc} forza_trigger.py
        cp ${./test_forza_trigger.py} test_forza_trigger.py
        python -m unittest discover -p 'test_*.py' -v
        touch $out
      '';

  # The binary the system actually depends on. overrideAttrs adds the test
  # derivation as a build dependency: if the tests fail, forzaTriggerTests
  # fails to build, forzaTriggerBin's buildInputs can't be satisfied, and
  # the system build fails. This is the standard nixpkgs idiom for build-time
  # gating without ceremony around system.checks / passthru.tests.
  forzaTrigger = forzaTriggerBin.overrideAttrs (old: {
    buildInputs = (old.buildInputs or [ ]) ++ [ forzaTriggerTests ];
  });

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 ];

    # Socket-activated by Forza's Data Out UDP stream on port 5300.
    # The service starts when Forza Horizon 4/5 sends its first telemetry
    # packet, runs for the entire session, and exits on 3 s of silence
    # (--exit-on-idle). systemd restarts automatically on the next launch.
    #
    # Runs as a system service (not user service) so there is exactly one
    # instance regardless of how many users have active sessions. The service
    # runs as `cfg.user` (primary) for hidraw uaccess compatibility with the
    # udev TAG+="uaccess" rules.
    #
    # Crash-recovery: Restart=on-failure restarts the daemon mid-session if
    # it dies abnormally (non-zero exit). The clean exit-on-idle path returns
    # 0, which systemd leaves alone — the socket unit stays listening.
    systemd.services.forza-trigger = {
      description = "Forza Horizon → DualSense adaptive trigger bridge";
      # No wantedBy — socket activation pulls it in when Forza sends UDP.
      serviceConfig = {
        ExecStart = "${forzaTrigger}/bin/forza-trigger --host 127.0.0.1 --port ${toString cfg.port} --exit-on-idle";
        Restart = "on-failure";
        RestartSec = 3;
        User = cfg.user;
        # KillMode=control-group is the default, but be explicit.
        # systemd sends SIGTERM by default; Python doesn't route that to
        # KeyboardInterrupt, so we tell systemd to send SIGINT instead.
        # The daemon registers handlers for both for robustness.
        KillMode = "control-group";
        KillSignal = "SIGINT";
        TimeoutStopSec = 5;
      };
    };

    systemd.sockets.forza-trigger = {
      wantedBy = [ "sockets.target" ];
      socketConfig = {
        ListenDatagram = "127.0.0.1:${toString cfg.port}";
        # Guard against rapid re-triggering. One activation per Forza session
        # (minutes+) is normal; > 3 within 10 s is pathological (bug).
        TriggerLimitIntervalSec = 10;
        TriggerLimitBurst = 3;
      };
    };
  };
}