snakemake_bridge 0.1.1

Embed a real-time Snakemake workflow monitor in a Dart desktop app: WebSocket server, typed event models, state reducer and process launcher.

snakemake_bridge for Dart

Monitor Snakemake 9+ workflows in real time inside a Dart Desktop application.

The snakemake_bridge package opens a local WebSocket server, receives the events emitted by Snakemake and turns them into state that is easy to render: jobs, progress, DAG, logs and errors.

This package is the Dart half of the integration. The other half is the Python plugin snakemake-logger-plugin-dart, which must be installed in the same Python environment as Snakemake.

How it works

Dart application
    │
    ├── starts WorkflowServer on 127.0.0.1
    │
    ├── launches the Snakemake process
    │       │
    │       └── Python plugin connects to the server over WebSocket
    │
    └── receives events → WorkflowRunState → updates the UI

The Dart application is responsible for starting the server before the workflow. When launching Snakemake, it tells the plugin the port and an authentication token. The plugin streams events without blocking workflow execution.

If the connection drops, the plugin reconnects and resends the events that were missed. The server discards duplicates automatically.

Requirements

• Dart SDK ^3.5.0 (that is, >= 3.5.0 and < 4.0.0);
• Python >= 3.11;
• Snakemake >= 9.0;
• snakemake-logger-plugin-dart installed in the same Python environment as Snakemake.

The package uses dart:io and targets desktop or command-line applications. It is not compatible with Dart Web.

Installation

1. Install the Dart package

In your application's project:

dart pub add snakemake_bridge

Or add the dependency to pubspec.yaml by hand:

dependencies:
  snakemake_bridge: ^0.1.0

Then fetch the dependencies:

dart pub get

2. Install the Python plugin

Activate the Python environment Snakemake uses and run:

pip install snakemake-logger-plugin-dart

Confirm that Snakemake found the plugin:

snakemake --help | grep logger-dart

The command should list options such as --logger-dart-address. If it does not, check that the plugin and Snakemake were installed into the same Python environment.

Your first monitored workflow

The example below:

1. creates a secure token;
2. starts the local server on an available port;
3. folds the events into a WorkflowRunState;
4. launches Snakemake;
5. releases the resources at the end.

import 'dart:async';

import 'package:snakemake_bridge/snakemake_bridge.dart';

Future<void> main() async {
  final token = WorkflowServer.generateToken();
  final server = WorkflowServer(token: token);
  await server.start();

  print('Server listening on ${server.address}');

  final state = WorkflowRunState();
  final subscription = server.events.listen((event) {
    if (!state.apply(event)) return;

    print(
      'status=${state.status.name} '
      'jobs=${state.jobs.length} '
      'progress=${state.progress}',
    );
  });

  try {
    final runner = await SnakemakeRunner.start(
      workflowDir: '/path/to/workflow',
      serverPort: server.boundPort,
      token: token,
      cores: 4,
    );

    runner.stdout.listen((line) => print('[snakemake] $line'));
    runner.stderr.listen((line) => print('[snakemake:error] $line'));

    final exitCode = await runner.exitCode;
    print('Snakemake finished with exit code $exitCode');
  } finally {
    await subscription.cancel();
    await server.close();
  }
}

Replace /path/to/workflow with the directory containing the Snakefile.

An exit code of 0 means the Snakemake process finished successfully. Beyond that, state.status, state.progress, state.jobs, state.logLines and state.errors expose the state built from the events received.

Updating a graphical interface

WorkflowRunState does not depend on any particular state management solution. It can be wrapped in a ChangeNotifier, Riverpod, Bloc or anything else.

apply() returns false for the events that never change the state — PongEvent, ResourcesInfoEvent and UnknownEvent — and true for the rest. It is a conservative filter rather than exact change detection: a HelloEvent from a reconnection to the same run, for instance, returns true without having changed anything.

server.events.listen((event) {
  if (state.apply(event)) {
    notifyUi();
  }
});

In practice, notifyUi() should be replaced by whatever mechanism the application uses, such as notifyListeners(), updating a Notifier or emitting a new state from a Bloc.

