23063344532

Committed 13 Mar 2026 05:44PM UTC coverage: 91.645% (-2.0%) from 93.683%

Build # 23063344532

Build Type

Pull #11

github

Committed by

web-flow

Commit Message

Merge 35d59e3f2 into 1a20ac061

Pull Request Pull Request #11: Streamlit frontend, CI workflow, and tests with pipeline refactor

Coverage Stats

1785 of 1969 new or added lines in 26 files covered. (90.66%)

11 existing lines in 4 files now uncovered.

2490 of 2717 relevant lines covered (91.65%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.75

/app/utils/location_utils.py

"""Shared location and cluster helper utilities."""

from __future__ import annotations

import numpy as np
import pandas as pd


NYC_LAT_MIN = 40.49
NYC_LAT_MAX = 40.92
NYC_LNG_MIN = -74.27
NYC_LNG_MAX = -73.68

DEFAULT_SURVIVAL_MONTHS = [12, 36, 60, 120]


def clamp_to_nyc_bounds(latitude: float, longitude: float) -> tuple[float, float]:
    """Clamp a latitude/longitude pair to the NYC study area bounds."""
    lat = min(max(latitude, NYC_LAT_MIN), NYC_LAT_MAX)
    lng = min(max(longitude, NYC_LNG_MIN), NYC_LNG_MAX)
    return lat, lng


def build_cluster_reference_df(
    reference_df: pd.DataFrame,
    lat_column: str,
    lng_column: str,
    cluster_column: str = "location_cluster",
) -> pd.DataFrame:
    """Extract unique cluster centroid rows from a reference dataframe."""
    required = {lat_column, lng_column}
    if not required.issubset(reference_df.columns):
        return pd.DataFrame()

    cluster_cols = [
        column
        for column in [cluster_column, lat_column, lng_column]
        if column in reference_df.columns
    ]

    cluster_df = reference_df[cluster_cols].rename(
        columns={
            cluster_column: "location_cluster",
            lat_column: "location_cluster_lat",
            lng_column: "location_cluster_lng",
        }
    )

    return (
        cluster_df
        .dropna(subset=["location_cluster_lat", "location_cluster_lng"])
        .drop_duplicates()
        .reset_index(drop=True)
    )


def assign_nearest_cluster_info(
    latitude: float,
    longitude: float,
    cluster_df: pd.DataFrame,
) -> tuple[float | None, float, float]:
    """Find the nearest cluster centroid to a latitude/longitude pair."""
    if cluster_df.empty:
        return None, float(latitude), float(longitude)

    cluster_coords = cluster_df[
        ["location_cluster_lat", "location_cluster_lng"]
    ].to_numpy(dtype=float)
    point = np.array([latitude, longitude], dtype=float)

    distances = np.sqrt(((cluster_coords - point) ** 2).sum(axis=1))
    nearest_idx = int(np.argmin(distances))

    cluster_id: float | None = None
    if "location_cluster" in cluster_df.columns:
        raw_cluster = cluster_df.loc[nearest_idx, "location_cluster"]
        if pd.notna(raw_cluster):
            cluster_id = float(raw_cluster)

    return (
        cluster_id,
        float(cluster_df.loc[nearest_idx, "location_cluster_lat"]),
        float(cluster_df.loc[nearest_idx, "location_cluster_lng"]),
    )

1	"""Shared location and cluster helper utilities."""
2
3	from __future__ import annotations	1✔
4
5	import numpy as np	1✔
6	import pandas as pd	1✔
7
8
9	NYC_LAT_MIN = 40.49	1✔
10	NYC_LAT_MAX = 40.92	1✔
11	NYC_LNG_MIN = -74.27	1✔
12	NYC_LNG_MAX = -73.68	1✔
13
14	DEFAULT_SURVIVAL_MONTHS = [12, 36, 60, 120]	1✔
15
16
17	def clamp_to_nyc_bounds(latitude: float, longitude: float) -> tuple[float, float]:	1✔
18	"""Clamp a latitude/longitude pair to the NYC study area bounds."""
19	lat = min(max(latitude, NYC_LAT_MIN), NYC_LAT_MAX)	1✔
20	lng = min(max(longitude, NYC_LNG_MIN), NYC_LNG_MAX)	1✔
21	return lat, lng	1✔
22
23
24	def build_cluster_reference_df(	1✔
25	reference_df: pd.DataFrame,
26	lat_column: str,
27	lng_column: str,
28	cluster_column: str = "location_cluster",
29	) -> pd.DataFrame:
30	"""Extract unique cluster centroid rows from a reference dataframe."""
31	required = {lat_column, lng_column}	1✔
32	if not required.issubset(reference_df.columns):	1✔
NEW 33	return pd.DataFrame()	×
34
35	cluster_cols = [	1✔
36	column
37	for column in [cluster_column, lat_column, lng_column]
38	if column in reference_df.columns
39	]
40
41	cluster_df = reference_df[cluster_cols].rename(	1✔
42	columns={
43	cluster_column: "location_cluster",
44	lat_column: "location_cluster_lat",
45	lng_column: "location_cluster_lng",
46	}
47	)
48
49	return (	1✔
50	cluster_df
51	.dropna(subset=["location_cluster_lat", "location_cluster_lng"])
52	.drop_duplicates()
53	.reset_index(drop=True)
54	)
55
56
57	def assign_nearest_cluster_info(	1✔
58	latitude: float,
59	longitude: float,
60	cluster_df: pd.DataFrame,
61	) -> tuple[float \| None, float, float]:
62	"""Find the nearest cluster centroid to a latitude/longitude pair."""
63	if cluster_df.empty:	1✔
NEW 64	return None, float(latitude), float(longitude)	×
65
66	cluster_coords = cluster_df[	1✔
67	["location_cluster_lat", "location_cluster_lng"]
68	].to_numpy(dtype=float)
69	point = np.array([latitude, longitude], dtype=float)	1✔
70
71	distances = np.sqrt(((cluster_coords - point) ** 2).sum(axis=1))	1✔
72	nearest_idx = int(np.argmin(distances))	1✔
73
74	cluster_id: float \| None = None	1✔
75	if "location_cluster" in cluster_df.columns:	1✔
76	raw_cluster = cluster_df.loc[nearest_idx, "location_cluster"]	1✔
77	if pd.notna(raw_cluster):	1✔
78	cluster_id = float(raw_cluster)	1✔
79
80	return (	1✔
81	cluster_id,
82	float(cluster_df.loc[nearest_idx, "location_cluster_lat"]),
83	float(cluster_df.loc[nearest_idx, "location_cluster_lng"]),
84	)

sun770311 / bizsurvival515 / 23063344532

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous