### Kiso Multi-Site Configuration (Edge + Cloud)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Demonstrates combining different site configurations within a single experiment. This example includes a Chameleon Edge setup for containers and a Chameleon KVM setup for cloud VMs, allowing for hybrid deployments.

```yaml
sites:
  - kind: chameleon-edge
    walltime: "04:00:00"
    lease_name: edge-lease
    rc_file: secrets/chi-edge-openrc.sh
    resources:
      machines:
        - labels:
            - edge-execute
          machine_name: raspberrypi4-64
          count: 1
          container:
            name: execute
            image: pegasus/myapp

  - kind: chameleon
    walltime: "04:00:00"
    lease_name: cloud-lease
    rc_file: secrets/chi-tacc-openrc.sh
    key_name: my-key
    image: CC-Ubuntu22.04
    resources:
      machines:
        - labels:
            - central-manager
            - submit
            - cloud-execute
          flavour: compute_zen3
          number: 1
      networks:
        - sharednet1
```

--------------------------------

### Complete Pegasus Experiment on Vagrant (YAML)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

A comprehensive example of setting up a Pegasus workflow experiment for local testing using Vagrant. It defines sites, resources, software (Apptainer), HTCondor deployment, and the experiment details including inputs, setup, and outputs.

```yaml
name: tnseq-experiment

sites:
  - kind: vagrant
    backend: virtualbox
    box: bento/rockylinux-9
    user: vagrant
    config_extra: 'config.vm.synced_folder ".", "/vagrant", disabled: true'
    resources:
      machines:
        - labels:
            - submit
          flavour: "large"
          number: 1
        - labels:
            - execute
          flavour: "medium"
          number: 2
      networks:
        - labels:
            - r1
          cidr: "172.16.42.0/16"

software:
  apptainer:
    labels:
      - execute

deployment:
  htcondor:
    - kind: central-manager
      labels:
        - submit
    - kind: submit
      labels:
        - submit
    - kind: execute
      labels:
        - execute

experiments:
  - kind: pegasus
    name: tnseq-workflow
    description: TnSeq bioinformatics pipeline
    count: 1
    main: bin/workflow_generator.py
    submit_node_labels:
      - submit
    inputs:
      - labels:
          - submit
        src: data/
        dst: ~kiso/tnseq-workflow/
    setup:
      - labels:
          - submit
        script: |
          #!/bin/bash
          pip install pegasus-wms pyyaml
          chmod +x bin/workflow_generator.py
    outputs:
      - labels:
          - submit
        src: ~kiso/tnseq-workflow/output/
        dst: ./output/
```

--------------------------------

### Implement Built-in Test Mode with Sample Data Download (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Shows how to add a `--test` flag to a workflow generator script. When enabled, this flag triggers the download of a small sample dataset, allowing new users to run the workflow with minimal setup and friction.

```python
parser.add_argument("--test", action="store_true",
                    help="Download test data and run with minimal settings")

def download_test_data(self):
    """Download small test dataset for validation."""
    test_dir = os.path.join(self.wf_dir, "data", "test")
    os.makedirs(test_dir, exist_ok=True)

    test_files = {
        "sample1_R1.fastq.gz": "https://example.com/test/sample1_R1.fastq.gz",
        "sample1_R2.fastq.gz": "https://example.com/test/sample1_R2.fastq.gz",
    }
    for name, url in test_files.items():
        path = os.path.join(test_dir, name)
        if not os.path.exists(path):
            urllib.request.urlretrieve(url, path)

    # Return sample list for workflow generation
    return [{"id": "sample1", "fastq_1": ..., "fastq_2": ...}]

# In main():
if args.test:
    samples = workflow.download_test_data()
elif args.samplesheet:
    samples = parse_samplesheet(args.samplesheet)
else:
    print("Error: Either --test or --samplesheet required")
    sys.exit(1)
```

--------------------------------

### Python Container Configuration Example

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/pegasus-dockerfile/SKILL.md

An example Python code snippet demonstrating how to configure a container object in Pegasus. It specifies the container name, type, image URI, and image site. It also includes a note about not using mounts for caches/databases.