The properties a monitoring screen usually needs are:

Property	Contents
state.status	Overall state: idle, running, finished or failed.
state.progress	Progress between 0.0 and 1.0, when the total is known.
state.jobs	Jobs indexed by their numeric id.
state.rulegraph	Rule DAG sent by Snakemake.
state.logLines	Most recent log lines.
state.errors	Structured errors received during the run.

Cancelling the workflow

Keep the SnakemakeRunner instance around while the workflow runs, and call:

final exitCode = await runner.cancel();

The runner first asks the process to shut down gracefully. If it has not terminated within 15 seconds, termination is forced. The interval can be changed:

await runner.cancel(killAfter: const Duration(seconds: 30));

Manual mode, in two terminals

You can also start just the example server and run Snakemake by hand. This is useful to validate the installation and watch the events as they arrive.

In the first terminal, inside this repository's dart/ directory:

SNAKEMAKE_LOGGER_DART_TOKEN=dev \
  dart run example/snakemake_bridge_example.dart

The program prints an address similar to:

LISTENING port=43127 address=ws://127.0.0.1:43127

In the second terminal, activate Snakemake's Python environment, change into the workflow directory and use the port printed by the first terminal:

export SNAKEMAKE_LOGGER_DART_TOKEN=dev

snakemake --cores 2 \
  --logger dart \
  --logger-dart-address ws://127.0.0.1:43127

The first terminal should then show events such as:

EVENT HelloEvent seq=0
EVENT WorkflowStartedEvent seq=1
EVENT ProgressEvent seq=8 1/3
EVENT ByeEvent seq=9
DONE status=finished jobs=3 progress=1.0

The dev token is only appropriate for this local test. In a real application, always use WorkflowServer.generateToken() and generate a fresh token per run.

Main components

File	Role
lib/src/workflow_server.dart	Embedded WebSocket server (shelf), bound to 127.0.0.1, validates the token, deduplicates by seq, requests replay on every reconnection.
lib/src/workflow_events.dart	Typed event models (sealed classes, no codegen; unknown types become UnknownEvent).
lib/src/workflow_state.dart	Reducer: event stream → queryable run state (jobs, progress, DAG, logs, errors).
lib/src/snakemake_runner.dart	Launches and owns the snakemake process (cancellation = SIGTERM).
example/snakemake_bridge_example.dart	Minimal headless "app" for testing the bridge from the terminal.

WorkflowServer

Embedded WebSocket server that:

• listens on 127.0.0.1 only;
• validates the plugin's token;
• converts the JSON messages into typed Dart events;
• discards duplicates by sequence number;
• requests a resend of events after a reconnection.

WorkflowRunState

Folds the event stream into a queryable state. This pattern is also known as a reducer: each new event is applied on top of the current state.

SnakemakeRunner

Starts the snakemake process, wires up the connection to the server and exposes:

• standard output on runner.stdout;
• error output on runner.stderr;
• the process result on runner.exitCode;
• cancellation through runner.cancel().

Typed events

Incoming events are represented by classes such as WorkflowStartedEvent, JobInfoEvent, JobStartedEvent, JobFinishedEvent, ProgressEvent, LogLineEvent and WorkflowErrorEvent.

Types the installed version of the package does not know yet are delivered as UnknownEvent, so the application keeps working when the protocol gains new events.

If nothing shows up in the application

Check, in this order:

1. that the address passed to --logger-dart-address carries the right port;
2. that the application and the plugin use exactly the same token;
3. that snakemake --help | grep logger-dart finds the plugin;
4. that the plugin was installed in the same Python environment as the snakemake executable;
5. that the Dart server was started before the workflow.

For safety, the server only accepts local connections on 127.0.0.1. It is not designed to receive events directly from another machine.

Development and tests

Inside the dart/ directory:

dart pub get
dart analyze
dart test

Next steps

• See the full example.
• Read the architecture to understand authentication, reconnection and failure modes.
• Use the protocol specification to implement another compatible client or server.
• See the repository installation guide to run the end-to-end validation from source.