demo

Bundled demo workflow for snakesee demo.

Functions:

run_demo

run_demo(cores: int = 4, duration: Duration = 'short', keep_runs: int = 5, no_tui: bool = False, clean: bool = False) -> int

Run the bundled demo workflow with snakesee.

Parameters:

Name	Type	Description	Default
cores	int	Number of cores for snakemake --cores.	4
duration	Duration	Sleep-range preset. Scales the workflow's per-job duration.	'short'
keep_runs	int	Maximum demo dirs to keep in the cache; older are pruned at startup.	5
no_tui	bool	If True, run snakemake to completion and skip launching the TUI. Used by tests and CI; not advertised to end users.	False
clean	bool	If True, delete every demo dir in the cache and exit without running anything.	False

Returns:

Type	Description
int	0 on clean exit, non-zero if snakemake errored.

Source code in snakesee/demo/runner.py

def run_demo(
    cores: int = 4,
    duration: Duration = "short",
    keep_runs: int = 5,
    no_tui: bool = False,
    clean: bool = False,
) -> int:
    """Run the bundled demo workflow with snakesee.

    Args:
        cores: Number of cores for `snakemake --cores`.
        duration: Sleep-range preset. Scales the workflow's per-job duration.
        keep_runs: Maximum demo dirs to keep in the cache; older are pruned at startup.
        no_tui: If True, run snakemake to completion and skip launching the TUI.
                Used by tests and CI; not advertised to end users.
        clean: If True, delete every demo dir in the cache and exit without
               running anything.

    Returns:
        0 on clean exit, non-zero if snakemake errored.
    """
    cache_root = _cache_root()

    if clean:
        if cache_root.is_dir():
            shutil.rmtree(cache_root)
            print(f"Removed {cache_root}", file=sys.stderr)
        return 0

    cache_root.mkdir(parents=True, exist_ok=True)
    _prune_old_runs(cache_root, keep_runs)

    demo_dir = cache_root / _utc_timestamp()
    copy_workflow(demo_dir)

    sleep_min, sleep_max = _DURATION_RANGES[duration]
    use_plugin = logger_plugin_available()
    if not use_plugin:
        print(
            "[snakesee demo] snakemake-logger-plugin-snakesee not installed; "
            "falling back to log-file polling. Install the plugin for realtime events.",
            file=sys.stderr,
        )

    argv = build_snakemake_argv(cores, sleep_min, sleep_max, use_plugin)

    snakemake = shutil.which(argv[0])
    if snakemake is None:
        print(
            "[snakesee demo] could not find `snakemake` on PATH; "
            "install snakemake (e.g. `pixi add snakemake` or `pip install snakemake`).",
            file=sys.stderr,
        )
        return 127
    argv[0] = snakemake

    proc: subprocess.Popen[bytes] = subprocess.Popen(
        argv,
        cwd=demo_dir,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    if no_tui:
        # Wait synchronously and report Snakemake's exit code.
        try:
            return proc.wait()
        except KeyboardInterrupt:
            _terminate(proc)
            return 130

    # Closing the terminal sends SIGHUP, which by default kills Python without
    # running `finally` blocks — leaving the snakemake child as an orphan.
    # Install a handler so the subprocess goes down with the parent.  SIGHUP
    # is Unix-only; on Windows, terminal close raises CTRL_CLOSE_EVENT instead
    # and there is nothing to install here.
    has_sighup = hasattr(signal, "SIGHUP")
    prev_sighup = signal.getsignal(signal.SIGHUP) if has_sighup else None

    if has_sighup:

        def _on_sighup(signum: int, frame: object) -> None:
            _terminate(proc)
            raise SystemExit(128 + signum)

        signal.signal(signal.SIGHUP, _on_sighup)

    try:
        # Give snakemake a moment to write the first event/log so the TUI has
        # something to show on initial mount.
        time.sleep(0.5)
        app = SnakeseeApp(workflow_dir=demo_dir)
        app.run()
    finally:
        # If snakemake already exited on its own (e.g. workflow finished), capture
        # its rc; otherwise terminate it (returncode is None for "still running").
        completed_rc = proc.poll()
        if completed_rc is None:
            _terminate(proc)
        if has_sighup and prev_sighup is not None:
            signal.signal(signal.SIGHUP, prev_sighup)

    return completed_rc if completed_rc is not None else 0

Modules

runner

Demo runner: spawn Snakemake on a bundled workflow and launch the TUI.

Classes

Functions:

build_snakemake_argv

build_snakemake_argv(cores: int, sleep_min: int, sleep_max: int, use_logger_plugin: bool) -> list[str]

Build the Snakemake command line for the demo workflow.

Parameters:

Name	Type	Description	Default
cores	int	Number of cores for snakemake --cores.	required
sleep_min	int	Minimum per-job sleep in seconds, passed via --config.	required
sleep_max	int	Maximum per-job sleep in seconds, passed via --config.	required
use_logger_plugin	bool	Whether to add --logger snakesee for realtime events.	required

Returns:

Type	Description
list[str]	The argv list, with the bare snakemake program name at index 0.

Source code in snakesee/demo/runner.py

def build_snakemake_argv(
    cores: int,
    sleep_min: int,
    sleep_max: int,
    use_logger_plugin: bool,
) -> list[str]:
    """Build the Snakemake command line for the demo workflow.

    Args:
        cores: Number of cores for `snakemake --cores`.
        sleep_min: Minimum per-job sleep in seconds, passed via `--config`.
        sleep_max: Maximum per-job sleep in seconds, passed via `--config`.
        use_logger_plugin: Whether to add `--logger snakesee` for realtime events.

    Returns:
        The argv list, with the bare `snakemake` program name at index 0.
    """
    argv = ["snakemake", "all", "--cores", str(cores), "--keep-going"]
    if use_logger_plugin:
        argv += ["--logger", "snakesee"]
    argv += [
        "--config",
        f"sleep_min={sleep_min}",
        f"sleep_max={sleep_max}",
    ]
    return argv

copy_workflow

copy_workflow(target: Path) -> None

Copy the bundled demo Snakefile and inputs/ into a target directory.

Parameters:

Name	Type	Description	Default
target	Path	Directory to populate; created (with parents) if it does not exist.	required

Source code in snakesee/demo/runner.py

def copy_workflow(target: Path) -> None:
    """Copy the bundled demo Snakefile and inputs/ into a target directory.

    Args:
        target: Directory to populate; created (with parents) if it does not exist.
    """
    target.mkdir(parents=True, exist_ok=True)
    (target / "inputs").mkdir(exist_ok=True)
    snakefile = importlib.resources.files("snakesee.demo") / "Snakefile"
    target.joinpath("Snakefile").write_text(snakefile.read_text())
    inputs_pkg = importlib.resources.files("snakesee.demo.inputs")
    for sample in ("A", "B", "C", "D", "E"):
        src = inputs_pkg / f"{sample}.txt"
        target.joinpath("inputs", f"{sample}.txt").write_text(src.read_text())

logger_plugin_available

logger_plugin_available() -> bool

Return True if the snakesee realtime logger plugin is importable.

Source code in snakesee/demo/runner.py

def logger_plugin_available() -> bool:
    """Return True if the snakesee realtime logger plugin is importable."""
    return find_spec(_LOGGER_PLUGIN_MODULE) is not None

run_demo

run_demo(cores: int = 4, duration: Duration = 'short', keep_runs: int = 5, no_tui: bool = False, clean: bool = False) -> int

Run the bundled demo workflow with snakesee.

Parameters:

Name	Type	Description	Default
cores	int	Number of cores for snakemake --cores.	4
duration	Duration	Sleep-range preset. Scales the workflow's per-job duration.	'short'
keep_runs	int	Maximum demo dirs to keep in the cache; older are pruned at startup.	5
no_tui	bool	If True, run snakemake to completion and skip launching the TUI. Used by tests and CI; not advertised to end users.	False
clean	bool	If True, delete every demo dir in the cache and exit without running anything.	False

Returns:

Type	Description
int	0 on clean exit, non-zero if snakemake errored.

Source code in snakesee/demo/runner.py

def run_demo(
    cores: int = 4,
    duration: Duration = "short",
    keep_runs: int = 5,
    no_tui: bool = False,
    clean: bool = False,
) -> int:
    """Run the bundled demo workflow with snakesee.

    Args:
        cores: Number of cores for `snakemake --cores`.
        duration: Sleep-range preset. Scales the workflow's per-job duration.
        keep_runs: Maximum demo dirs to keep in the cache; older are pruned at startup.
        no_tui: If True, run snakemake to completion and skip launching the TUI.
                Used by tests and CI; not advertised to end users.
        clean: If True, delete every demo dir in the cache and exit without
               running anything.

    Returns:
        0 on clean exit, non-zero if snakemake errored.
    """
    cache_root = _cache_root()

    if clean:
        if cache_root.is_dir():
            shutil.rmtree(cache_root)
            print(f"Removed {cache_root}", file=sys.stderr)
        return 0

    cache_root.mkdir(parents=True, exist_ok=True)
    _prune_old_runs(cache_root, keep_runs)

    demo_dir = cache_root / _utc_timestamp()
    copy_workflow(demo_dir)

    sleep_min, sleep_max = _DURATION_RANGES[duration]
    use_plugin = logger_plugin_available()
    if not use_plugin:
        print(
            "[snakesee demo] snakemake-logger-plugin-snakesee not installed; "
            "falling back to log-file polling. Install the plugin for realtime events.",
            file=sys.stderr,
        )

    argv = build_snakemake_argv(cores, sleep_min, sleep_max, use_plugin)

    snakemake = shutil.which(argv[0])
    if snakemake is None:
        print(
            "[snakesee demo] could not find `snakemake` on PATH; "
            "install snakemake (e.g. `pixi add snakemake` or `pip install snakemake`).",
            file=sys.stderr,
        )
        return 127
    argv[0] = snakemake

    proc: subprocess.Popen[bytes] = subprocess.Popen(
        argv,
        cwd=demo_dir,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    if no_tui:
        # Wait synchronously and report Snakemake's exit code.
        try:
            return proc.wait()
        except KeyboardInterrupt:
            _terminate(proc)
            return 130

    # Closing the terminal sends SIGHUP, which by default kills Python without
    # running `finally` blocks — leaving the snakemake child as an orphan.
    # Install a handler so the subprocess goes down with the parent.  SIGHUP
    # is Unix-only; on Windows, terminal close raises CTRL_CLOSE_EVENT instead
    # and there is nothing to install here.
    has_sighup = hasattr(signal, "SIGHUP")
    prev_sighup = signal.getsignal(signal.SIGHUP) if has_sighup else None

    if has_sighup:

        def _on_sighup(signum: int, frame: object) -> None:
            _terminate(proc)
            raise SystemExit(128 + signum)

        signal.signal(signal.SIGHUP, _on_sighup)

    try:
        # Give snakemake a moment to write the first event/log so the TUI has
        # something to show on initial mount.
        time.sleep(0.5)
        app = SnakeseeApp(workflow_dir=demo_dir)
        app.run()
    finally:
        # If snakemake already exited on its own (e.g. workflow finished), capture
        # its rc; otherwise terminate it (returncode is None for "still running").
        completed_rc = proc.poll()
        if completed_rc is None:
            _terminate(proc)
        if has_sighup and prev_sighup is not None:
            signal.signal(signal.SIGHUP, prev_sighup)

    return completed_rc if completed_rc is not None else 0