```python
container = Container(
    "my_container",
    container_type=Container.SINGULARITY,
    image="docker://username/image:latest",
    image_site="docker_hub",
    # Do NOT add mounts=[] for caches/databases — use CondorIO transfer_input_files instead
)
```

--------------------------------

### Pegasus Python API Example

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A comprehensive Python script demonstrating the use of the Pegasus API to define workflow components such as Properties, Site Catalog, Container, Transformation Catalog, Replica Catalog, and the Workflow itself, including job definition and argument parsing.

```python
from Pegasus.api import *

# Properties
props = Properties()
props["pegasus.transfer.threads"] = "16"

# Site Catalog
sc = SiteCatalog()
site = Site("condorpool").add_condor_profile(universe="vanilla")
sc.add_sites(site)

# Container
container = Container("name", Container.SINGULARITY, "docker://img:tag", "docker_hub")

# Transformation Catalog
tc = TransformationCatalog()
tx = Transformation("name", site="condorpool", pfn="/path/to/script.py",
                    is_stageable=True, container=container)
tx.add_pegasus_profile(memory="4 GB", cores=2)
# Transfer external data directories via CondorIO (preferred over container mounts)
tx.add_profiles(Namespace.CONDOR, key="transfer_input_files", value="/path/to/cache_dir")
tc.add_containers(container)
tc.add_transformations(tx)

# Replica Catalog
rc = ReplicaCatalog()
rc.add_replica("local", "logical_name", "file:///absolute/path")

# Workflow
wf = Workflow("name", infer_dependencies=True)

# Files
input_f = File("input.txt")
output_f = File("output.txt")

# Job
job = (Job("transformation_name", _id="unique_id", node_label="label")
       .add_args("--input input.txt --output output.txt")
       .add_inputs(input_f)
       .add_outputs(output_f, stage_out=True, register_replica=False)
       .add_pegasus_profiles(label="group"))
wf.add_jobs(job)

```

--------------------------------

### Kiso Software Installation: Apptainer

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Specifies the installation of Apptainer (formerly Singularity) software within the Kiso experiment. It uses labels to define which machines should have Apptainer installed.

```yaml
software:
  apptainer:
    labels:
      - execute
```

--------------------------------

### Kiso Software Installation: Docker

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Specifies the installation of Docker software within the Kiso experiment. It uses labels to define which machines should have Docker installed.

```yaml
software:
  docker:
    labels:
      - execute
```

--------------------------------

### Kiso Installation Command: FABRIC

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Provides the pip installation command for Kiso with FABRIC testbed support. This command installs Kiso along with the necessary dependencies for provisioning resources on the FABRIC testbed.

```bash
pip install kiso[fabric]
```

--------------------------------

### Kiso Installation Command: Chameleon

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Provides the pip installation command for Kiso with Chameleon testbed support. This command installs Kiso along with the necessary dependencies for interacting with Chameleon cloud and edge resources.

```bash
pip install kiso[chameleon]
```

--------------------------------

### Dockerfile Template for Ubuntu 22.04

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A Dockerfile template to build a Ubuntu 22.04 environment, installing Python 3, pip, and specified tools. It also ensures python3 is aliased as python and cleans up apt cache.

```dockerfile
FROM ubuntu:22.04

ENV DEBIAN_FRONTEND=noninteractive

RUN apt-get update && \
    apt-get install -y \
        python3 python3-pip \
        tool1 tool2 tool3 \
    && rm -rf /var/lib/apt/lists/*

# Install tools not available via apt
RUN pip3 install --no-cache-dir some-python-tool==1.2.3

# Ensure python3 is available as python
RUN ln -sf /usr/bin/python3 /usr/bin/python

```

--------------------------------

### Kiso Installation Command: Vagrant

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Provides the pip installation command for Kiso with Vagrant support. This command installs Kiso along with the necessary dependencies to manage Vagrant environments.

```bash
pip install kiso[vagrant]
```

--------------------------------

### Configure Headless Execution for GUI Tools

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Provides a bash wrapper pattern to disable GUI displays and use virtual framebuffers for tools like FastQC, along with the necessary Dockerfile configuration.

