----------------------------------------------------------------------
This is the API documentation for the stagecoach library.
----------------------------------------------------------------------


## Classes

Core classes



## Dataclasses

Data-holding classes



## Enumerations

Enum types



## Functions

Public functions



## Constants

Module-level constants and data



## Other

Additional exports



----------------------------------------------------------------------
This is the User Guide documentation for the package.
----------------------------------------------------------------------

# Manifest Checks


The goal of this module is simple: once the user has filled out the
manifest, we want to make sure all is compliant with the lab and
Frontier’s working philosophy. Only once this is done can the user then
run `stagecoach haul` to stage the data they requested in their
environment.

Validating a manifest is critical as it ensures that downstream data
access will be successful and/or fail gracefully with informative error
messages. This is the step where we check that the user has access to
the data they requested, and that the data dependencies they specified
are met. This is also the step where we check that the manifest is
compliant with the lab’s working philosophy, which is to only access
data that is necessary for the project, and to do so in a way that is
secure and reproducible.

We implement this using a series of checks against our [4 core
principles](https://goldenplanetaryhealthlab.github.io/01_orientation/start-here.html#the-working-philosophy):

1.  Always Strive for Remote and Reproducible Workflows 🌐

2.  Code is The Source of Truth 💎

3.  Analysis Should Be Narrated 📇

4.  Robust Habits Compound 🌱

To implement this test, we’ll first define a dataclass that will hold
the result of any given test. This will make it easy to generate a large
number of tests, and gather or manipulate the results in a standard
procedure.

``` python
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from pathlib import Path

class Severity(Enum):
    """
    Severity levels for manifest validation results.

    Attributes
    ----------
    PASS : str
        The check succeeded.
    WARNING : str
        The check surfaced a non-blocking issue.
    ERROR : str
        The check found a blocking issue.
    """
    PASS = "pass"
    WARNING = "warning"
    ERROR = "error"

SEVERITY_RANK = {
    Severity.PASS: 0,
    Severity.WARNING: 1,
    Severity.ERROR: 2,
}

PRINCIPLES = {
    1: "Remote and reproducible workflows 🌐",
    2: "Code is the source of truth 💎",
    3: "Analysis should be narrated 📇",
    4: "Robust habits compound 🌱"
}

@dataclass
class CheckResult:
    """
    Represent the outcome of a single manifest check.

    Attributes
    ----------
    name : str
        Stable identifier for the check.
    state : Severity
        Severity assigned to the check result.
    message : str
        Human-readable explanation of the outcome.
    principle : int | list[int]
        Frontier principle number, or numbers, associated with the check.
    """
    name: str
    state: Severity
    message: str
    principle: int | list[int]
```

``` python
ex_check = CheckResult(
    name="Example Check",
    state=Severity.WARNING,
    message="This is an example check that passed with a warning.",
    principle=1
)

ex_check
```

And with that, we can propagate checks to any number of manifest fields.
In future, this may prove useful in enumerating checks across projects,
or in generating a manifest scorecard that can be used to track
improvements in project quality over time.

Now, let’s define a few checks:

``` python
def check_project_exists(directory: Path) -> CheckResult:
    """
    Verify that the project directory exists.

    Parameters
    ----------
    directory : Path
        Directory declared as the project working directory.

    Returns
    -------
    CheckResult
        Passes when ``directory`` exists and is a directory.
    """
    if directory.exists() and directory.is_dir():
        return CheckResult(
            name="project_exists",
            state=Severity.PASS,
            message=f"Project directory exists: {directory}",
            principle=1
        )
    return CheckResult(
        name="project_exists",
        state=Severity.ERROR,
        message=f"Project directory does not exist: {directory}",
        principle=1
    )

def check_git_repo_exists(directory: Path) -> CheckResult:
    """
    Verify that the project directory is under Git version control.

    Parameters
    ----------
    directory : Path
        Project directory to inspect.

    Returns
    -------
    CheckResult
        Passes when a ``.git`` directory is present.
    """
    if (directory / ".git").exists():
        return CheckResult(
            name="git_repo",
            state=Severity.PASS,
            message="Project is under git version control.",
            principle=[1, 4]
        )

    return CheckResult(
        name="git_repo",
        state=Severity.ERROR,
        message="Project must be a git repository before data can be staged.",
        principle=[1, 4]
    )

def check_environment_exists(directory: Path) -> CheckResult:
    """
    Look for a project environment specification or lockfile.

    Parameters
    ----------
    directory : Path
        Project directory to inspect.

    Returns
    -------
    CheckResult
        Passes when at least one supported environment file exists and
        warns otherwise.
    """
    candidates = [
        "rv.lock",
        "renv.lock",
        "environment.yml",
        "requirements.txt",
        "pyproject.toml",
        "uv.lock",
        "DESCRIPTION"
    ]
    found = [name for name in candidates if (directory / name).exists()]
    if found:
        return CheckResult(
            name="environment_spec_exists",
            state=Severity.PASS,
            message=f"Environment specification found: {', '.join(found)}",
            principle=1
        )

    return CheckResult(
        name="environment_spec_exists",
        state=Severity.WARNING,
        message=(
            "Project needs a lockfile or environment specification, such as "
            "rv.lock, renv.lock, environment.yml, requirements.txt, "
            "pyproject.toml, uv.lock, or DESCRIPTION."
        ),
        principle=1
    )
```

To assert principle 2, we can simply check that there is at least one
code file in the project directory. This is a very basic check and will
only raise a warning, but it serves to illustrate the point, just in
case a user hasn’t yet started writing code:

``` python
def check_code_exists(directory: Path) -> CheckResult:
    """
    Check whether the project contains analysis or source code files.

    Parameters
    ----------
    directory : Path
        Project directory to inspect recursively.

    Returns
    -------
    CheckResult
        Passes when at least one supported code or notebook file is found
        outside ignored directories.
    """
    patterns = [
        "**/*.py",
        "**/*.R",
        "**/*.Rmd",
        "**/*.qmd",
        "**/*.ipynb",
        "**/*.sh",
        "**/Snakefile",
        "**/_targets.R",
    ]

    ignored_parts = {
        ".git",
        ".venv",
        "renv",
        "rv",
        "__pycache__",
        ".quarto",
        "_site",
        "_targets",
        "node_modules",
    }
    
    files: list[Path] = []
    
    for pattern in patterns:
        files.extend(
            path
            for path in directory.glob(pattern)
            if not any(part in ignored_parts for part in path.parts)
        )
    if files:
        return CheckResult(
            name="analysis_code_exists",
            state=Severity.PASS,
            message=f"Found {len(files)} script or notebook file(s).",
            principle=2,
        )

    return CheckResult(
        name="analysis_code_exists",
        state=Severity.WARNING,
        message=(
            "WARNING: Project must include code scripts or notebooks..."
        ),
        principle=2
    )
```

Principle 3 is similar to principle 2, but we check for the presence of
things like READMEs, notebooks, or markdown files:

``` python
def check_narrative_exists(directory: Path) -> CheckResult:
    """
    Check whether the project contains narrated analysis notebooks.

    Parameters
    ----------
    directory : Path
        Project directory to inspect recursively.

    Returns
    -------
    CheckResult
        Passes when Quarto, R Markdown, or Jupyter notebooks are present
        outside ignored directories.
    """

    patterns = [
        "**/*.qmd",
        "**/*.Rmd",
        "**/*.ipynb",
    ]

    ignored_parts = {
        ".git",
        ".venv",
        "renv",
        "rv",
        "__pycache__",
        ".quarto",
        "_site",
        "_targets",
        "node_modules",
    }
    files: list[Path] = []
    for pattern in patterns:
        files.extend(
            path
            for path in directory.glob(pattern)
            if not any(part in ignored_parts for part in path.parts)
        )

    if files:
        return CheckResult(
            name="notebook_exists",
            state=Severity.PASS,
            message=f"Found {len(files)} narrated analysis file(s).",
            principle=3
        )

    return CheckResult(
        name="notebook_exists",
        state=Severity.WARNING,
        message=(
            "WARNING: Project should include narrated analysis using Quarto, "
            "Marimo, R Markdown, or Jupyter notebooks."
        ),
        principle=3
    )

def check_readme_exists(directory: Path) -> CheckResult:
    """
    Check whether the project contains a top-level README document.

    Parameters
    ----------
    directory : Path
        Project directory to inspect.

    Returns
    -------
    CheckResult
        Passes when a supported README filename exists and errors
        otherwise.
    """
    if (directory / "README.md").exists() or (directory / "README.Rmd").exists() or (directory / "README.qmd").exists() or (directory / "README").exists():
        return CheckResult(
            name="readme_exists",
            state=Severity.PASS,
            message="README file found.",
            principle=3,
        )

    return CheckResult(
        name="readme_exists",
        state=Severity.ERROR,
        message=(
            "Project should include a README.md file that describes the project, "
            "data sources, and analysis plan."
        ),
        principle=3
    )
```

Finally, for principle 4, we’ll gently nudge the user to scaffold their
project using an R project file, or a Python project structure with a
`src` directory.

``` python
def check_project_structure(directory: Path) -> CheckResult:
    has_rproj = (directory / "*.Rproj").exists()
    has_src = (directory / "src").exists() and (directory / "src").is_dir()

    if has_rproj or has_src:
        return CheckResult(
            name="project_structure",
            state=Severity.PASS,
            message="Project structure looks good.",
            principle=4,
        )

    return CheckResult(
        name="project_structure",
        state=Severity.WARNING,
        message=(
            "Project should be scaffolded using an R project file, or a Python "
            "project structure with a `src` directory."
        ),
        principle=4
    )
```

In the next module, we define the ManifestChecker class, which will run
all of these checks and more, and return a structured report that can be
displayed in `stagecoach inspect`.


# Command Line Interface


To run stagecoach, we define a simple CLI using the `typer` library.
This allows users to run stagecoach from the command line, and provides
a nice interface for specifying options.

Once the CLI is defined as “hailing” the stagecoach, stagecoach will run
`issue_manifest` to create the manifest, and then proceed with the rest
of the stagecoach.

Here’s the code for the CLI. First, hailing the stagecoach:

``` python
from pathlib import Path
from enum import Enum
from rich.console import Console
from typing import Annotated
from sheriff.sheriff import Sheriff
from stagecoach.stagecoach import StageCoach
from stagecoach.checks import Severity
from stagecoach.ui import failure_panel
import typer

app = typer.Typer(
    name="stagecoach",
    help="Stage authorized Frontier data into reproducible working environments.",
)

class FailureLevel(str, Enum):
    """
    Severity thresholds exposed by the CLI.

    Attributes
    ----------
    WARNING : str
        Treat warnings as command failures.
    ERROR : str
        Treat only errors as command failures.
    """
    WARNING = "warning"
    ERROR = "error"

@app.command()
def hail(
    interactive: bool = typer.Option(
        True,
        "--interactive/--no-interactive",
        help="Prompt for manifest fields interactively.",
    ),
    output_path: Path = typer.Option(
        Path("stagecoach_manifest.yml"),
        "--output",
        "-o",
        help="Where to write the manifest.",
    ),
    outpost: bool = typer.Option(
        False,
        "--outpost/--no-outpost",
        help="Whether to create a temporary mock Frontier citizenship scaffold for manifest creation outside the assigned Frontier.",
    ),
    overwrite: bool = typer.Option(
        False,
        "--overwrite/--no-overwrite",
        help="Whether to overwrite an existing manifest.",
    ),
) -> None:
    """
    Create a Stagecoach manifest.

    Parameters
    ----------
    interactive : bool, default=True
        Whether to prompt for manifest fields interactively.
    output_path : Path, default=Path("stagecoach_manifest.yml")
        Destination path for the generated manifest.
    overwrite : bool, default=False
        Whether to overwrite an existing manifest file.
    """

    console = Console()
    customs_sheriff = Sheriff(console)

    try:
        StageCoach(
            sheriff=customs_sheriff,
            manifest_path=output_path,
            console=console,
        ).hail(
            interactive=interactive,
            overwrite=overwrite,
            outpost=outpost
        )

    except Exception as exc:
        failure_panel(console, str(exc))
        raise typer.Exit(code=1)
```

Inspecting the manifest:

``` python
@app.command()
def inspect(
    manifest_path: Path = typer.Option(
        Path("stagecoach_manifest.yml"),
        "--manifest",
        "-m",
        help="Path to the manifest to inspect.",
    ),
    level: Annotated[
        FailureLevel,
        typer.Option(
            "--level",
            "-l",
            help="Minimum severity that causes the manifest check to fail.",
            case_sensitive=False,
        ),
    ] = FailureLevel.ERROR,

) -> None:
    """
    Inspect a Stagecoach manifest.

    Parameters
    ----------
    manifest_path : Path, default=Path("stagecoach_manifest.yml")
        Path to the manifest to validate.
    level : FailureLevel, default=FailureLevel.ERROR
        Minimum severity that should cause the command to exit with failure.
    """

    console = Console()
    customs_sheriff = Sheriff(console)

    try:
        passed = StageCoach(
            sheriff=customs_sheriff,
            manifest_path=manifest_path,
            console=console,
        ).inspect(
            level=Severity(level.value),
        )

    except Exception as exc:
        failure_panel(console, str(exc))
        raise typer.Exit(code=1)
    if not passed:
        raise typer.Exit(code=1)
```

And, staging:

``` python
@app.command()
def stage(
    manifest_path: Path = typer.Option(
        Path("stagecoach_manifest.yml"),
        "--manifest",
        "-m",
        help="Path to the manifest to inspect.",
        )
    ):
    """
    Stage data declared by the manifest.

    Returns
    -------
    None
        The command exits with a nonzero status when staging fails.
    """
    console = Console()
    customs_sheriff = Sheriff(console)
    staged = StageCoach(
        sheriff=customs_sheriff,
        console=console,
        manifest_path=manifest_path
    ).stage()
    if not staged:
        raise typer.Exit(code=1)
```

``` python
if __name__ == "__main__":
    app()
```


# Customs


We want to have a secure way of knowing whether a user has the correct
permissions to have data delivered to them via globus. This kind of
validation is going to be called “customs”. As a reminder, the
`Stagecoach` class is a handler for delivering data to townspeople, and
works by issuing users a manifest. This manifest is a JSON file that
users fill in with their credentials and desired data from the Frontier
Gold Mine. Once the user returns the manifest, and the `Stagecoach`
recognizes that the user wants access to data via `globus`, `Dataverse`,
or from the Gold Mine, the `Stagecoach` will call the `customs` handler
and pass this information along. The customs handler then validates the
credentials to ensure that the user has the correct permissions to
access the data. This validation process uses the `Globus SDK`,
`Dataverse API`, or a few simple local file permission checks to verify
the credentials against the data provider service. If the credentials
are valid, the customs handler returns a success message, allowing the
`Stagecoach` to proceed with the data delivery. If the credentials are
invalid, the handler returns an error message, and the `Stagecoach` can
inform the user of the issue and prompt them to correct their
credentials in the manifest. This process ensures that only authorized
users can access the data, maintaining the security of The Frontier’s
Gold Mine resources.

## Globus SDK Authentication and Validation

Globus SDK in Python works as below. Once you’ve set up an App in the
[Developer
panel](https://globus-sdk-python.readthedocs.io/en/stable/user_guide/getting_started/register_app.html)
of the Globus website, you can use the (non-secure) client UUID to
create a user app:

The first method shows us that login is required to use this app:

Next, we can use this app to define a client group:

What’s great is that this method forces the current user to log in to
their globus account:

![Globus Login](images/groups_client-get-my_g.png)

And from there, I can see the groups I’m allowed to access defined in
the Globus dashboard:

The explicit login can be called deliberately:

We can see that the UserApp object is the main functionality we’re
looking for as it deals solely with the authentication and authorization
of users:

> GlobusApp provides a number of useful abstractions in the SDK. It
> handles login flows and storage of tokens, coupled with later
> retrieval of those tokens for use. It can keep track of which clients
> have been created and registered with an app, and therefore make
> intelligent decisions about how and when to prompt users to login.

So, to make sure someone is logged in, we can simply create a function
that triggers this login flow. Fortunately, Globus has provided an
example we can work with. Let’s take an example of trying to get this
file:

`/n/holylabs/cgolden_lab/Lab/projects/sandbox/ReaProject/MadagascarWeatherStations/_pkgdown.yml`

Let’s list my available data:

The `.get_endpoint()` method is used to get the endpoint information for
a given collection. An important piece of information we’ll need is the
data_access scope of the endpoint, which is a little bit of a
complicated discussion; the tldr I believe is that the transfer client
can be modified to have the correct access scope but only if you check
for it first; hence, the `.get_endpoint()` step. There are two cases
under which the endpoint will NOT have data_access:

1.  If the endpoint `entity_type` IS NOT `GCSv5_mapped_collection`, and
2.  If the endpoint IS `high_assurance`.

If either of the above is true, then the endpoint will need special
access and additional steps. If neither of the above is true, then the
endpoint should be accessible with the default client

So neither of these are true, which means that the endpoint should be
accessible with the default client. If it weren’t the case, we would
have to modify the client with
`tc.add_app_data_access_scope(SRC_COLLECTION)`.

To be sure that this transfer is valid, we can first do a dir_stat on
the collection. If we get a successful response, we can assume that the
transfer will work in future:

Finally, we can wrap the transfer submission in a try-except block to
catch any errors that may arise from invalid credentials or permissions:

That works! So now we can create a validation function that allows the
sheriff to validate the globus section of a stagecoach manifest.

The manifest will have the following shape:

``` python
# import yaml
# from pathlib import Path
# from pprint import pprint
# from pyprojroot import here

# manifest = yaml.safe_load(Path(here() / "stagecoach_manifest.yml").read_text())
# pprint(manifest['sources'])
```

In the customs manager, we will have to do two things: authenticate with
the Globus client, and send the actual data. After wrestling with this
for a few tries, and bouncing design ideas off of chatgpt and jupyter
notebooks, I’ve settled on adding a context helper to do the UserApp
creation:

``` python
from globus_sdk import GlobusAppConfig, TransferClient, UserApp, TransferData
from contextlib import contextmanager

@contextmanager
def globus_transfer_client():
    """
    Yield an authenticated Globus transfer client.

    Yields
    ------
    TransferClient
        Transfer client configured for the Frontier Stagecoach app.
    """
    with UserApp(
        "Frontier-Stagecoach",
        client_id="7723dff4-fa63-4639-903b-ba6541e24e98",
        config=GlobusAppConfig(auto_redrive_gares=True),
    ) as app:
        with TransferClient(app=app) as client:
            yield client
```

This will allow us to create transfer clients with the `with` statement,
allowing us to iterate over the items in the Globus request
individually, granting each of them a unique request and clearance:

``` python
from dataclasses import dataclass, field
from typing import Any

@dataclass
class Clearance:
    """
    Represent the outcome of a source access check.

    Attributes
    ----------
    source : str
        Source identifier being validated.
    cleared : bool
        Whether access validation succeeded.
    message : str, default=""
        Human-readable summary of the validation result.
    details : dict[str, Any]
        Source-specific metadata captured during validation.
    """
    
    source: str
    cleared: bool
    message: str = ""
    details: dict[str, Any] = field(default_factory=dict)

def check_globus_clearance(
    globus_info: dict
    ) -> Clearance:
    """
    Validate access to Globus endpoints and requested source paths.

    Parameters
    ----------
    globus_info : dict
        Manifest subsection describing Globus endpoints and staged items.

    Returns
    -------
    Clearance
        Clearance result describing whether Globus access checks passed.
    """
    try:
        with globus_transfer_client() as client:
            client.get_endpoint(globus_info["source_endpoint"])
            client.get_endpoint(globus_info["destination_endpoint"])
            for item in globus_info["items"]:
                client.operation_stat(
                    globus_info["source_endpoint"],
                    path=item["source_path"],
                )

        return Clearance(
            source="02_globus",
            cleared=True, 
            message="Globus clearance passed.",
            details={
                    "source_endpoint": globus_info["source_endpoint"],
                    "destination_endpoint": globus_info["destination_endpoint"],
                    "items_checked": len(globus_info["items"])
                }
            )

    except Exception as exc:
        return Clearance(
            source="02_globus",
            cleared=False,
            message=str(exc),
            details={
                "source_endpoint": globus_info.get("source_endpoint"),
                "destination_endpoint": globus_info.get("destination_endpoint"),
            }
        )
```

If the checks pass, we can move on to building the transfer logic. This
is built mostly on what we learned above:

``` python
def build_globus_transfer(
    manifest: dict,
    clearance: Clearance,
    fix_holylabs: bool = True,
    label: str = "Stagecoach transfer",
    ) -> TransferData:
    """
    Build a Globus transfer request from manifest settings.

    Parameters
    ----------
    globus_info : dict
        Manifest subsection describing Globus endpoints and staged items.
    clearance : Clearance
        Successful clearance result for the same Globus configuration.
    fix_holylabs : bool, default=True
        Whether to apply Holylabs-specific path fix to the transfer (removes redundant LAB segment from paths).
    label: str, optional
        Optional label for the transfer task.

    Returns
    -------
    TransferData
        Transfer request populated with all requested items.

    Raises
    ------
    ValueError
        Raised when ``clearance`` indicates that access checks failed.
    """
    globus_info = manifest.get("sources", {}).get("02_globus", {})

    if not globus_info:
        raise ValueError("Globus information is missing from the manifest.")

    if not clearance.cleared:
        raise ValueError(f"Cannot build transfer: clearance failed with message: {clearance.message}")
    
    transfer = TransferData(
        source_endpoint=globus_info["source_endpoint"],
        destination_endpoint=globus_info["destination_endpoint"],
        label=label,
    )

    for item in globus_info["items"]:
        
        source_path = item["source_path"].replace("/n/holylabs/LABS/", "/n/holylabs/") if fix_holylabs else item["source_path"]
        
        if item.get("destination_path", None):
            destination_root = item["destination_path"]
        else:
            destination_root = manifest["project"]["input_data_dir"]
        destination_root = destination_root.replace("/n/holylabs/LABS/", "/n/holylabs/") if fix_holylabs else manifest["project"]["input_data_dir"]
        destination_path = Path(destination_root) / "02_globus" / item['name'] / Path(source_path).name
        
        transfer.add_item(
            str(source_path),
            str(destination_path),
            recursive=item.get("recursive", True),
        )

    return transfer
```

Now with that, the customs manager can first check for clearance, and
additionally build the transfer behind the scenes.

## Gold Mine

Accessing the Gold Mine can only be done from FASRC, for now. It will
implement a simple check for whether the user has access to the
specified path. This will reuse the `Clearance` class from earlier:

``` python
import glob
import os
from pathlib import Path

def check_gold_mine_clearance(
    gold_mine_info: dict
    ) -> Clearance:
    """
    Validate readability of requested Gold Mine paths on FASRC.

    Parameters
    ----------
    gold_mine_info : dict
        Manifest subsection describing Gold Mine staging items.

    Returns
    -------
    Clearance
        Clearance result describing whether all required paths were found
        and were readable.
    """

    try:
        items = gold_mine_info.get("items", [])
        checked = []

        for item in items:
            name = item.get("name", "unnamed")
            path_regex = item.get("path_regex")

            if not path_regex:
                return Clearance(
                    source="01_gold_mine",
                    cleared=False,
                    message=f"{name}: missing path_regex",
                )

            matches = [Path(p) for p in glob.glob(path_regex)]

            if not matches and item.get("required", True):
                return Clearance(
                    source="01_gold_mine",
                    cleared=False,
                    message=f"{name}: no matches for {path_regex}",
                )

            for match in matches:
                if not os.access(match, os.R_OK):
                    return Clearance(
                        source="01_gold_mine",
                        cleared=False,
                        message=f"{name}: path is not readable: {match}",
                    )

            checked.append(
                {
                    "name": name,
                    "path_regex": path_regex,
                    "matches": [str(p) for p in matches],
                }
            )

        return Clearance(
            source="01_gold_mine",
            cleared=True,
            message="Gold Mine clearance passed.",
            details={"items_checked": checked},
        )

    except Exception as exc:
        return Clearance(
            source="01_gold_mine",
            cleared=False,
            message=str(exc),
        )
```

## Dataverse

🚧 Not yet implemented 🚧

## Conclusion

When that sends back a clearance object with `cleared=True`, the
`Stagecoach` can proceed with the transfer using the provided
`transfer_client`. If `cleared=False`, the `Stagecoach` can inform the
user of the issue and prompt them to correct their credentials in the
manifest.

# Script file

The code for this document can be found here:

- [../src/stagecoach/customs.py](../src/stagecoach/customs.py)


# Manifest File


The manifest file is a YAML file that describes the terms of use for a
dataset. It includes some simple automated checks to ensure that your
project is compliant with the terms of use. The terms of use are as
follows:

- You must be a person in the frontier

- You must be working in the town of the frontier

- Your project must have a git repository

- Your project must have a name and a description

This is easily translatable to a couple of simple checks.

> [!TIP]
>
> ### Check Summary tl;dr
>
> | Check | What it means | How to fix it if it fails |
> |----|----|----|
> | project exists | You have created your project underneath the Golden Lab lab space path | Move your project to `/n/holylabs/cgolden_lab/Lab/frontier/works/town/<YOUR NAME>/<YOUR PROJECT>` |
> | project is `git` tracked | Your project has a `git` repository to track changes to code files, and uses `.gitignore` to ignore sensitive data or sandbox directories | Run `git init` in your project directory |
> | project has a readme | Your project has a README file that clearly describes what the project is about | Create a README file |
> | project has a reproducible environment mechanism | Your project has a `requirements.txt`, `environment.yml`, `pyproject.toml`, or similar kind of file to define the software environment you are using | Create a `requirements.txt`, `pyproject.toml` or `environment.yml` file with `uv`, `conda`, `rv`, `renv`, or similar |
> | project has code | Your project has a `src` or `R` directory with code in it, and any scientific conclusions you have can be reproduced from the code | Create a `src` directory and add some “hello world” code to it |
> | project has a narrative | Your project uses literate programming to communicate the science clearly and effectively | Create at least one Quarto, Jupyter, or Rmarkdown notebook in your project to describe your work, what you are trying to do and why it is important |
> | project has a reasonable structure | Your project is reasonably organized and not a mish-mash of scripts | Formally initialize your project with an [R project file](https://r4ds.hadley.nz/workflow-scripts.html) (`.Rproj`) or Python [`src`](https://playbooks.omsf.io/developer/repository-structure/use-a-src-or-flat-layout/) format |
>
> The checks are designed to be simple and easy to implement, and each
> one’s code is documented below. If you believe a particular check
> needs to be implemented or improved, please submit a [Pull
> Request](https://github.com/GoldenPlanetaryHealthLab/stagecoach/pulls)!

First, we import the `Sheriff` class from the `sheriff` module. Then, we
create an instance of the `Sheriff` class and call the `check_citizen`
method to perform the checks.

If the sheriff verifies citizenship, the checks will pass.

``` python
from rich.console import Console
from sheriff.sheriff import Sheriff

def check_citizenship(customs_sheriff: Sheriff, console: Console) -> None:
    """
    Validate Frontier citizenship through the sheriff.

    Parameters
    ----------
    customs_sheriff : Sheriff
        Sheriff instance used to verify citizenship.
    console : Console
        Rich console used for progress messages.

    Raises
    ------
    RuntimeError
        Raised when the sheriff does not validate the current user.
    """

    info(console, "Checking citizenship...")

    if not customs_sheriff.check_citizen():
        raise RuntimeError(
            "You are not a citizen of the Frontier. "
            "Please check your citizenship and try again."
        )

    success(console, "Citizenship verified")
```

We can see this function in action below:

``` python
# mysheriff = Sheriff()

# check_citizenship(mysheriff)
```

Now that we know that the user is a citizen, we can check that they are
working in a designated [`town`](#TODO%20create%20town%20documentation).

Next, we can implement the manifest. It’s going to be simple for now,
but what it really needs to know is where IT is, and where the data is,
and figure out how to get the data to the user’s specified location.

``` python
# we don't export this portion of script
# because it will cause side effects to anyone who
# loads it at runtime

# from pyprojroot import here

# manifest_template = {
#     "frontier": {
#         "name": "golden-lab",
#         "schema_version": 1.0,
#         "stagecoach_version": 0.1,

#         # critical: we must know where the frontier starts
#         "root": "/n/holylabs/cgolden_lab/Lab/frontier"
#     },

#     "paths": {
#         "goldmine": "goldmine",
#         "works": "works",
#         "town": "town",
#         "governance": "governance",
#     },

#     "citizen": {
#         "name": None,
#         "email": None,
#     },

#     "project": {
#         "project_name": None,
#         "project_working_dir": str(here()),
#         "input_data_dir": str(here() / "data" / "inputs"),
#         "intermediate_data_dir": str(here() / "data" / "intermediates"),
#         "output_data_dir": str(here() / "data" / "outputs"),
#         "sandbox_dir": str(here() / "sandbox"),
#     },

#     "sources": {
#         "01_gold_mine": {
#             "enabled": False,

#             # Each item stages into:
#             # <input_data_dir>/01_gold_mine/<name>/<basename>
#             "items": [
#                 {
#                     "name": "example_goldmine_item",
#                     "path_regex": None,
#                     "required": True,
#                 }
#             ],
#         },

#         "02_globus": {
#             "enabled": False,

#             # Globus collection UUIDs
#             "source_endpoint": None,
#             "destination_endpoint": None,

#             # Each item stages into:
#             # <input_data_dir>/02_globus/<name>/<basename>
#             "items": [
#                 {
#                     "name": "example_globus_item",
#                     "source_path": None,
#                     "destination_path": None,  # optional override
#                     "files_regex": [],
#                     "recursive": True,
#                     "required": True,
#                 }
#             ],
#         },

#         "03_dataverse": {
#             "enabled": False,

#             "server_url": None,
#             "dataset_pid": None,
#             "dataset_version": "latest",
#             "api_token_file": None,

#             # Each item stages into:
#             # <input_data_dir>/03_dataverse/<name>/<basename>
#             "items": [
#                 {
#                     "name": "example_dataverse_item",
#                     "files_regex": [],
#                     "required": True,
#                 }
#             ],
#         },
#     },
# }

# import yaml
# from pathlib import Path

# output_path = Path("src/stagecoach/templates/manifest.yml")
# output_path.parent.mkdir(parents=True, exist_ok=True)

# with output_path.open("w") as f:
#     yaml.dump(manifest_template, f, sort_keys=False)
```

If a citizen can pass the citizenship check, then the stagecoach can
issue a manifest which the user can fill in with the necessary
information to get access to the data. The manifest is a yaml file in
the `src/stagecoach` directory, so to expose it to the user, we can
create a simple function that writes the manifest to the file system.

``` python
import yaml
from typing import Any
from importlib.resources import files

def load_template() -> dict[str, Any]:
    """
    Load the packaged Stagecoach manifest template.

    Returns
    -------
    dict[str, Any]
        Parsed manifest template bundled with the package.
    """

    template_path = files("stagecoach.templates").joinpath("manifest.yml")
    return yaml.safe_load(template_path.read_text())
```

This function makes use of the `importlib.resources` module, and so long
as the user’s version of `stagecoach` is up to date, the above YAML will
be the template they receive.

Next, to actually issue the manifest, we can create a function that
prompts the user to fill in the necessary information. This is intended
to be the primary interface at the command line. This is made by two
subfunctions, `fill_manifest_interactively`, which prompts the user to
fill in the manifest fields, and `write_manifest`, which writes the
manifest to disk.

We use `questionary` to make the function interactive, but it can also
be used in a non-interactive way by passing the necessary information as
arguments.

``` python
import questionary
from typing import Any

def fill_manifest_interactively(manifest: dict[str, Any]) -> dict[str, Any]:
    """
    Prompt the user for manifest values at the command line.

    Parameters
    ----------
    manifest : dict[str, Any]
        Manifest template to populate in place.

    Returns
    -------
    dict[str, Any]
        Updated manifest containing the collected user input.
    """

    citizen_name = questionary.text(
        "Citizen name (Your name as it appears in Town):"
    ).ask()

    citizen_email = questionary.text("Citizen email (optional):").ask()

    project_name = questionary.text("Project name:").ask()

    project_working_dir = questionary.text(
        "Project working directory:",
        default=str(here()),
    ).ask()

    input_data_dir = questionary.text(
        "Input data directory:",
        default=str(Path(project_working_dir) / "data" / "inputs"),
    ).ask()

    output_data_dir = questionary.text(
        "Output data directory:",
        default=str(Path(project_working_dir) / "data" / "outputs"),
    ).ask()

    # use_globus = questionary.confirm(
    #     "Use Globus for transport?",
    #     default=False,
    # ).ask()

    # manifest.setdefault("citizen", {})
    # manifest.setdefault("project", {})
    # manifest.setdefault("source", {})
    # manifest["source"].setdefault("globus", {})
    manifest["citizen"]["name"] = citizen_name
    manifest["citizen"]["email"] = citizen_email
    manifest["project"]["project_name"] = project_name
    manifest["project"]["project_working_dir"] = project_working_dir
    manifest["project"]["input_data_dir"] = input_data_dir
    manifest["project"]["output_data_dir"] = output_data_dir

    # if use_globus:
    #     manifest["source"]["02_globus"]["globus_username"] = questionary.text(
    #         "Globus username:"
    #     ).ask()
    #     manifest["source"]["02_globus"]["globus_endpoint"] = questionary.text(
    #         "Globus endpoint:"
    #     ).ask()
    
    # if use_dataverse:
    #     manifest["source"]["03_dataverse"]["dataverse_server_url"] = questionary.text(
    #         "Dataverse server URL:"
    #     ).ask()
    #     manifest["source"]["03_dataverse"]["dataverse_dataset_pid"] = questionary.text(
    #         "Dataverse dataset PID:"
    #     ).ask()

    return manifest
```

In practice, this process looks like:

``` python
# mymanifest = load_template()

# filled_manifest = fill_manifest_interactively(mymanifest)
```

Next, writing the manifest should be straight forward:

``` python
from pathlib import Path
from rich.console import Console

def write_manifest(
    manifest: dict[str, Any],
    output_path: str | Path,
    console: Console,
    overwrite: bool = False
    ) -> None:
    """
    Write a manifest to disk as YAML.

    Parameters
    ----------
    manifest : dict[str, Any]
        Manifest data to serialize.
    output_path : str | Path
        Destination path for the YAML file.
    console : Console
        Rich console used for progress messages.
    overwrite : bool, default=False
        Whether to bypass the overwrite confirmation prompt.

    Raises
    ------
    RuntimeError
        Raised when the target file exists and overwrite is declined.
    """

    output_path = Path(output_path)

    if output_path.exists() and not overwrite:
        overwrite_confirmed = questionary.confirm(
            f"{output_path} already exists. Overwrite?",
            default=False,
        ).ask()

        if not overwrite_confirmed:
            raise RuntimeError(f"Manifest not written: {output_path} already exists.")

    output_path.parent.mkdir(parents=True, exist_ok=True)
    info(console, f"Writing manifest to: {output_path}")
    with console.status("[bold green]Saving manifest..."):
        with output_path.open("w") as f:
            yaml.dump(manifest, f, sort_keys=False)

    success(console, "Manifest created successfully")
```

To test:

``` python
# from tempfile import TemporaryDirectory

# with TemporaryDirectory() as tmpdir:
#     temp_manifest_path = Path(tmpdir) / "my_manifest.yml"
#     write_manifest(mymanifest, temp_manifest_path, overwrite=True)

#     # show contents of the file
#     with temp_manifest_path.open() as f:
#         print(f.read())
```

Looks great! We now have the pieces to issue a manifest:

``` python
from rich.console import Console
from stagecoach.ui import info, success
from sheriff.sheriff import Sheriff
from pyprojroot import here

def issue_manifest(
    customs_sheriff: Sheriff,
    console: Console,
    interactive: bool = True,
    output_path: str | Path = here() / "stagecoach_manifest.yml",
    overwrite: bool = False
) -> None:

    """
    Generate a Stagecoach manifest for a project.

    Parameters
    ----------
    customs_sheriff : Sheriff
        Sheriff instance used to validate Frontier citizenship.
    console : Console
        Rich console used for progress messages.
    interactive : bool, default=True
        Whether to prompt the user for manifest fields.
    output_path : str | Path, default=here() / "stagecoach_manifest.yml"
        Destination path for the generated manifest.
    overwrite : bool, default=False
        Whether to overwrite an existing manifest at the output path.

    Returns
    -------
    None
        This function is called for its side effect of writing the manifest.
    """

    check_citizenship(customs_sheriff, console)
    info(console, "Loading manifest template...")
    manifest = load_template()
    success(console, "Template loaded\n")

    if interactive:
        manifest = fill_manifest_interactively(manifest)

    write_manifest(manifest, output_path, overwrite=overwrite, console=console)

    success(console, "Next step: stagecoach inspect")
```

Issuing a manifest writes the manifest to the file system, where the
user can modify the necessary bits and pieces to ensure they have
correct access to the Gold Mine.

Next, check out the CLI module to see how we “hail,” a stagecoach.


# Manifest Checker Class


Here’s a manifest checker class that performs various checks on a
project directory based on a manifest file. Each check returns a
structured result that can be displayed nicely in `stagecoach inspect`.
The class also includes a method to determine if all error-level checks
pass.

``` python
import yaml
from pathlib import Path
from stagecoach.checks import *

class ManifestChecker:
    """
    Run the standard Stagecoach manifest checks for a project.

    Parameters
    ----------
    manifest_file : str | Path
        Path to the manifest file that defines the project directory.
    severity_level : Severity
        Minimum severity that should cause ``passes`` to return ``False``.

    Attributes
    ----------
    manifest : dict
        Parsed manifest contents.
    project_dir : str
        Project working directory declared by the manifest.
    severity_level : Severity
        Failure threshold used by ``passes``.
    """
    def __init__(self, manifest_file: str | Path, severity_level: Severity):
        self.manifest =  yaml.safe_load(Path(manifest_file).read_text())
        self.project_dir = self.manifest['project']['project_working_dir']
        self.severity_level = severity_level

    def run_all(self) -> list[CheckResult]:
        """
        Run all configured project checks.

        Returns
        -------
        list[CheckResult]
            Results from each Stagecoach project validation check.
        """

        return [
            check_project_exists(Path(self.project_dir)),
            check_git_repo_exists(Path(self.project_dir)),
            check_readme_exists(Path(self.project_dir)),
            check_environment_exists(Path(self.project_dir)),
            check_code_exists(Path(self.project_dir)),
            check_narrative_exists(Path(self.project_dir)),
            check_project_structure(Path(self.project_dir))
        ]

    def passes(self) -> bool:
        """
        Determine whether the manifest passes the configured threshold.

        Returns
        -------
        bool
            ``True`` when every check result falls below
            ``self.severity_level`` and ``False`` otherwise.
        """

        results = self.run_all()
        threshold = SEVERITY_RANK[self.severity_level]

        return all(
            SEVERITY_RANK[check_result.state] < threshold
            for check_result in results
        )
```

So, to ensure that a manifest passes, we just issue:


# Outpost


We have a problem. The `stagecoach` tool is designed to run on the FASRC
cluster, and it relies on the `sheriff` to run a citizenship test
(i.e. determine that you are bought in to use The Frontier). This means
that if you want to run `stagecoach` outside of FASRC, you need to have
those documents, which will be a barrier for many users.

Here, we introduce the `outpost` module, which will temporarily scaffold
a fake Frontier directory that the sheriff will accept as valid. This
will allow users to run `stagecoach` on their local machines, without
needing to have the actual citizenship documents.

We’re going ot implement this as a CLI flag, `--outpost`, which will
trigger the creation of this fake directory. Additionally, ChatGPT
recommends using the `contextlib` module to create a context manager
that will handle the setup and teardown of this fake directory, ensuring
that it is cleaned up after use.

The `outpost` module will have a `create_outpost` function context.

``` python
from __future__ import annotations

from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
import os
import yaml


@contextmanager
def create_outpost(
    *,
    mode: str = "outpost",
    frontier_dirname: str = "frontier",
    frontier_file: str = "frontier.yml",
):
    """
    Create a temporary mock Frontier file and point THE_FRONTIER at it.

    This does not grant access to real Frontier resources. It only satisfies
    Sheriff's file-based citizenship check for commands that need to run
    outside the assigned Frontier, such as `stagecoach hail --outpost`.
    """
    old_the_frontier = os.environ.get("THE_FRONTIER")

    with TemporaryDirectory(prefix="stagecoach-outpost-") as tmp:
        root = Path(tmp)
        frontier_root = root / frontier_dirname
        frontier_root.mkdir(parents=True, exist_ok=True)

        frontier_path = frontier_root / frontier_file

        mock_frontier = {
            "frontier": {
                "mode": mode,
                "kind": "mock",
                "created_by": "stagecoach.forger",
                "purpose": "temporary citizenship scaffold",
            }
        }

        frontier_path.write_text(
            yaml.safe_dump(mock_frontier, sort_keys=False),
            encoding="utf-8",
        )

        os.environ["THE_FRONTIER"] = str(frontier_path)

        try:
            yield frontier_path
        finally:
            if old_the_frontier is None:
                os.environ.pop("THE_FRONTIER", None)
            else:
                os.environ["THE_FRONTIER"] = old_the_frontier
```


# Stage Data


Finally, it’s time to deliver the user’s data to them. In this module,
we define several staging methods that the stagecoach can use.

There are a couple of things that will need to happen to make this work.
One of them will be creating the directories for data usage:

``` python
import os
import glob
from pathlib import Path
from stagecoach.ui import info, error, success, warning
from rich.console import Console
```

``` python
def create_stage_directories(manifest: dict, console: Console, gitignore=True):
    """
    Create the standard Stagecoach data directories.

    Parameters
    ----------
    manifest : dict
        Manifest containing the ``project`` directory configuration.
    console : Console
        Rich console used for progress messages.
    gitignore : bool, default=True
        Whether to create a minimal ``.gitignore`` file in each directory.

    Raises
    ------
    ValueError
        Raised when one or more required project directories are missing
        from the manifest.
    """
    
    input_dir = manifest.get("project", {}).get("input_data_dir")
    intermediate_dir = manifest.get("project", {}).get("intermediate_data_dir")
    output_dir = manifest.get("project", {}).get("output_data_dir")
    sandbox_dir = manifest.get("project", {}).get("sandbox_dir")

    paths = [input_dir, intermediate_dir, output_dir, sandbox_dir]

    for p in paths:
        if p is None:
            error(console, f"Stage directory {p} is not defined in the manifest.")
            raise ValueError("One or more stage directories are not defined in the manifest.")
        if not os.path.exists(p):
            os.makedirs(p)
            info(console, f"Created stage directory at {p}")
        else:
            info(console, f"Stage directory already exists at {p}")
        
        if gitignore:
            gitignore_path = Path(p) / ".gitignore"
            if not gitignore_path.exists():
                with open(gitignore_path, "w") as f:
                    f.write("*\n!.gitignore\n")
                info(console, f"Created .gitignore in {p}")
            else:
                info(console, f".gitignore already exists in {p}")
    
    success(console, "Stage directories are set up.")
```

Let’s create a quick `StageResult` class to represent the results of
staging operations, so we can report them nicely in the UI and
potentially use them in the future for more detailed reporting:

``` python
from dataclasses import dataclass, field
from typing import Any

@dataclass
class StageResult:
    """
    Summarize the result of a staging operation.

    Attributes
    ----------
    source : str
        Source identifier, such as ``01_gold_mine`` or ``02_globus``.
    staged : bool
        Whether the staging operation completed successfully.
    message : str, default=""
        Human-readable summary of the outcome.
    details : dict[str, Any]
        Source-specific metadata about the staging attempt.
    """
    source: str
    staged: bool
    message: str = ""
    details: dict[str, Any] = field(default_factory=dict)
```

Next, we can work on fetching the data from Globus. Fortunately, much of
the work has been done for us in the `customs` module, so we just need
to call the appropriate functions.

``` python
from stagecoach.customs import (

    check_globus_clearance, 
    build_globus_transfer,
    globus_transfer_client,

    check_gold_mine_clearance
) 

def stage_data_from_globus(
    manifest: dict,
    console: Console,
    issue_transfer: bool = True
) -> StageResult:
    """
    Submit a Globus transfer for manifest-declared items.

    Parameters
    ----------
    manifest : dict
        Manifest containing the ``02_globus`` source configuration.
    console : Console
        Rich console used for progress messages.
    issue_transfer : bool, default=True
        Whether to submit the transfer after validating access.

    Returns
    -------
    StageResult
        Summary of the Globus staging attempt.

    Raises
    ------
    ValueError
        Raised when the manifest does not define the required Globus
        source information.
    """

    if not issue_transfer:
        info(console, "Globus clearance check passed. Skipping transfer as per configuration.")
        return StageResult(
            source="02_globus",
            staged=False,
            message="Globus clearance passed. Transfer skipped.",
        )

    globus_info = manifest.get("sources", {}).get("02_globus", {})
    input_root = Path(manifest["project"]["input_data_dir"])
    source_root = input_root / "02_globus"
    source_root.mkdir(parents=True, exist_ok=True)

    if not globus_info:
        error(console, "No Globus information found in the manifest.")
        raise ValueError("Globus information is missing from the manifest.")
    clearance = check_globus_clearance(globus_info)
    transfer = build_globus_transfer(manifest, clearance, fix_holylabs=True, label = manifest.get("project", {}).get("project_name", "stagecoach_transfer"))
    try:
        with globus_transfer_client() as client:
            client.get_endpoint(globus_info["source_endpoint"])
            client.get_endpoint(globus_info["destination_endpoint"])
            task = client.submit_transfer(transfer)
            info(console, f"Globus transfer submitted with task ID: {task['task_id']}")

            return StageResult(
                source="02_globus",
                staged=True,
                message=f"Globus transfer submitted with task ID: {task['task_id']}",
                details=task,
            )
    except Exception as exc:
        error(console, f"Error during Globus staging: {exc}")
        return StageResult(
            source="02_globus",
            staged=False,
            message="Globus staging failed with transfer client error.",
            details=exc,
        )
```

Next we implement symlink staging via FASRC:

``` python
def stage_data_from_fasrc(
    manifest: dict,
    console: Console,
) -> StageResult:
    """
    Stage Gold Mine data by creating symlinks on FASRC.

    Parameters
    ----------
    manifest : dict
        Manifest containing the ``01_gold_mine`` source configuration.
    console : Console
        Rich console used for progress messages.

    Returns
    -------
    StageResult
        Summary of symlink creation outcomes for all requested items.
    """
    fasrc_info = manifest.get("sources", {}).get("01_gold_mine", {})
    items = fasrc_info.get("items", [])

    task_list = {
        "succeeded": [],
        "failed": [],
        "skipped": [],
    }

    clearance = check_gold_mine_clearance(fasrc_info)
    
    if not clearance.cleared:
        return StageResult(
            source="01_gold_mine",
            staged=False,
            message=f"Gold Mine clearance failed: {clearance.message}",
            details=task_list
        )

    if not items:
        return StageResult(
            source="01_gold_mine",
            staged=False,
            message="No FASRC source items found in the manifest.",
            details=task_list,
        )

    try:
        input_root = Path(manifest["project"]["input_data_dir"])
        source_root = input_root / "01_gold_mine"
        source_root.mkdir(parents=True, exist_ok=True)

        for item in items:
            item_name = item.get("name")
            path_regex = item.get("path_regex")

            if not item_name or not path_regex:
                task_list["failed"].append(item)
                continue

            info(console, f"Staging Gold Mine item: {item_name}")

            item_stage_dir = source_root / item_name
            item_stage_dir.mkdir(parents=True, exist_ok=True)

            files_found = glob.glob(path_regex)

            if not files_found:
                warning(console, f"No files found matching pattern: {path_regex}")
                task_list["skipped"].append(path_regex)
                continue

            for source_path_raw in files_found:
                source_path = Path(source_path_raw).resolve()
                destination_path = item_stage_dir / source_path.name

                if destination_path.exists() or destination_path.is_symlink():
                    existing_target = None

                    if destination_path.is_symlink():
                        existing_target = Path(os.readlink(destination_path)).resolve()

                    if existing_target == source_path:
                        task_list["skipped"].append(str(source_path))
                        continue

                    task_list["failed"].append(str(source_path))
                    continue

                try:
                    os.symlink(
                        source_path,
                        destination_path,
                        target_is_directory=source_path.is_dir(),
                    )
                    task_list["succeeded"].append(str(source_path))

                except OSError:
                    task_list["failed"].append(str(source_path))

    except Exception as exc:
        return StageResult(
            source="01_gold_mine",
            staged=False,
            message=f"Gold Mine staging failed: {exc}",
            details=task_list,
        )

    staged = len(task_list["succeeded"]) > 0 or len(task_list["skipped"]) > 0
    failed = len(task_list["failed"]) > 0

    return StageResult(
        source="01_gold_mine",
        staged=staged and not failed,
        message=(
            "Symlink staging summary: "
            f"{len(task_list['succeeded'])} succeeded, "
            f"{len(task_list['skipped'])} skipped, "
            f"{len(task_list['failed'])} failed."
        ),
        details=task_list,
    )
```

That should do it for now! We will implement the dataverse staging later
on.

# Script file

The code for this document can be found here:

- [../src/stagecoach/stage.py](../src/stagecoach/stage.py)


# StageCoach Explained


The stagecoach is a data ferrying system that automagically configures
how to get access to source data for your project. It is primarily
designed to be used in two ways:

1.  To get access to data on FASRC while developing on FASRC

2.  To get access to data on FASRC while developing on your local
    machine

To do this, stagecoach uses a manifest file that provides the user with
options for how to access the data. Once the user has made a selection,
stagecoach will try to set up the necessary access to the data, and when
ready, ferry ONLY what is needed to the user.

If you are on FASRC, this will likely be setting up a symbolic link to
the Gold Mine. If you are on your local machine, stagecoach will work
with you to access Globus and set up an economical, minimal — but secure
— transfer to get the data to you.

To get access to data, stagecoach will first ask you to select a
manifest file. This file is a YAML file that contains information about
the data and how to access it. The manifest file will also contain
information about the data dependencies for your project, and how to
access those dependencies. When you agree to a manifest, stagecoach will
read your manifest file and set up the necessary access to the data.
This may involve setting up a symbolic link to the Gold Mine, or it may
involve setting up a Globus transfer to get the data to you. Once you’ve
been authorized to access, the data will be ferried to your project
directory, and you can start working with it.

Here, we define the `stagecoach` class, while each of the methods are
defined in their own modules. There are three main methods that are
defined in the `stagecoach` class:

- `hail`: This is the first method that is called when you run
  `stagecoach` from the command line. It is responsible for creating the
  manifest file.

- `inspect`: This is the second step of staging, in which the
  `stagecoach` class reads the manifest file and checks that everything
  is in order. This is where we check that the user has access to the
  data, and that the data dependencies are met.

- `stage`: This is the final step of staging, in which the `stagecoach`
  class brings the data into the user’s environment. This is where we
  set up the symbolic link to the `Gold Mine`, or set up the Globus
  transfer to get the data to the user.

``` python
import yaml
from contextlib import nullcontext
from stagecoach.outpost import create_outpost
from stagecoach.issue_manifest import issue_manifest
from stagecoach.manifest_checker import ManifestChecker
from stagecoach.checks import Severity
from stagecoach.customs import check_gold_mine_clearance, check_globus_clearance
from stagecoach.stage import create_stage_directories, stage_data_from_fasrc, stage_data_from_globus, StageResult
from sheriff.sheriff import Sheriff
from rich.console import Console
from pathlib import Path

from stagecoach.ui import (
    banner,
    success,
    warning,
    error,
    info,
    failure_panel,
    handbook_note,
    check_result,
)

class StageCoach:
    """
    Orchestrate manifest issuance, inspection, and data staging.

    Parameters
    ----------
    sheriff : Sheriff | None, default=None
        Sheriff instance used for identity and policy checks.
    manifest_path : str | Path, default="stagecoach_manifest.yml"
        Path to the manifest used by instance methods.
    console : Console | None, default=None
        Rich console used for user-facing output.

    Methods
    -------
    hail
        Create or partially fill a manifest.

    inspect
        Read and validate a manifest without moving data.

    stage
        Symlink or materialize declared data into project directories.
    """
    def __init__(
        self,
        sheriff: Sheriff | None = None,
        manifest_path: str | Path = "stagecoach_manifest.yml",
        console: Console | None = None,
    ) -> None:
        self.sheriff = sheriff or Sheriff()
        self.manifest_path = Path(manifest_path)
        self.console = console or Console()

    def hail(
        self,
        interactive: bool = True,
        overwrite: bool = False,
        sheriff: Sheriff | None = None,
        console: Console | None = None,
        manifest_path: str | Path | None = None,
        outpost: bool = False,
    ) -> bool:
        """
        Create a manifest for a new Stagecoach workflow.

        Parameters
        ----------
        interactive : bool, default=True
            Whether to prompt for manifest fields interactively.
        overwrite : bool, default=False
            Whether to overwrite an existing manifest file.
        sheriff : Sheriff | None, default=None
            Optional sheriff override for this invocation.
        console : Console | None, default=None
            Optional console override for this invocation.
        manifest_path : str | Path | None, default=None
            Optional manifest destination override.
        outpost : bool, default=False
            Whether to create a temporary mock Frontier citizenship scaffold for
            manifest creation outside the assigned Frontier.

        Returns
        -------
        bool
            ``True`` when manifest creation completes successfully.
        """
        sheriff = sheriff or self.sheriff
        console = console or self.console
        manifest_path = Path(manifest_path) if manifest_path else self.manifest_path
        
        banner(console, "🚂 Hailing the Stagecoach...")

        banner(console, "The stagecoach manifest is a YAML file that describes the data sources for your project and how to access them. You can create a manifest interactively, or stagecoach will provide a template manifest and you will fill in the details. The manifest will also include information about the data dependencies for your project, and how to access those dependencies.")

        if outpost:

            banner(
                console,
                "Using Outpost mode for manifest creation. ⛺"
                "Stagecoach will create a temporary mock Frontier file for the "
                "citizenship check, allowing the Stagecoach to "
                "ferry the data to a fictional Frontier."
            )

        outpost = create_outpost() if outpost else nullcontext()

        with outpost:
            issue_manifest(
                customs_sheriff=sheriff,
                interactive=interactive,
                output_path=manifest_path,
                overwrite=overwrite,
                console=console,
            )

        banner(console, f"✅ Manifest created at: {manifest_path}. Fill it out and then run `stagecoach inspect` to check it!")

        return True

    def inspect(
        self,
        console: Console | None = None,
        level: Severity = Severity.ERROR
    ) -> bool:
        """
        Validate a manifest and report any blocking issues.

        Parameters
        ----------
        console : Console | None, default=None
            Optional console override for this invocation.
        level : Severity, default=Severity.ERROR
            Minimum severity that should cause inspection to fail.

        Returns
        -------
        bool
            ``True`` when the manifest passes all checks at the requested
            severity threshold.
        """

        console = console or self.console
        banner(console, f"📋 Inspecting manifest at level: {level.value}")
        
        checker = ManifestChecker(self.manifest_path, level)
        checks = checker.run_all()

        for result in checks:
            check_result(console, result)

        manifest = yaml.safe_load(self.manifest_path.read_text())

        # check gold mine access
        if manifest.get("sources", {}).get("01_gold_mine", {}).get("enabled"):
            info(console, "Gold Mine access requested. Checking with customs...")
            gold_mine_info = manifest.get("sources", {}).get("01_gold_mine", {})
            clearance = check_gold_mine_clearance(gold_mine_info)

            if clearance.cleared:
                success(console, "Gold Mine access cleared by customs.")
            else:
                error(
                    console,
                    "Gold Mine access clearance failed. Please check your access and try again.",
                )
                console.print(clearance)
                raise ValueError("Gold Mine access clearance failed.")
                

        # check globus
        if manifest.get("sources", {}).get("02_globus", {}).get("enabled"):
            info(console, "Globus access requested. Checking with customs...")

            globus_info = manifest.get("sources", {}).get("02_globus", {})

            clearance = check_globus_clearance(
                globus_info=globus_info
                )
            
            if clearance.cleared:
                success(console, "Globus credentials validated.")
            else:
                error(
                    console,
                    "Globus credentials validation failed. Please check your credentials.",
                )
                console.print(clearance)
                raise ValueError("Globus credentials validation failed.")
        # check dataverse
        if manifest.get("sources", {}).get("03_dataverse", {}).get("enabled"):
            info(console, "Dataverse access requested. Checking with customs...")
            warning(console, "Dataverse access checks are not yet implemented. Please ensure you have access to the Dataverse dataset before proceeding.")

        if not checker.passes():
            console.print()
            failure_panel(console, "⚠️ Manifest checks failed. Please review the results above.")
            handbook_note(console)

            return False
        
        console.print()
        banner(console, "✅ Manifest checks passed, you're cleared for staging! Run `stagecoach stage` to bring the data into your environment.")
        
        return True
        

    def stage(
        self,
        console: Console | None = None,
    ) -> bool:
        """
        Stage data declared by the instance manifest.

        Parameters
        ----------
        console : Console | None, default=None
            Optional console override for this invocation.

        Returns
        -------
        bool
            ``True`` when every requested source stages successfully.
        """
        console = console or self.console
        banner(console, "🚂 Staging the data...")

        manifest = yaml.safe_load(self.manifest_path.read_text())

        # create stage directories
        create_stage_directories(manifest, console, gitignore=True)

        # fasrc
        fasrc = manifest.get("sources", {}).get("01_gold_mine", {})
        if fasrc['enabled']:
            fasrc_stageresult = stage_data_from_fasrc(manifest, console)
            if fasrc_stageresult.staged:
                success(console, "Gold Mine data staged successfully.")
            else:
                error(console, f"Gold Mine staging failed: {fasrc_stageresult.message}")
        else:
            fasrc_stageresult = StageResult(
                source="01_gold_mine",
                staged=True,
                message="Gold Mine staging not requested.",
            )
        
        # globus
        globus = manifest.get("sources", {}).get("02_globus", {})
        if globus['enabled']:
            globus_stageresult = stage_data_from_globus(manifest, console)
            if globus_stageresult.staged:
                success(console, "Globus data staged successfully.")
            else:
                error(console, f"Globus staging failed: {globus_stageresult.message}")
        else:
            globus_stageresult = StageResult(
                source="02_globus",
                staged=True,
                message="Globus staging not requested.",
            )

        # dataverse
        dataverse = manifest.get("sources", {}).get("03_dataverse", {})
        if dataverse['enabled']:
            warning(console, "Dataverse staging is not yet implemented. Please stage the data manually and place it in the appropriate directory.")
            dataverse_stageresult = StageResult(
                source="03_dataverse",
                staged=True,
                message="Dataverse staging is not yet implemented.",
            )
        else:
            dataverse_stageresult = StageResult(
                source="03_dataverse",
                staged=True,
                message="Dataverse staging not requested.",
            )
        
        final_staging = all(
            result.staged
            for result in [fasrc_stageresult, globus_stageresult, dataverse_stageresult]
        )

        if final_staging:
            banner(console, "🪎 All data staged successfully! You're ready to start working!")
            return True
        else:
            error(console, "Some data sources failed to stage. Please review the messages above and address any issues in your manifest.")

            failure_panel(console, "⚠️ Staging failed. Please try again.")
            return False
```

Click on each section below to learn how stagecoach works in more
detail, and to see examples of how to use it.

1.  [Manifest File](#manifest-file)


# Aesthetic UI


The following is a set of interchangeable UI components that will
hopefully make it easier to standardize the look of the tools across the
frontier.

``` python
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel

from stagecoach.checks import Severity


HANDBOOK_URL = (
    "https://goldenplanetaryhealthlab.github.io/"
    "01_orientation/start-here.html#the-working-philosophy"
)


def banner(console: Console, message: str) -> None:
    """
    Render a highlighted banner message.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    message : str
        Message to display inside the banner panel.
    """
    console.print(
        Panel.fit(
            message,
            style="bold cyan",
            border_style="cyan",
        )
    )


def success(console: Console, message: str) -> None:
    """
    Render a success message.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    message : str
        Message body to print.
    """
    console.print(f"[bold green]✓[/bold green] {escape(message)}")


def warning(console: Console, message: str) -> None:
    """
    Render a warning message.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    message : str
        Message body to print.
    """
    console.print(f"[bold yellow]⚠[/bold yellow] {escape(message)}")


def error(console: Console, message: str) -> None:
    """
    Render an error message.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    message : str
        Message body to print.
    """
    console.print(f"[bold red]✖[/bold red] {escape(message)}")


def info(console: Console, message: str) -> None:
    """
    Render an informational message.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    message : str
        Message body to print.
    """
    console.print(f"[bold cyan]•[/bold cyan] {escape(message)}")


def failure_panel(console: Console, message: str) -> None:
    """
    Render an error message inside a bordered panel.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    message : str
        Message body to print.
    """
    console.print(
        Panel.fit(
            f"[bold red]✖[/bold red] {escape(message)}",
            border_style="red",
        )
    )


def handbook_note(console: Console) -> None:
    """
    Print a link to the Frontier handbook principles.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    """
    console.print(
        "[dim]See handbook for explanation of principles:[/dim] "
        f"[link={HANDBOOK_URL}]{HANDBOOK_URL}[/link]"
    )


def format_principle(principle: int | list[int]) -> str:
    """
    Format one or more principle identifiers for display.

    Parameters
    ----------
    principle : int | list[int]
        Principle number or collection of principle numbers.

    Returns
    -------
    str
        Comma-separated principle identifiers.
    """
    if isinstance(principle, list):
        return ", ".join(str(item) for item in principle)

    return str(principle)


def check_result(console: Console, result) -> None:
    """
    Render a manifest check result using severity-specific styling.

    Parameters
    ----------
    console : Console
        Rich console used for output.
    result
        Object with ``name``, ``state``, ``message``, and ``principle``
        attributes, typically a ``CheckResult`` instance.
    """
    principle = format_principle(result.principle)

    if result.state == Severity.PASS:
        console.print(
            f"[bold green]✓ {escape(result.name)}[/bold green] "
            f"[dim]principle {principle}[/dim]\n"
            f"  {escape(result.message)}"
        )

    elif result.state == Severity.WARNING:
        console.print(
            f"[bold yellow]⚠ {escape(result.name)}[/bold yellow] "
            f"[dim]principle {principle}[/dim]\n"
            f"  {escape(result.message)}"
        )

    else:
        console.print(
            f"[bold red]✖ {escape(result.name)}[/bold red] "
            f"[dim]principle {principle}[/dim]\n"
            f"  {escape(result.message)}"
        )
```


# Validate Manifest


The goal of this module is simple: once the user has filled out the
manifest, we want to make sure all is compliant with the lab and
Frontier’s working philosophy. Once this is done, the user can run
`stagecoach` to stage the data they requested in their environment.

Here’s some dummy code to get us started:

``` python
def validate_manifest():
    print("Validating manifest...")
    return True
```