Command line

The gen_surv command is a Typer application. These are the functions behind its two subcommands; they can also be called directly from Python.

cli

Command-line interface for gen_surv.

This module provides a command-line interface for generating survival data using the gen_surv package.

dataset

dataset(
    model: str = Argument(
        ...,
        help="Model to simulate [cphm, cmm, tdcm, thmm, aft_ln, aft_weibull, aft_log_logistic, competing_risks, competing_risks_weibull, mixture_cure, piecewise_exponential, recurrent_events]",
    ),
    n: int = Option(100, help="Number of samples"),
    model_cens: str = Option(
        "uniform",
        help="Censoring model: 'uniform' or 'exponential'",
    ),
    cens_par: float = Option(
        1.0, help="Censoring parameter"
    ),
    beta: List[float] = Option(
        [0.5],
        help="Regression coefficient(s). Provide multiple values for multi-parameter models.",
    ),
    covariate_range: float | None = Option(
        2.0,
        "--covariate-range",
        "--covar",
        help="Upper bound for covariate values (for CPHM, CMM, THMM)",
    ),
    sigma: float | None = Option(
        1.0,
        help="Standard deviation parameter (for log-normal AFT)",
    ),
    shape: float | None = Option(
        1.5, help="Shape parameter (for Weibull AFT)"
    ),
    scale: float | None = Option(
        2.0, help="Scale parameter (for Weibull AFT)"
    ),
    n_risks: int = Option(
        2, help="Number of competing risks"
    ),
    baseline_hazards: List[float] = Option(
        [], help="Baseline hazards for competing risks"
    ),
    shape_params: List[float] = Option(
        [],
        help="Shape parameters for Weibull competing risks",
    ),
    scale_params: List[float] = Option(
        [],
        help="Scale parameters for Weibull competing risks",
    ),
    cure_fraction: float | None = Option(
        None, help="Cure fraction for mixture cure model"
    ),
    baseline_hazard: float | None = Option(
        None, help="Baseline hazard for mixture cure model"
    ),
    breakpoints: List[float] = Option(
        [],
        help="Breakpoints for piecewise exponential model",
    ),
    hazard_rates: List[float] = Option(
        [],
        help="Hazard rates for piecewise exponential model",
    ),
    process: str = Option(
        "ag",
        help="Recurrent event process: 'ag' (Andersen-Gill), 'pwp_tt' or 'pwp_gt' (Prentice-Williams-Peterson in total or gap time)",
    ),
    baseline: str = Option(
        "exponential",
        help="Baseline hazard for recurrent events: 'exponential', 'weibull' or 'gompertz'",
    ),
    rate: List[float] = Option(
        [],
        help="Rate parameter(s). One value for recurrent events (exponential and Gompertz baselines); six for cmm and three for thmm, repeating the flag",
    ),
    dist: str = Option(
        "weibull",
        help="Marginal distribution for tdcm: 'weibull' or 'exponential'",
    ),
    corr: float = Option(
        0.5,
        help="Correlation between the covariate and the crossover time (tdcm)",
    ),
    dist_par: List[float] = Option(
        [],
        help="Distribution parameters for tdcm: four values for 'weibull', two for 'exponential', repeating the flag",
    ),
    lam: float = Option(
        1.0, help="Baseline hazard rate for tdcm"
    ),
    stratum_effects: List[float] = Option(
        [],
        help="Per-event intensity factors for the PWP recurrent processes",
    ),
    max_events: int | None = Option(
        None,
        help="Stop following a subject after this many recurrent events",
    ),
    followup_time: float = Option(
        10.0,
        help="Administrative end of follow-up for recurrent events",
    ),
    seed: int | None = Option(
        None, help="Random seed for reproducibility"
    ),
    output: str | None = Option(
        None,
        "-o",
        help="Output CSV file. Prints to stdout if omitted.",
    ),
) -> None

Generate survival data and optionally save to CSV.

Examples: # Generate data from CPHM model $ gen_surv dataset cphm --n 100 --beta 0.5 --covariate-range 2.0 -o cphm_data.csv

# Generate data from Weibull AFT model
$ gen_surv dataset aft_weibull --n 200 --beta 0.5 --beta -0.3 --shape 1.5 --scale 2.0 -o aft_data.csv