```bash
#!/bin/bash
unset DISPLAY
export JAVA_TOOL_OPTIONS="-Djava.awt.headless=true"

if command -v xvfb-run &> /dev/null; then
    xvfb-run --auto-servernum fastqc "$@"
else
    fastqc "$@"
fi
```

```dockerfile
RUN apt-get install -y xvfb libgl1-mesa-glx libfontconfig1
```

--------------------------------

### Configure Tool Resources (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to centralize and apply per-tool resource configurations (memory and CPU cores) during transformation catalog creation in Pegasus. This allows for easy adjustment of resource requirements for different tools.

```python
TOOL_CONFIGS = {
    "fastqc":   {"memory": "2 GB",  "cores": 2},
    "fastp":    {"memory": "4 GB",  "cores": 4},
    "megahit":  {"memory": "16 GB", "cores": 8},
    "quast":    {"memory": "8 GB",  "cores": 4},
    "gtdbtk":   {"memory": "64 GB", "cores": 8},   # Taxonomy is expensive
    "train":    {"memory": "4 GB",  "cores": 2},    # ML training
    "predict":  {"memory": "2 GB",  "cores": 1},    # ML inference
}

# Apply during transformation catalog creation
for tool_name, config in TOOL_CONFIGS.items():
    tx = Transformation(tool_name, site=exec_site_name, ...)
    tx.add_pegasus_profile(
        memory=config["memory"],
        cores=config.get("cores", 1),
    )
```

--------------------------------

### Flatten Nested Output Directories

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to use wrapper scripts to copy and flatten output files from nested tool directories into the working directory for Pegasus stage-out.

```bash
quast "$@"
cp "${OUTPUT_DIR}/report.tsv" "${SAMPLE}_quast_report.tsv"
cp "${OUTPUT_DIR}/report.html" "${SAMPLE}_quast_report.html"

find "${OUTPUT_DIR}" -name "*.summary.tsv" -exec cp {} "${SAMPLE}_taxonomy.tsv" \;

cp "${OUTPUT_DIR}/${PREFIX}.gff" "${SAMPLE}_annotation.gff"
cp "${OUTPUT_DIR}/${PREFIX}.faa" "${SAMPLE}_proteins.faa"
```

--------------------------------

### Implement Dual Pipeline Architecture

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Shows how to structure a workflow with shared initial steps and conditional, branching downstream pipelines in Python.

```python
extract_job = Job("extract_timeseries", ...)

analyze_job.add_inputs(timeseries_file)
anomaly_job.add_inputs(timeseries_file, analysis_file)

if not args.skip_forecast:
    fetch_hist_job.add_inputs(catalog_file)
    prepare_job.add_inputs(timeseries_file, historical_file)
    train_job.add_inputs(features_file)
    predict_job.add_inputs(model_file, timeseries_file)
    visualize_job.add_inputs(prediction_file, timeseries_file)
```

--------------------------------

### Building and Running a Docker Container

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Commands to build a Docker image from a Dockerfile and then run an interactive bash session within the newly created container. This is useful for testing container contents and configurations.

```bash
docker build -t username/image:latest -f Docker/My_Dockerfile .
docker run --rm -it username/image:latest bash

```

--------------------------------

### Design Parallel DAG Execution Patterns

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Examples of structuring Pegasus jobs for per-sample parallelism, branching logic, and fan-out/fan-in operations to optimize workflow performance.

```python
# Per-Sample Parallelism
for sample in samples:
    clip_job = Job("clip", _id=f"clip_{sample}", ...)
    align_job = Job("align", _id=f"align_{sample}", ...)

# Branching
bw_job.add_inputs(marked_bam)
genomecov_job.add_inputs(marked_bam)

# Fan-out / Fan-in
for variant in variants:
    map_job = Job("map", _id=f"map_{variant}_{sample}", ...)
concat_job.add_inputs(*all_tab_files)
```

--------------------------------

### Docker Build and Test Commands

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/pegasus-dockerfile/SKILL.md

Provides essential bash commands for building, testing, and pushing Docker images. Includes commands for interactive testing, verifying tool installation, and pushing the image to a registry.

