cli ¶

@app.callback(
    epilog="Check out the [link=https://github.com/Hydrowelder/trendify/]full documentation[/link] for more info."
)
def main(
    ctx: typer.Context,
    version: bool = typer.Option(
        False,
        "--version",
        callback=version_callback,
        is_eager=True,
        help="Show version and exit.",
    ),
) -> None:
    """
    [bold indian_red1]Trendify:[/bold indian_red1] Generate visual records and static assets from raw data.
    """

@app.command(name="generate", context_settings={"allow_extra_args": True})
def generate(
    ctx: typer.Context,
    input_directories: list[str] = InputDirectoriesOption,
    record_generator: str = RecordGeneratorOption,
    output_directory: Path = OutputDirectoryOption,
    n_procs: int = NProcsOption,
    verbose: int = VerboseOption,
    quiet: int = QuietOption,
) -> None:
    """Generate tagged records from raw data directories."""
    _configure_logging(verbose, quiet)
    pipeline = TrendifyPipeline(output_dir=output_directory, n_procs=n_procs)
    _total = pipeline.generate(
        record_generator=_resolve_record_generator(record_generator),
        data_dirs=_resolve_input_directories(input_directories, ctx),
    )

@app.command(name="render")
def render(
    ctx: typer.Context,
    output_directory: Path = OutputDirectoryOption,
    verbose: int = VerboseOption,
    quiet: int = QuietOption,
) -> None:
    """Render CSV tables and matplotlib figures from already-generated records."""
    _configure_logging(verbose, quiet)
    pipeline = TrendifyPipeline(output_dir=output_directory)
    pipeline.render()
    typer.echo(f"Rendered assets to {pipeline.assets_dir}")

@app.command(name="run", context_settings={"allow_extra_args": True})
def run(
    ctx: typer.Context,
    input_directories: list[str] = InputDirectoriesOption,
    record_generator: str = RecordGeneratorOption,
    output_directory: Path = OutputDirectoryOption,
    n_procs: int = NProcsOption,
    verbose: int = VerboseOption,
    quiet: int = QuietOption,
) -> None:
    """Generate records and render assets in one step."""
    logger = _configure_logging(verbose, quiet)
    pipeline = TrendifyPipeline(output_dir=output_directory, n_procs=n_procs)
    total = pipeline.run(
        record_generator=_resolve_record_generator(record_generator),
        data_dirs=_resolve_input_directories(input_directories, ctx),
    )
    logger.info(f"Wrote {total} records; assets under {pipeline.assets_dir}")

@app.command(name="example-data")
def example_data(
    ctx: typer.Context,
    workdir: Path = ExampleWorkdirOption,
    n_folders: int = ExampleNFoldersOption,
    n_cols: int = ExampleNColsOption,
    verbose: int = VerboseOption,
    quiet: int = QuietOption,
) -> None:
    """Generate sample raw-data directories for exercising the trendify pipeline."""
    logger = _configure_logging(verbose, quiet)
    make_example_data(workdir=workdir, n_folders=n_folders, n_cols=n_cols)
    logger.info(f"Wrote {n_folders} sample data folders under {workdir / 'models'}")

def get_local_ip() -> str:
    """Returns the actual local IP address of this machine."""
    import socket

    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        # Does not actually need to connect to 8.8.8.8 to work
        s.connect(("8.8.8.8", 1))
        ip = s.getsockname()[0]
    except Exception:
        ip = "127.0.0.1"
    finally:
        s.close()
    return ip

@app.command(name="viewer")
def viewer(
    ctx: typer.Context,
    db_path: Path = DbPathArgument,
    host: str = HostOption,
    port: int = PortOption,
    verbose: int = VerboseOption,
    quiet: int = QuietOption,
) -> None:
    """Launch a local web dashboard for browsing a trendify.db's records."""
    _configure_logging(verbose, quiet, logo=False)
    db_path = db_path.resolve()
    if not db_path.exists():
        raise typer.BadParameter(f"No such database file: {db_path}")

    # Fail fast on a bad/corrupt db before uvicorn even starts, rather than a 500 on first request.
    with RecordStore.open(db_path, readonly=True):
        pass

    # Deferred import: keeps `trendify.cli` import light for callers of generate/render/run
    # who never touch the dashboard (FastAPI/Jinja2/uvicorn stay unimported until needed).
    from trendify.viewer.app import create_app, make_server

    connection_info = (
        f"Local: [bold indian_red1 u]http://127.0.0.1:{port}[/bold indian_red1 u]"
    )
    if host == "0.0.0.0":
        connection_info += f"\nMobile: [bold indian_red1 u]http://{get_local_ip()}:{port}[/bold indian_red1 u]"
    else:
        connection_info += "\n\n[dim]Tip: To view on other devices (and to make pages shareable), run with[/dim] [yellow]--host 0.0.0.0[/yellow]"

    console.print(
        Panel(
            f"""\n{connection_info}\n\n[yellow]Press CTRL+C to stop[/yellow]""",
            border_style="indian_red1",
            expand=False,
            title="Trendify Viewer is Live!",
        )
    )

    make_server(create_app(db_path), host=host, port=port).run()