Source code in gen_surv/cli.py

@app.command()
def dataset(
    model: str = typer.Argument(
        ...,
        help=(
            "Model to simulate [cphm, cmm, tdcm, thmm, aft_ln, aft_weibull, aft_log_logistic, competing_risks, competing_risks_weibull, mixture_cure, piecewise_exponential, recurrent_events]"
        ),
    ),
    n: int = typer.Option(100, help="Number of samples"),
    model_cens: str = typer.Option(
        "uniform", help="Censoring model: 'uniform' or 'exponential'"
    ),
    cens_par: float = typer.Option(1.0, help="Censoring parameter"),
    beta: List[float] = typer.Option(
        [0.5],
        help="Regression coefficient(s). Provide multiple values for multi-parameter models.",
    ),
    covariate_range: float | None = typer.Option(
        2.0,
        "--covariate-range",
        "--covar",
        help="Upper bound for covariate values (for CPHM, CMM, THMM)",
    ),
    sigma: float | None = typer.Option(
        1.0, help="Standard deviation parameter (for log-normal AFT)"
    ),
    shape: float | None = typer.Option(1.5, help="Shape parameter (for Weibull AFT)"),
    scale: float | None = typer.Option(2.0, help="Scale parameter (for Weibull AFT)"),
    n_risks: int = typer.Option(2, help="Number of competing risks"),
    baseline_hazards: List[float] = typer.Option(
        [], help="Baseline hazards for competing risks"
    ),
    shape_params: List[float] = typer.Option(
        [], help="Shape parameters for Weibull competing risks"
    ),
    scale_params: List[float] = typer.Option(
        [], help="Scale parameters for Weibull competing risks"
    ),
    cure_fraction: float | None = typer.Option(
        None, help="Cure fraction for mixture cure model"
    ),
    baseline_hazard: float | None = typer.Option(
        None, help="Baseline hazard for mixture cure model"
    ),
    breakpoints: List[float] = typer.Option(
        [], help="Breakpoints for piecewise exponential model"
    ),
    hazard_rates: List[float] = typer.Option(
        [], help="Hazard rates for piecewise exponential model"
    ),
    process: str = typer.Option(
        "ag",
        help=(
            "Recurrent event process: 'ag' (Andersen-Gill), 'pwp_tt' or "
            "'pwp_gt' (Prentice-Williams-Peterson in total or gap time)"
        ),
    ),
    baseline: str = typer.Option(
        "exponential",
        help="Baseline hazard for recurrent events: 'exponential', 'weibull' or 'gompertz'",
    ),
    rate: List[float] = typer.Option(
        [],
        help=(
            "Rate parameter(s). One value for recurrent events (exponential and "
            "Gompertz baselines); six for cmm and three for thmm, repeating the flag"
        ),
    ),
    dist: str = typer.Option(
        "weibull", help="Marginal distribution for tdcm: 'weibull' or 'exponential'"
    ),
    corr: float = typer.Option(
        0.5, help="Correlation between the covariate and the crossover time (tdcm)"
    ),
    dist_par: List[float] = typer.Option(
        [],
        help=(
            "Distribution parameters for tdcm: four values for 'weibull', two for "
            "'exponential', repeating the flag"
        ),
    ),
    lam: float = typer.Option(1.0, help="Baseline hazard rate for tdcm"),
    stratum_effects: List[float] = typer.Option(
        [], help="Per-event intensity factors for the PWP recurrent processes"
    ),
    max_events: int | None = typer.Option(
        None, help="Stop following a subject after this many recurrent events"
    ),
    followup_time: float = typer.Option(
        10.0, help="Administrative end of follow-up for recurrent events"
    ),
    seed: int | None = typer.Option(None, help="Random seed for reproducibility"),
    output: str | None = typer.Option(
        None, "-o", help="Output CSV file. Prints to stdout if omitted."
    ),
) -> None:
    """Generate survival data and optionally save to CSV.

    Examples:
        # Generate data from CPHM model
        $ gen_surv dataset cphm --n 100 --beta 0.5 --covariate-range 2.0 -o cphm_data.csv

        # Generate data from Weibull AFT model
        $ gen_surv dataset aft_weibull --n 200 --beta 0.5 --beta -0.3 --shape 1.5 --scale 2.0 -o aft_data.csv
    """
    # Helper to unwrap Typer Option defaults when function is called directly
    from typer.models import OptionInfo

    T = TypeVar("T")

    def _val(v: T | OptionInfo) -> T:
        return v if not isinstance(v, OptionInfo) else cast(T, v.default)

    # Prepare arguments based on the selected model
    model_str: str = _val(model)
    kwargs: Dict[str, Any] = {
        "model": model_str,
        "n": _val(n),
        "model_cens": _val(model_cens),
        "cens_par": _val(cens_par),
        "seed": _val(seed),
    }

    # Add model-specific parameters
    if model_str == "cphm":
        # A single coefficient and a covariate range.
        beta_values = cast(List[float], _val(beta))
        kwargs["beta"] = beta_values[0] if len(beta_values) > 0 else 0.5
        kwargs["covariate_range"] = _val(covariate_range)

    elif model_str in ["cmm", "thmm"]:
        # Three coefficients, one per transition, and a rate vector: six values
        # for cmm (an intensity and a shape per transition), three for thmm.
        kwargs["beta"] = _val(beta)
        kwargs["covariate_range"] = _val(covariate_range)
        rates = cast(List[float], _val(rate))
        if rates:
            kwargs["rate"] = rates
        elif model_str == "cmm":
            kwargs["rate"] = [0.1, 1.0, 0.2, 1.0, 0.1, 1.0]
        else:
            kwargs["rate"] = [0.2, 0.3, 0.4]

    elif model_str == "tdcm":
        # The bivariate draw's parameters have no counterpart in the other
        # models, so they get their own options.
        kwargs["beta"] = _val(beta)
        kwargs["dist"] = _val(dist)
        kwargs["corr"] = _val(corr)
        kwargs["lam"] = _val(lam)
        parameters = cast(List[float], _val(dist_par))
        if parameters:
            kwargs["dist_par"] = parameters
        elif _val(dist) == "weibull":
            kwargs["dist_par"] = [1.0, 2.0, 1.0, 2.0]
        else:
            kwargs["dist_par"] = [1.0, 2.0]

    elif model_str == "aft_ln":
        # Log-normal AFT model uses beta list and sigma
        kwargs["beta"] = _val(beta)
        kwargs["sigma"] = _val(sigma)

    elif model_str == "aft_weibull":
        # Weibull AFT model uses beta list, shape, and scale
        kwargs["beta"] = _val(beta)
        kwargs["shape"] = _val(shape)
        kwargs["scale"] = _val(scale)

    elif model_str == "aft_log_logistic":
        kwargs["beta"] = _val(beta)
        kwargs["shape"] = _val(shape)
        kwargs["scale"] = _val(scale)

    elif model_str == "competing_risks":
        kwargs["n_risks"] = _val(n_risks)
        if _val(baseline_hazards):
            kwargs["baseline_hazards"] = _val(baseline_hazards)
        if _val(beta):
            kwargs["betas"] = [_val(beta) for _ in range(_val(n_risks))]

    elif model_str == "competing_risks_weibull":
        kwargs["n_risks"] = _val(n_risks)
        if _val(shape_params):
            kwargs["shape_params"] = _val(shape_params)
        if _val(scale_params):
            kwargs["scale_params"] = _val(scale_params)
        if _val(beta):
            kwargs["betas"] = [_val(beta) for _ in range(_val(n_risks))]

    elif model_str == "mixture_cure":
        if _val(cure_fraction) is not None:
            kwargs["cure_fraction"] = _val(cure_fraction)
        if _val(baseline_hazard) is not None:
            kwargs["baseline_hazard"] = _val(baseline_hazard)
        kwargs["betas_survival"] = _val(beta)
        kwargs["betas_cure"] = _val(beta)

    elif model_str == "piecewise_exponential":
        kwargs["breakpoints"] = _val(breakpoints)
        kwargs["hazard_rates"] = _val(hazard_rates)
        kwargs["betas"] = _val(beta)

    elif model_str == "recurrent_events":
        baseline_str: str = _val(baseline)
        kwargs["process"] = _val(process)
        kwargs["baseline"] = baseline_str
        kwargs["followup_time"] = _val(followup_time)

        # Only the keys that belong to the chosen baseline: the generator
        # rejects the others rather than ignoring them.
        if baseline_str == "exponential":
            kwargs["baseline_params"] = {"rate": _first_rate(_val(rate))}
        elif baseline_str == "weibull":
            kwargs["baseline_params"] = {
                "shape": _val(shape),
                "scale": _val(scale),
            }
        elif baseline_str == "gompertz":
            kwargs["baseline_params"] = {
                "rate": _first_rate(_val(rate)),
                "shape": _val(shape),
            }

        if _val(beta):
            kwargs["betas"] = _val(beta)
        if _val(stratum_effects):
            kwargs["stratum_effects"] = _val(stratum_effects)
        if _val(max_events) is not None:
            kwargs["max_events"] = _val(max_events)

    # Generate the data
    try:
        df = generate(**kwargs)
    except ValidationError as exc:
        typer.echo(f"Input error: {exc}")
        raise typer.Exit(1)

    # Output the data
    if output:
        df.to_csv(output, index=False)
        typer.echo(f"Saved dataset to {output}")
    else:
        typer.echo(df.to_csv(index=False))