```bash
# Build
docker build -t username/image:latest -f Docker/My_Dockerfile .

# Test (interactive shell)
docker run --rm -it username/image:latest bash

# Verify tools are installed
docker run --rm username/image:latest which tool1 tool2 tool3

# Push to Docker Hub
docker push username/image:latest
```

--------------------------------

### Initialize Pegasus Workflow Generator Class

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A template class for managing Pegasus workflow generation, including directory setup and catalog initialization. It provides a structure to write configuration files and define execution sites.

```python
class MyWorkflow:
    wf = None
    sc = None
    tc = None
    rc = None
    props = None

    def __init__(self, dagfile="workflow.yml"):
        self.dagfile = dagfile
        self.wf_dir = str(Path(__file__).parent.resolve())
        self.shared_scratch_dir = os.path.join(self.wf_dir, "scratch")
        self.local_storage_dir = os.path.join(self.wf_dir, "output")

    def write(self):
        if self.sc is not None:
            self.sc.write()
        self.props.write()
        self.rc.write()
        self.tc.write()
        self.wf.write(file=self.dagfile)
```

--------------------------------

### Micromamba Container for Complex Tool Stacks (Dockerfile)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

This Dockerfile uses micromamba to manage complex bioinformatics tool stacks with conflicting dependencies. It installs Python 3.8 and various bioinformatics tools (FastQC, STAR, SAMtools, etc.) using a single solver, ensuring compatibility. It also installs system dependencies and wrapper scripts.

```dockerfile
FROM mambaorg/micromamba:1.5-jammy

# Single solver resolves all version conflicts
RUN micromamba install -y -n base -c conda-forge -c bioconda \
    python=3.8 \
    fastqc fastp multiqc \
    megahit spades quast \
    prodigal metabat2 samtools \
    checkm2 gtdbtk prokka

# System dependencies for headless tools
USER root
RUN apt-get update && apt-get install -y xvfb libgl1-mesa-glx && \
    rm -rf /var/lib/apt/lists/*

# Install wrapper scripts
COPY bin/*.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/*.sh
```

--------------------------------

### Kiso Software Installation: Ollama

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Configures the installation of Ollama for serving Large Language Models. This includes specifying target machine labels, desired models, and environment variables for the Ollama service.

```yaml
software:
  ollama:
    - labels:
        - llm-node
      models:
        - llama3:8b
      environment:
        OLLAMA_MAX_QUEUE: 512
```

--------------------------------

### Implement Graceful ML Dependency Fallback

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Shows how to handle optional ML dependencies in a wrapper script by checking for library availability and falling back to rule-based logic.

```python
try:
    import torch
    ML_AVAILABLE = True
except ImportError:
    ML_AVAILABLE = False
    logger.warning("PyTorch not available — using rule-based predictions")

if ML_AVAILABLE and model_path:
    result = ml_predict(model_path, features)
else:
    result = rule_based_predict(features)
```

--------------------------------

### Define Container and Transformations in Pegasus

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to register Singularity or Docker containers and map them to specific transformation executables within the Pegasus Transformation Catalog.

```python
container = Container(
    "my_container",
    container_type=Container.SINGULARITY,
    image="docker://username/image:latest",
    image_site="docker_hub",
)

self.tc = TransformationCatalog()
my_step = Transformation(
    "my_step",
    site=exec_site_name,
    pfn=os.path.join(self.wf_dir, "bin/my_step.py"),
    is_stageable=True,
    container=container,
).add_pegasus_profile(memory="4 GB", cores=2)

self.tc.add_containers(container)
self.tc.add_transformations(my_step)
```

--------------------------------

### Python Wrapper Script Example

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/pegasus-convert/SKILL.md

Illustrates a basic Python wrapper script for a Pegasus transformation. This script would execute the core command of a Snakemake rule or Nextflow process, taking inputs and producing outputs as defined in the Pegasus workflow.

