unsprawl.cli

Command-line interface and application orchestrator for Unsprawl.

This module contains: - UnsprawlApp: Main orchestrator class that wires the entire pipeline - CLI argument parser and main entry point - Integration logic for all components

Attributes

console
app
_TY_APP

Classes

UnsprawlApp

Application orchestrator wiring the pipeline and providing both programmatic and

Functions

cmd_docs([port, open_browser])	Launch Sphinx documentation server with live reload.
cmd_cache([clear, transport_cache_dir])	Manage transport scoring cache.
cmd_fetch([region, limit, datasets, force])	Fetch and cache regional datasets.
cmd_advect([agents, steps, grid_res, device, ...])	Run GPU-accelerated agent advection simulation.
cmd_advect_replay([agents, frames, steps_per_frame, ...])	Generate replay frames for Streamlit dashboard.
cmd_valuate([region, input_path, mrt_catalog, ...])	Run property valuation and scoring pipeline.
showcase([scenario, dev])	🚀 Launch the Full Stack Showcase (Backend + Frontend).
main([argv])	Entry point that dispatches to Typer app.

Module Contents

console

class UnsprawlApp(schema=None, transport_cache_dir=None)

Application orchestrator wiring the pipeline and providing both programmatic and CLI access.

This class can be used directly as a Python module or via the CLI. For programmatic usage, use the process() method with explicit parameters. For CLI usage, use the run() method with parsed arguments.

Example (Module Usage)

>>> app = UnsprawlApp()
>>> results = app.process(
...     input_path="resale.csv",
...     town="PUNGGOL",
...     budget=600000,
...     top_n=10
... )
>>> print(results.head())

Example (With MRT Accessibility - Default)

>>> app = UnsprawlApp()
>>> results = app.process(
...     input_path="resale.csv",
...     town="BISHAN"
... )

Example (Custom MRT Catalog)

>>> results = app.process(
...     input_path="resale.csv",
...     mrt_catalog="stations.geojson",
...     town="BISHAN"
... )

Initialize the valuation engine with optional custom schema and cache directory.

Parameters:

• schema (Schema | None) – Custom schema definition. If None, uses default Schema().

• transport_cache_dir (Optional[str]) – Directory for caching transport KDTree data. If None, uses default .cache_transport.

schema

loader

fe

engine

transport

reporter

logger

_data: DataFrame | None = None

load_data(input_path)

Load HDB resale data from CSV file.

Parameters:

input_path (str) – Path to the HDB resale CSV file.

Returns:

Loaded and normalized DataFrame.

Return type:

pd.DataFrame

Raises:

• FileNotFoundError – If the file does not exist.

• ValueError – If the CSV cannot be parsed.

process(input_path=None, data=None, mrt_catalog=None, clear_transport_cache=False, group_by=None, enable_accessibility_adjust=True, town=None, town_like=None, budget=None, flat_type=None, flat_type_like=None, flat_model=None, flat_model_like=None, storey_min=None, storey_max=None, area_min=None, area_max=None, lease_min=None, lease_max=None, top_n=10, return_full=False)

Process HDB resale data and return filtered, scored results.

This is the main programmatic entry point for using the valuation engine as a module.

Parameters:

• input_path (Optional[str]) – Path to HDB resale CSV. Required if data is not provided.

• data (Optional[pd.DataFrame]) – Pre-loaded DataFrame. If provided, input_path is ignored.

• mrt_catalog (Optional[str]) – Path to MRT stations GeoJSON or CSV for transport scoring.

• clear_transport_cache (bool) – Whether to clear transport cache before processing.

• group_by (Optional[List[str]]) – Columns to group by for peer comparison z-scores. Defaults to [town, flat_type].

• enable_accessibility_adjust (bool) – Whether to adjust price efficiency based on MRT accessibility. Default True.

• town (Optional[str]) – Exact town filter (case-insensitive).

• town_like (Optional[str]) – Partial town match (substring).

• budget (Optional[float]) – Maximum resale price.

• flat_type (Optional[str]) – Exact flat type filter.

• flat_type_like (Optional[str]) – Partial flat type match.

• flat_model (Optional[str]) – Exact flat model filter.

• flat_model_like (Optional[str]) – Partial flat model match.

• storey_min (Optional[int]) – Minimum storey number.

• storey_max (Optional[int]) – Maximum storey number.

• area_min (Optional[float]) – Minimum floor area (sqm).

• area_max (Optional[float]) – Maximum floor area (sqm).

• lease_min (Optional[float]) – Minimum remaining lease (years).

• lease_max (Optional[float]) – Maximum remaining lease (years).

• top_n (int) – Number of top results to return. Default 10.