visualize

visualize(
    input_file: str = Argument(
        ..., help="Input CSV file containing survival data"
    ),
    time_col: str = Option(
        "time",
        help="Column containing time/duration values",
    ),
    status_col: str = Option(
        "status",
        help="Column containing event indicator (1=event, 0=censored)",
    ),
    group_col: str | None = Option(
        None, help="Column to use for stratification"
    ),
    output: str = Option(
        "survival_plot.png", help="Output image file"
    ),
) -> None

Visualize survival data from a CSV file.

Examples: # Generate a Kaplan-Meier plot from a CSV file $ gen_surv visualize data.csv --time-col time --status-col status -o km_plot.png

# Generate a stratified plot using a grouping variable
$ gen_surv visualize data.csv --group-col X0 -o stratified_plot.png

Source code in gen_surv/cli.py

@app.command()
def visualize(
    input_file: str = typer.Argument(
        ..., help="Input CSV file containing survival data"
    ),
    time_col: str = typer.Option("time", help="Column containing time/duration values"),
    status_col: str = typer.Option(
        "status", help="Column containing event indicator (1=event, 0=censored)"
    ),
    group_col: str | None = typer.Option(None, help="Column to use for stratification"),
    output: str = typer.Option("survival_plot.png", help="Output image file"),
) -> None:
    """Visualize survival data from a CSV file.

    Examples:
        # Generate a Kaplan-Meier plot from a CSV file
        $ gen_surv visualize data.csv --time-col time --status-col status -o km_plot.png

        # Generate a stratified plot using a grouping variable
        $ gen_surv visualize data.csv --group-col X0 -o stratified_plot.png
    """
    try:
        import matplotlib.pyplot as plt
        import pandas as pd

        from gen_surv.visualization import plot_survival_curve
    except ImportError:
        typer.echo(
            "Error: Visualization requires matplotlib and lifelines. "
            "Install them with: pip install matplotlib lifelines"
        )
        raise typer.Exit(1)

    # Load the data
    try:
        data = pd.read_csv(input_file)
    except Exception as e:
        typer.echo(f"Error loading CSV file: {str(e)}")
        raise typer.Exit(1)

    # Check required columns
    if time_col not in data.columns:
        typer.echo(f"Error: Time column '{time_col}' not found in data")
        raise typer.Exit(1)

    if status_col not in data.columns:
        typer.echo(f"Error: Status column '{status_col}' not found in data")
        raise typer.Exit(1)

    if group_col is not None and group_col not in data.columns:
        typer.echo(f"Error: Group column '{group_col}' not found in data")
        raise typer.Exit(1)

    # Create the plot
    fig, ax = plot_survival_curve(
        data=data, time_col=time_col, status_col=status_col, group_col=group_col
    )

    # Save the plot
    plt.savefig(output, dpi=300, bbox_inches="tight")
    plt.close(fig)
    typer.echo(f"Plot saved to {output}")