```python
import subprocess
import os

def main(input_file, output_file):
    cmd = f"some_command --input {input_file} --output {output_file}"
    subprocess.run(cmd, shell=True, check=True)

if __name__ == "__main__":
    # Example usage: Assume input and output files are passed as arguments or environment variables
    input_path = os.environ.get("PEGASUS_INPUT_FILE")
    output_path = os.environ.get("PEGASUS_OUTPUT_FILE")
    main(input_path, output_path)
```

--------------------------------

### Create Manual Local Testing Scripts

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A template for a standalone bash script to validate pipeline steps locally before submitting to the Pegasus workflow engine.

```bash
#!/bin/bash
set -e

echo "=== Step 1: Fetching data ==="
./bin/fetch_data.py --output test_data.csv

echo "=== Step 2: Analysis ==="
./bin/analyze.py --input test_data.csv --output test_analysis.json

echo "=== Step 3: ML Training ==="
./bin/train_model.py --input test_data.csv --output test_model.pt --epochs 10

echo "=== Step 4: Prediction ==="
./bin/predict.py --input test_analysis.json --model test_model.pt --output test_pred.json
```

--------------------------------

### Define Workflow Dependencies and Write Catalogs

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to explicitly define job dependencies in a Pegasus workflow and write out the necessary configuration catalogs (site, transformation, replica, etc.) to a YAML file.

```python
wf.add_dependency(parent_job, children=[child_job])

props.write()
sc.write()
tc.write()
rc.write()
wf.write(file="workflow.yml")
```

--------------------------------

### Handle Explicit File Arguments in Pegasus

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to pass multiple input files to a Pegasus job and configure the corresponding Python argparse parser to accept repeated arguments.

```python
file_args = ' '.join([f"--input {f.lfn}" for f in tab_files])
concat_job.add_args(f"--output merged.tsv " + file_args)

# In the wrapper script:
parser.add_argument("--input", action="append", required=True,
                    help="Input file (can be specified multiple times)")
```

--------------------------------

### Locate Staged Support Files

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Best practice for referencing support files (like R scripts or JARs) staged by Pegasus in the job's working directory.

```python
# GOOD - Pegasus stages the file to the working directory
script_path = os.path.join(os.getcwd(), "analysis.R")

# BAD - __file__ path is unreliable in staged environments
script_path = os.path.join(os.path.dirname(__file__), "analysis.R")
```

--------------------------------

### API Data Acquisition in Pegasus Jobs

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to fetch external data at runtime without pre-staged inputs, including secure credential handling and robust API request patterns.

```python
fetch_job = (
    Job("fetch_data", _id=f"fetch_{region}", node_label=f"fetch_{region}")
    .add_args(f"--region {region} --start-date {start_date} --output {region}_data.csv")
    .add_outputs(data_file, stage_out=False, register_replica=False)
)

analyze_job = Job("analyze", _id=f"analyze_{region}").add_inputs(data_file)
```

```python
import requests
response = requests.get(api_url, params=params, timeout=60)
response.raise_for_status()
```

```python
fetch_job.add_env(API_KEY=os.environ.get("API_KEY", ""))

api_key = os.environ.get("API_KEY")
if not api_key:
    print("Error: API_KEY environment variable not set", file=sys.stderr)
    sys.exit(1)
```

--------------------------------

### Creating Hierarchical File Paths (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

This Python snippet demonstrates how to define `File` objects with hierarchical paths for organized output. It also shows the corresponding wrapper script logic to create the necessary subdirectories before writing the output file.

```python
# Output will be staged to: output/read_counts/totalcounts_mid/sample.tab
tab_file = File("read_counts/totalcounts_mid/sample.totalcounts.tab")

# In the wrapper script
os.makedirs(os.path.dirname(args.output), exist_ok=True)
```

--------------------------------

### Standard Wrapper Script Template

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A robust template for Python wrapper scripts that handle argument parsing, directory creation, tool execution via subprocess, and proper exit code propagation.

```python
#!/usr/bin/env python3
import argparse, os, subprocess, sys

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    os.makedirs(os.path.dirname(args.output), exist_ok=True)
    cmd = f"mytool --input {args.input} --output {args.output}"
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if result.returncode != 0:
        sys.exit(result.returncode)

if __name__ == "__main__":
    main()
```

--------------------------------