• return_full (bool) – If True, return all filtered results instead of just top_n.

Returns:

Filtered and scored results, sorted by valuation_score descending.

Return type:

pd.DataFrame

Raises:

• ValueError – If neither input_path nor data is provided and the default dataset path is not available.

• FileNotFoundError – If input_path does not exist.

Examples

>>> app = UnsprawlApp()
>>> results = app.process(
...     input_path="resale.csv",
...     town="PUNGGOL",
...     budget=600000,
...     top_n=5
... )
>>> print(f"Found {len(results)} undervalued properties")

render_report(data=None, town=None, town_like=None, budget=None, flat_type=None, flat_type_like=None, flat_model=None, flat_model_like=None, storey_min=None, storey_max=None, area_min=None, area_max=None, lease_min=None, lease_max=None, top_n=10)

Render a formatted string report from processed data.

Parameters:

• data (Optional[pd.DataFrame]) – Pre-processed DataFrame with scores. If None, uses internally stored data.

• top_n (int) – Number of results to include in report.

Notes

This method accepts the same filter arguments as process().

Returns:

Formatted table string ready for console output.

Return type:

str

render_rich_table(df, title='🏠 Top Undervalued Residential Properties')

Render a Rich table from results DataFrame.

Parameters:

• df (pd.DataFrame) – Results DataFrame with valuation scores.

• title (str) – Table title.

Returns:

Formatted Rich table ready for console output.

Return type:

rich.table.Table

app

_TY_APP

cmd_docs(port=typer.Option(8000, help='Docs server port'), open_browser=typer.Option(True, help='Open browser after launch'))

Launch Sphinx documentation server with live reload.

cmd_cache(clear=typer.Option(False, help='Clear transport cache and exit'), transport_cache_dir=typer.Option(None, help='Transport cache directory'))

Manage transport scoring cache.

cmd_fetch(region=typer.Option('SG', help='Region (currently SG)'), limit=typer.Option(5000, help='Synthetic fallback row cap'), datasets=typer.Option('all', help='Datasets: all|resale|mrt'), force=typer.Option(False, help='Force re-download/regenerate'))

Fetch and cache regional datasets.

cmd_advect(agents=100000, steps=600, grid_res=256, device=None, amplitude=1.0, frequency=3.0, phase=0.0, dt=1.0 / 60.0, output=None, output_format='npy')

Run GPU-accelerated agent advection simulation.

cmd_advect_replay(agents=100000, frames=60, steps_per_frame=5, grid_res=256, device=None, out_dir=str(Path('dashboard') / 'data' / 'advect_frames'), stride=10, parquet=None, fps=10.0)

Generate replay frames for Streamlit dashboard.

cmd_valuate(region=typer.Option('SG', help='Region code'), input_path=typer.Option(None, '--input', help='Path to resale CSV'), mrt_catalog=typer.Option(None, help='MRT GeoJSON/CSV path (or blank to skip)'), transport_cache_dir=typer.Option(None, help='Transport cache directory'), clear_transport_cache=typer.Option(False, help='Clear cache before scoring'), town=typer.Option(None, help='Exact town filter'), town_like=typer.Option(None, help='Substring town filter'), flat_type=typer.Option(None, help='Exact flat type'), flat_type_like=typer.Option(None, help='Substring flat type'), flat_model=typer.Option(None, help='Exact flat model'), flat_model_like=typer.Option(None, help='Substring flat model'), storey_min=typer.Option(None, help='Minimum storey'), storey_max=typer.Option(None, help='Maximum storey'), area_min=typer.Option(None, help='Minimum area'), area_max=typer.Option(None, help='Maximum area'), lease_min=typer.Option(None, help='Minimum lease years'), lease_max=typer.Option(None, help='Maximum lease years'), budget=typer.Option(None, help='Maximum price'), group_by=typer.Option(None, help='Group-by columns for z-scores (comma-separated)'), top=typer.Option(10, help='Top-N to display'), no_accessibility_adjust=typer.Option(False, help='Do not adjust price_efficiency'), output=typer.Option(None, help='Optional export path (csv/json/parquet)'), export_full=typer.Option(False, help='Export full filtered dataset'), output_format=typer.Option('csv', help='Export format: csv|json|parquet'))

Run property valuation and scoring pipeline.

showcase(scenario=typer.Argument('genesis', help='Simulation scenario to load'), dev=typer.Option(True, help='Run in dev mode (hot reload)'))

🚀 Launch the Full Stack Showcase (Backend + Frontend).

main(argv=None)

Entry point that dispatches to Typer app.

Keeps return code semantics for tests, and supports legacy calls without a subcommand by defaulting to the valuate command when argv starts with flags.