### Region and Parameter-Based DAG Generation

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Techniques for parameterizing workflow generation loops to handle multiple regions or locations, including safe string naming and loading parameters from external JSON configurations.

```python
for region in args.regions:
    self._add_region_pipeline(region)

safe_name = location_name.replace(" ", "_").replace("-", "_").replace("/", "_")

if not args.polygon_ids:
    with open(args.polygons_file) as f:
        polygons = json.load(f)
    args.polygon_ids = [p["id"] for p in polygons]
```

--------------------------------

### Wrapper Script Argument Parsing (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

This Python snippet demonstrates how a wrapper script can parse command-line arguments, specifically focusing on the `--cache-dir` argument. This argument is used to specify the local path of a cache directory within the job's working directory.

```python
parser.add_argument("--cache-dir", required=True,
                    help="Cache directory (local path in job working directory)")
# ...
cmd = ["mytool", "--data", args.cache_dir, ...]
```

--------------------------------

### Kiso Chameleon Edge Site Configuration

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Configures edge devices or containers on the Chameleon testbed. This setup requires walltime, lease name, credentials, and specifies machine types, counts, and container details if applicable.

```yaml
sites:
  - kind: chameleon-edge
    walltime: "04:00:00"
    lease_name: edge-lease
    rc_file: secrets/chi-edge-app-cred-openrc.sh
    resources:
      machines:
        - labels:
            - edge-execute
          machine_name: raspberrypi4-64
          count: 2
          container:
            name: execute
            image: rockylinux:8
```

--------------------------------

### Enable Real-time Logging in Container (Dockerfile)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

This Dockerfile instruction sets the environment variable `PYTHONUNBUFFERED` to `1`. This ensures that Python's standard output and error streams are unbuffered, allowing logs to appear in real-time, which is crucial for monitoring and debugging containerized applications.

```dockerfile
ENV PYTHONUNBUFFERED=1
```

--------------------------------

### Registering Files with Replica Catalog (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

This Python code snippet demonstrates how to initialize and use the Replica Catalog to register various types of files, including reference data, support scripts, JARs, and input data files. It shows how to add replicas with different logical names and physical file URIs.

```python
self.rc = ReplicaCatalog()

# Reference files
self.rc.add_replica("local", "reference.fasta",
                    "file://" + os.path.abspath("references/reference.fasta"))

# Support scripts/tools called by wrappers (NOT transformations)
jar_path = os.path.join(self.wf_dir, "bin/tool.jar")
self.rc.add_replica("local", "tool.jar", "file://" + jar_path)

r_script = os.path.join(self.wf_dir, "bin/analysis.R")
self.rc.add_replica("local", "analysis.R", "file://" + r_script)

# Input data files
for sample in samples:
    path = os.path.join(data_dir, f"{sample}.fq.gz")
    self.rc.add_replica("local", f"{sample}.fq.gz",
                        "file://" + os.path.abspath(path))
```

--------------------------------

### Snakemake Rule to Pegasus Wrapper Comparison

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/pegasus-convert/SKILL.md

Illustrates the mapping between a Snakemake rule and its corresponding Pegasus wrapper script, showing how inputs, outputs, shell commands, and resource allocations are translated. This helps in verifying the conversion accuracy.

```text
Snakemake rule: align          →  Wrapper: bin/align.py
  input: "{sample}.fq.gz"      →    --input {sample}.fq.gz
  output: "{sample}.bam"        →    --output {sample}.bam
  shell: "bwa mem ..."          →    subprocess.run(["bwa", "mem", ...])
  threads: 4                    →    .add_pegasus_profile(cores=4)
```

--------------------------------

### Manage Experiment Infrastructure with Kiso CLI

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

The Kiso CLI provides commands to validate experiment configurations, provision necessary infrastructure, execute experiments, and clean up resources. It defaults to using an 'experiment.yml' file in the current working directory.

```sh
kiso check [experiment.yml]   # validate the configuration
kiso up [experiment.yml]      # provision infrastructure
kiso run [experiment.yml]     # run the experiments
kiso down [experiment.yml]    # destroy infrastructure
```

--------------------------------

### Execute Kiso Lifecycle Commands

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/SKILL.md

Standard command-line sequence for validating, provisioning, running, and tearing down a Kiso experiment environment.

```bash
# Validate the configuration
kiso check

# Provision infrastructure
kiso up

# Run the experiment
kiso run

# Destroy infrastructure when done
kiso down
```

--------------------------------

### Dockerfile Generation for Micromamba-based Images

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/pegasus-dockerfile/SKILL.md

Generates a Dockerfile for micromamba-based environments, suitable for complex bioinformatics workflows with potential dependency conflicts. It installs system packages, configures micromamba, installs tools using a single command for the solver, and cleans up cache. It also handles headless support and root user switching.

```dockerfile
FROM mambaorg/micromamba:1.5-jammy

USER root
RUN apt-get update && apt-get install -y --no-install-recommends \
    [system packages, xvfb if needed] \
    && rm -rf /var/lib/apt/lists/*

USER $MAMBA_USER
RUN micromamba install -y -n base -c conda-forge -c bioconda \
    python=3.8 \
    [all tools in ONE install command for solver] \
    && micromamba clean --all --yes
```

--------------------------------

### Embed Scripts in Docker Images

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A Dockerfile snippet to copy local wrapper scripts into the container and ensure they are executable.

```dockerfile
COPY bin/*.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/*.sh
```

--------------------------------

### Build Conditional DAGs with CLI Flags (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Illustrates how to construct different Directed Acyclic Graph (DAG) topologies from a single Python script using command-line flags. This approach avoids maintaining separate workflow files and allows for faster testing by skipping steps.

```python
parser.add_argument("--skip-fastqc", action="store_true")
parser.add_argument("--skip-binning", action="store_true")
parser.add_argument("--skip-taxonomy", action="store_true")
parser.add_argument("--skip-forecast", action="store_true")

# In create_workflow():
if not args.skip_fastqc:
    fastqc_job = Job("fastqc", ...)
    self.wf.add_jobs(fastqc_job)

if not args.skip_binning:
    binning_job = Job("metabat2", ...)
    self.wf.add_jobs(binning_job)

    if not args.skip_taxonomy:
        taxonomy_job = Job("gtdbtk", ...)  # 64 GB RAM
        self.wf.add_jobs(taxonomy_job)
```

--------------------------------

### Workflow Visualization Commands

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Commands to generate a DOT graph of the Pegasus DAG and monitor the status of a submitted workflow.

```bash
pegasus-plan --submit -s condorpool -o local workflow.yml
pegasus-status <run-dir>
```

--------------------------------

### Configure Stageable Transformation Scripts

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Configures a Pegasus Transformation to transfer a script from the submit host to the execution workers by setting is_stageable to True.

```python
tx = Transformation(
    "my_step",
    site=exec_site_name,
    pfn=os.path.join(self.wf_dir, "bin/my_step.py"),
    is_stageable=True,
    container=my_container,
)
```

--------------------------------

### Convert Multiple Data Sources to Unified CSV (Python)

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

This Python code demonstrates how to handle workflows supporting different data sources by converting them to a common intermediate CSV format at workflow generation time. It defines a method to create a replica catalog, fetching and converting data from sources like OpenAQ or SAGE into a unified 'catalog.csv'.

```python
def create_replica_catalog(self):
    self.rc = ReplicaCatalog()

    # Convert source-specific format to unified CSV at generation time
    if self.data_source == "openaq":
        self.fetch_openaq_catalog()  # Creates openaq_catalog.csv
    elif self.data_source == "sage":
        self.load_sage_catalog()     # Also creates openaq_catalog.csv

    # All downstream jobs use the same unified format
    self.rc.add_replica("local", "catalog.csv",
                        "file://" + os.path.join(self.wf_dir, "catalog.csv"))
```

--------------------------------

### Pegasus Workflow Lifecycle Commands

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A sequence of bash commands to manage the lifecycle of a Pegasus workflow, including generation, planning, submission, monitoring, analysis, and statistics retrieval.

```bash
# 1. Generate workflow
./workflow_generator.py [options] --output workflow.yml

# 2. Plan and submit
pegasus-plan --submit -s condorpool -o local workflow.yml

# 3. Monitor
pegasus-status <run-directory>

# 4. Debug failures
pegasus-analyzer <run-directory>

# 5. Get statistics
pegasus-statistics <run-directory>

```

--------------------------------

### Define Pegasus Training and Prediction Jobs

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

Demonstrates how to define a training job and a series of prediction jobs in Python using the Pegasus API. It highlights input/output management and resource profiling.

```python
train_job = (
    Job("train_model", _id="train")
    .add_inputs(training_data)
    .add_outputs(model_file, model_metadata, stage_out=False)
    .add_pegasus_profile(memory="4 GB")
)

for location in locations:
    predict_job = (
        Job("predict", _id=f"predict_{location}")
        .add_inputs(location_data, model_file, model_metadata)
        .add_outputs(prediction_file, stage_out=True)
        .add_pegasus_profile(memory="2 GB")
    )
```

--------------------------------

### Snakemake to Pegasus Mapping

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/pegasus-convert/SKILL.md

Maps Snakemake constructs to their Pegasus equivalents. This includes rules to transformations/jobs, inputs/outputs to Pegasus File objects, shell commands to wrapper scripts, wildcards to loops, and configuration parameters to argparse arguments. It also covers resource mapping and the absence of a direct 'rule all' equivalent.

```markdown
| Snakemake | Pegasus |
|-----------|---------|
| `rule name:` | `Transformation("name", ...)` + `Job("name", ...)` |
| `input: "file.txt"` | `job.add_inputs(File("file.txt"))` |
| `output: "result.txt"` | `job.add_outputs(File("result.txt"), stage_out=..., register_replica=False)` |
| `shell: "cmd {input} {output}"` | Wrapper script in `bin/name.py` |
| `{wildcards.sample}` | `for sample in samples:` loop |
| `expand(...)` | Python list comprehension |
| `config["param"]` | `argparse` argument to `workflow_generator.py` |
| `conda: "env.yaml"` | `Dockerfile` with same packages |
| `threads: N` | `.add_pegasus_profile(cores=N)` |
| `resources: mem_mb=N` | `.add_pegasus_profile(memory="N MB")` |
| `params: data_dir="path"` | Explicit file paths (no directory scanning) |
| `rule all: input: [files]` | No equivalent — Pegasus runs all jobs in the DAG |
```

--------------------------------

### Normalize Output with Shell Wrapper Scripts

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/references/PEGASUS.md

A shell wrapper script that flattens nested directory structures produced by bioinformatics tools, ensuring files are placed in the root directory for Pegasus staging.

```bash
#!/bin/bash
# megahit.sh — wrapper that flattens nested tool output

OUTPUT_DIR=""
SAMPLE=""
while [[ $# -gt 0 ]]; do
    case "$1" in
        -o|--out-dir) OUTPUT_DIR="$2"; shift 2 ;;
        --sample) SAMPLE="$2"; shift 2 ;;
        *) ARGS+=("$1"); shift ;;
    esac
done

megahit "${ARGS[@]}" -o "$OUTPUT_DIR"

cp "${OUTPUT_DIR}/final.contigs.fa" "${SAMPLE}_contigs.fa"
```

--------------------------------

### Kiso Chameleon KVM Site Configuration

Source: https://github.com/pegasus-isi/claude-plugin-marketplace/blob/main/plugins/pegasus-ai/skills/kiso/references/experiment-configuration.md

Configures cloud virtual machines on the Chameleon testbed using KVM. It requires specifying walltime, lease name, credentials, SSH key, image, and resource allocation for machines and networks.

```yaml
sites:
  - kind: chameleon
    walltime: "04:00:00"
    lease_name: my-lease
    rc_file: secrets/chi-tacc-app-cred-openrc.sh
    key_name: my-ssh-key
    image: CC-Ubuntu22.04
    resources:
      machines:
        - labels:
            - submit
            - central-manager
          flavour: compute_zen3
          number: 1
        - labels:
            - execute
          flavour: compute_zen3
          number: 4
      networks:
        - sharednet1
```