diff --git a/coverage-badge.svg b/coverage-badge.svg index 812277b..7e909ff 100644 --- a/coverage-badge.svg +++ b/coverage-badge.svg @@ -1 +1 @@ -coverage: 37.58%coverage37.58% \ No newline at end of file +coverage: 39.34%coverage39.34% \ No newline at end of file diff --git a/src/ertimes/clean.py b/src/ertimes/clean.py index fd22e27..ced510d 100644 --- a/src/ertimes/clean.py +++ b/src/ertimes/clean.py @@ -1,5 +1,10 @@ import pandas as pd +# a set of functions that check +# always used inside the other functions +# change the columns in the dataframe +# no function + # Mapping of abbreviated column names to readable names COLUMN_RENAME_MAP = { 'oshpd_id': 'facility_id', diff --git a/src/ertimes/stats.py b/src/ertimes/stats.py index 7f51804..77d25f4 100644 --- a/src/ertimes/stats.py +++ b/src/ertimes/stats.py @@ -10,8 +10,37 @@ from pathlib import Path from folium.plugins import MarkerCluster +def _resolve_columns(column_map: dict[str, str] | None, columns: list[str]) -> dict[str, str]: + """ + Resolve column names using a mapping dictionary. + + If column_map is provided, maps each column name to its mapped value if present, + otherwise uses the original column name. + + Parameters + ---------- + column_map : dict[str, str] | None + Dictionary mapping column names to their actual names in the DataFrame. + columns : list[str] + List of column names to resolve. + + Returns + ------- + dict[str, str] + Dictionary mapping each input column name to its resolved name. + """ + if column_map is None: + return {col: col for col in columns} + else: + return {col: column_map.get(col, col) for col in columns} -def county_capacity_summary(state: str) -> pd.DataFrame: +def county_capacity_summary( + state: str, + county_col: str = "CountyName", + visits_col: str = "Tot_ED_NmbVsts", + stations_col: str = "EDStations", + bed_col: str = "LICENSED_BED_SIZE", +) -> pd.DataFrame: """ Aggregate emergency department capacity metrics at the county level. @@ -25,6 +54,14 @@ def county_capacity_summary(state: str) -> pd.DataFrame: ---------- state : str State name used to download emergency department data. + county_col : str + Column name for county identifier. Defaults to the raw dataset column. + visits_col : str + Column name for total ED visits. Defaults to the raw dataset column. + stations_col : str + Column name for ED stations. Defaults to the raw dataset column. + bed_col : str + Column name for licensed bed size. Defaults to the raw dataset column. Returns ------- @@ -38,47 +75,37 @@ def county_capacity_summary(state: str) -> pd.DataFrame: """ df = download_emergency_data(state).copy() # Load and isolate dataset for safe mutation - # Ensure required columns exist before processing - required_cols = [ - "county_name", - "total_ed_visits", - "ed_stations", - "licensed_bed_size", - ] - # Ensure dataset has all required fields before analysis + required_cols = [county_col, visits_col, stations_col, bed_col] missing = [col for col in required_cols if col not in df.columns] if missing: raise ValueError(f"Missing required columns: {missing}") # Convert key columns to numeric to avoid aggregation errors - df["total_ed_visits"] = pd.to_numeric(df["total_ed_visits"], errors="coerce") - df["ed_stations"] = pd.to_numeric(df["ed_stations"], errors="coerce") + df[visits_col] = pd.to_numeric(df[visits_col], errors="coerce") + df[stations_col] = pd.to_numeric(df[stations_col], errors="coerce") # Convert bed size categories (e.g., "50-99", "500+") to numeric values - df["bed_size_numeric"] = df["licensed_bed_size"].apply(_bed_size_to_numeric) + df["bed_size_numeric"] = df[bed_col].apply(_bed_size_to_numeric) # Aggregate metrics at the county level summary = ( - df.groupby("county_name", dropna=False) + df.groupby(county_col, dropna=False) .agg( - total_visits=("total_ed_visits", "sum"), - total_stations=("ed_stations", "sum"), - total_beds=("bed_size_numeric", "sum"), + tot_ed_visits=(visits_col, "sum"), + ed_stations=(stations_col, "sum"), + licensed_bed_size=("bed_size_numeric", "sum"), ) .reset_index() ) - # Calculate visits per station safely (avoid division by zero) - summary["visits_per_station"] = np.where( - summary["total_stations"] > 0, - summary["total_visits"] / summary["total_stations"], - np.nan, - ) + # Calculate visits per station as a measure of capacity burden + summary["visits_per_station"] = summary["tot_ed_visits"] / summary["ed_stations"] + summary["visits_per_station"] = summary["visits_per_station"].replace([np.inf, -np.inf], np.nan) return summary -def rank_counties_by_burden(summary: pd.DataFrame) -> pd.DataFrame: +def rank_counties_by_burden(summary: pd.DataFrame, visits_col: str = "visits_per_station") -> pd.DataFrame: """ Rank counties by emergency department burden. @@ -88,8 +115,9 @@ def rank_counties_by_burden(summary: pd.DataFrame) -> pd.DataFrame: Parameters ---------- summary : pd.DataFrame - DataFrame produced by county_capacity_summary, containing - 'visits_per_station'. + DataFrame produced by county_capacity_summary, containing visits-per-station values. + visits_col : str + Column name for visits per station. Defaults to the summary output. Returns ------- @@ -99,18 +127,17 @@ def rank_counties_by_burden(summary: pd.DataFrame) -> pd.DataFrame: Raises ------ ValueError - If 'visits_per_station' column is missing. + If the visits per station column is missing. """ - # Ensure required column exists - if "visits_per_station" not in summary.columns: - raise ValueError("summary must include 'visits_per_station' column") + if visits_col not in summary.columns: + raise ValueError(f"summary must include '{visits_col}' column") ranked = summary.copy() # Sort counties by burden (highest first) ranked = ranked.sort_values( - by="visits_per_station", + by=visits_col, ascending=False, na_position="last", ).reset_index(drop=True) @@ -179,66 +206,34 @@ def rank_hospitals_by_visits_per_station( return result -def generate_county_report(summary: pd.DataFrame, county_name: str) -> pd.DataFrame: - """ - Return a one-row county report from a county summary DataFrame. - - Parameters - ---------- - summary : pd.DataFrame - DataFrame containing county-level metrics, such as the output of - county_capacity_summary(). - county_name : str - Name of the county to report. - - Returns - ------- - pd.DataFrame - DataFrame with columns [facility_col, 'visits_per_station', 'rank'] sorted - by 'visits_per_station' descending. - """ - - if facility_col not in df.columns or visits_col not in df.columns: - missing = [c for c in (facility_col, visits_col) if c not in df.columns] - raise ValueError(f"Missing required columns: {missing}") - - if agg not in ("median", "mean"): - raise ValueError("agg must be 'median' or 'mean'") - - working = df[[facility_col, visits_col]].copy() - working[visits_col] = pd.to_numeric(working[visits_col], errors="coerce") - - # Aggregate per facility - if agg == "median": - grouped = working.groupby(facility_col, dropna=False)[visits_col].median() - else: - grouped = working.groupby(facility_col, dropna=False)[visits_col].mean() - - result = grouped.reset_index().rename(columns={visits_col: "visits_per_station"}) - - # Sort with NaNs last - result = result.sort_values(by="visits_per_station", ascending=False, na_position="last").reset_index(drop=True) - - # Add rank (1-based). Ties receive the same rank using dense ranking - result["rank"] = result["visits_per_station"].rank(method="dense", ascending=False).astype(int) - - if top_n is not None: - result = result.head(top_n).reset_index(drop=True) - - return result - - -def generate_county_report(summary: pd.DataFrame, county_name: str) -> pd.DataFrame: +def generate_county_report( + summary: pd.DataFrame, + county_name: str, + county_col: str = "county_name", + visits_col: str = "tot_ed_visits", + stations_col: str = "ed_stations", + beds_col: str = "licensed_bed_size", + visits_per_station_col: str = "visits_per_station", +) -> pd.DataFrame: """ Return a one-row county report from a county summary DataFrame. Parameters ---------- summary : pd.DataFrame - DataFrame containing county-level metrics, such as the output of - county_capacity_summary(). + DataFrame containing county-level summary metrics. county_name : str Name of the county to report. + county_col : str + Column name for county identifier in the summary table. + visits_col : str + Column name for total visits in the summary table. + stations_col : str + Column name for total stations in the summary table. + beds_col : str + Column name for licensed bed size total in the summary table. + visits_per_station_col : str + Column name for visits per station in the summary table. Returns ------- @@ -250,20 +245,13 @@ def generate_county_report(summary: pd.DataFrame, county_name: str) -> pd.DataFr ValueError If required columns are missing or the county is not found. """ - required_cols = [ - "county_name", - "total_visits", - "total_stations", - "total_beds", - "visits_per_station", - ] - # Ensure the summary has all required metrics before filtering + required_cols = [county_col, visits_col, stations_col, beds_col, visits_per_station_col] missing = [col for col in required_cols if col not in summary.columns] if missing: raise ValueError(f"summary is missing required columns: {missing}") - # Filter dataset to the requested county - report = summary[summary["county_name"] == county_name].copy() - # Validate that the county exists in the dataset + + report = summary[summary[county_col] == county_name].copy() + if report.empty: raise ValueError(f"No county found with name '{county_name}'") @@ -331,6 +319,7 @@ def find_capacity_volume_mismatch( facility_col: str = "facility_name", county_col: str = "county_name", year_col: str = "year", + column_map: dict[str, str] | None = None, high_visit_quantile: float = 0.75, low_capacity_quantile: float = 0.25, min_visits: int | None = None, @@ -469,21 +458,20 @@ def find_capacity_volume_mismatch( ).reset_index(drop=True) -def compute_capacity_pressure_score(df: pd.DataFrame) -> pd.DataFrame: +def compute_capacity_pressure_score( + df: pd.DataFrame, + *, + column_map: dict[str, str] | None = None, +) -> pd.DataFrame: """ - Computes a capacity pressure score (1–10) per facility, grouped by FacilityName2. - - Score interpretation: - 1 = Severely underutilized — low visits/station, adequate primary care, - mental health, and large bed size - 10 = Severely overutilized — high visits/station, shortage areas for - primary care and mental health, small bed size + Computes a capacity pressure score (1–10) per facility. Parameters: df: DataFrame containing the hospital ED data + column_map: Optional mapping from generic column keys to actual columns. Returns: - DataFrame with FacilityName2 and their capacity_pressure_score (1–10) + DataFrame with facility identifier and their capacity_pressure_score. """ df = clean_data(df) @@ -604,83 +592,82 @@ def find_duplicates( return duplicates -def plot_hospital_load_distribution(df: pd.DataFrame, group_col: str = 'hospital_ownership', output_dir: str = 'data', save: bool = False): +def plot_hospital_load_distribution( + df: pd.DataFrame, + group_col: str = "HospitalOwnership", + visits_col: str = "Visits_Per_Station", + output_dir: str = "data", + save: bool = False, +): """ Generates a statistical distribution plot of ED visits per station. - This function cleans the input data by removing records with missing values - in the analysis columns, calculates the mean visits per station for the + This function cleans the input data by removing records with missing values + in the analysis columns, calculates the mean visits per station for the specified grouping, and produces a boxplot to visualize data spread and outliers. Args: - df (pd.DataFrame): The Emergency Department dataset containing - 'visits_per_station' and the specified grouping column. - group_col (str, optional): The categorical column used to group the - hospitals. Defaults to 'hospital_ownership'. + df (pd.DataFrame): The Emergency Department dataset containing + visits per station and the specified grouping column. + group_col (str, optional): The categorical column used to group the + hospitals. Defaults to the raw dataset hospital ownership column. + visits_col (str, optional): The column for visits per station. + Defaults to the raw dataset visits-per-station column. output_dir (str, optional): Directory to save output files. Defaults to 'data'. + save (bool, optional): Whether to save the plot. Defaults to False. Returns: tuple: A tuple containing: - clean_df (pd.DataFrame): The filtered DataFrame used for the plot. - avg_load (pd.Series): The calculated mean values sorted descending. - + Raises: - KeyError: If 'visits_per_station' or group_col are missing from the DataFrame. - Prepares and cleans emergency department data for load distribution analysis. + ValueError: If visits_col or group_col are missing from the DataFrame. """ + if group_col not in df.columns or visits_col not in df.columns: + missing = [col for col in (group_col, visits_col) if col not in df.columns] + raise ValueError(f"Missing required columns: {missing}") + # 1. Data Cleaning: Remove rows where essential metrics or grouping labels are missing. - # Using .copy() ensures we don't accidentally modify the original source DataFrame. - clean_df = df.dropna(subset=['visits_per_station', group_col]).copy() - - # 2. Validation: Check if the resulting dataset is empty. - # This prevents the program from crashing during plotting if no valid data exists. + clean_df = df.dropna(subset=[visits_col, group_col]).copy() + if clean_df.empty: print(f"Warning: No valid data available for {group_col}.") - return None - - # 3. Numerical Computing: Aggregate data to find the average visit burden per category. - # Sorting descending provides an immediate insight into which categories have the highest load. - avg_load = clean_df.groupby(group_col)['visits_per_station'].mean().sort_values(ascending=False) - - print(f"\n--- Statistical Summary: Mean Visits per Station by {group_col} ---") + return None, None + + avg_load = clean_df.groupby(group_col)[visits_col].mean().sort_values(ascending=False) + + print(f"\n--- Statistical Summary: Mean {visits_col} by {group_col} ---") print(avg_load.head()) - - # 4. Visualization: Initialize a figure and generate a Seaborn boxplot. - # Boxplots are chosen over simple bar charts because they visualize the full distribution, - # including the median, quartiles, and outliers within each hospital category. + fig = plt.figure(figsize=(12, 6)) - sns.boxplot(data=clean_df, x=group_col, y='visits_per_station', palette="viridis") - - # 5. Aesthetic Polishing: Set titles, labels, and rotate x-axis text for readability. - # Tight_layout is used to ensure labels do not get cut off when the image is saved. - plt.title(f'Distribution of ED Visits per Station by {group_col}') + sns.boxplot(data=clean_df, x=group_col, y=visits_col, palette="viridis") + + plt.title(f'Distribution of {visits_col} by {group_col}') plt.xticks(rotation=45) - plt.ylabel('Visits per Station') + plt.ylabel(visits_col.replace('_', ' ').title()) plt.tight_layout() - + if save: - # 6. File I/O & Error Handling: Construct path and save the image safely. - # Using Path objects handles slashes correctly across different operating systems. output_path = Path(output_dir) / f"load_distribution_{group_col}.png" try: - # Ensure the target directory exists (mkdir) before attempting to write the file. output_path.parent.mkdir(parents=True, exist_ok=True) plt.savefig(output_path) - plt.close(fig) # Explicitly close the figure to free up system memory after the file is saved. + plt.close(fig) print(f"\nSuccess: Distribution plot saved to {output_path}") except PermissionError as e: - # Handle cases where the data folder is locked or read-only. print(f"Error: Permission denied when creating directory or saving file: {e}") plt.close(fig) raise except Exception as e: - # Catch-all for other I/O issues (e.g., disk full) to provide a clear error message. print(f"Error: Failed to save plot: {e}") plt.close(fig) raise else: plt.close(fig) + return clean_df, avg_load + def year_range(csv_file:str)->tuple[int,int]: # Load the dataset from the provided CSV file path df=pd.read_csv(csv_file) @@ -692,61 +679,63 @@ def year_range(csv_file:str)->tuple[int,int]: # Return the minimum and maximum year values as integers return int(df["year"].min()),int(df["year"].max()) -def plot_facility_trend(df: pd.DataFrame, facility_id: str): +def plot_facility_trend( + df: pd.DataFrame, + facility_id: str, + facility_col: str = "FacilityName2", + year_col: str = "year", + visits_col: str = "Tot_ED_NmbVsts", +): """ Plots a time series of ED visits over time for a single facility. - This function filters the dataset for a specified facility, cleans and converts - relevant columns to numeric values, and generates a line plot showing total - ED visits over time. - Parameters ---------- df : pd.DataFrame - DataFrame containing at least 'FacilityName2', 'year', and - 'Tot_ED_NmbVsts' columns. + DataFrame containing at least the facility, year, and visit count columns. facility_id : str - Name of the facility to plot (must match values in 'FacilityName2'). - + Name of the facility to plot. + facility_col : str + Column name identifying facilities. Defaults to the raw dataset facility name. + year_col : str + Column name for year values. Defaults to 'year'. + visits_col : str + Column name for ED visit totals. Defaults to the raw dataset visits column. + Returns ------- matplotlib.figure.Figure - A matplotlib Figure object containing the facility trend plot - for the specified facility. + A matplotlib Figure object containing the facility trend plot. Raises - ------------ + ------ ValueError - If required columns are missing, the facility is not found, - or no valid numeric data is available. + If required columns are missing, the facility is not found, + or no valid numeric data is available. """ - #Check for required columns - required_cols = ['FacilityName2', 'year', 'Tot_ED_NmbVsts'] + required_cols = [facility_col, year_col, visits_col] missing = [col for col in required_cols if col not in df.columns] if missing: raise ValueError(f"Missing required columns: {missing}") - facility_df = df[df['FacilityName2'] == facility_id].copy() - # Check if facility exists in the dataset after filtering + facility_df = df[df[facility_col] == facility_id].copy() if facility_df.empty: raise ValueError(f"No data found for facility '{facility_id}'") - facility_df['year'] = pd.to_numeric(facility_df['year'], errors='coerce') - facility_df['Tot_ED_NmbVsts'] = pd.to_numeric( - facility_df['Tot_ED_NmbVsts'], errors='coerce') - - facility_df = facility_df.dropna(subset=['year', 'Tot_ED_NmbVsts']) + facility_df[year_col] = pd.to_numeric(facility_df[year_col], errors='coerce') + facility_df[visits_col] = pd.to_numeric(facility_df[visits_col], errors='coerce') + facility_df = facility_df.dropna(subset=[year_col, visits_col]) if facility_df.empty: raise ValueError(f"No valid numeric data for facility '{facility_id}'") - facility_df = facility_df.sort_values('year') - # create line plot of total ED visits over time for the specified facility + facility_df = facility_df.sort_values(year_col) + plt.figure(figsize=(10, 6)) sns.lineplot( data=facility_df, - x='year', - y='Tot_ED_NmbVsts', + x=year_col, + y=visits_col, marker='o' ) @@ -757,47 +746,52 @@ def plot_facility_trend(df: pd.DataFrame, facility_id: str): return plt.gcf() -import pandas as pd -def per_category_burden_report(df, top_n=3): +def per_category_burden_report( + df: pd.DataFrame, + top_n: int = 3, + facility_col: str = "FacilityName2", + category_col: str = "Category", + visits_col: str = "Visits_Per_Station", +): """ Generates a per-category burden report for facilities. Parameters: - df (pd.DataFrame): Dataset containing at least 'FacilityName2', 'Category', 'Visits_Per_Station' - top_n (int): Number of top facilities to report per category (default 3) + df (pd.DataFrame): Dataset containing at least facility name, category, and visits-per-station columns. + top_n (int): Number of top facilities to report per category. + facility_col (str): Column name for facility identifier. + category_col (str): Column name for facility category. + visits_col (str): Column name for visits per station. Returns: - dict: Dictionary with categories as keys and a list of top facility names as values + dict: Dictionary with categories as keys and a list of top facility names as values. """ - # Ensure required columns exist - required_cols = ["FacilityName2", "Category", "Visits_Per_Station"] + required_cols = [facility_col, category_col, visits_col] missing_cols = [col for col in required_cols if col not in df.columns] if missing_cols: raise KeyError(f"Missing required columns: {missing_cols}") report = {} - - # Get unique categories - categories = df["Category"].unique() - + categories = df[category_col].unique() + for category in categories: - # Filter data for this category - cat_df = df[df["Category"] == category] - - # Sort by Visits_Per_Station descending - cat_df = cat_df.sort_values(by="Visits_Per_Station", ascending=False) - - # Select top_n facilities - top_facilities = cat_df["FacilityName2"].head(top_n).tolist() - - # Add to report + cat_df = df[df[category_col] == category] + cat_df = cat_df.sort_values(by=visits_col, ascending=False) + top_facilities = cat_df[facility_col].head(top_n).tolist() report[category] = top_facilities - + return report -def run_er_analysis(df, hospital_name=None): +def run_er_analysis( + df, + hospital_name=None, + facility_col: str = "facility_name", + year_col: str = "year", + visits_col: str = "total_ed_visits", + visits_per_station_col: str = "visits_per_station", +): """ ER analysis: - Compute year-over-year (YoY) changes @@ -805,49 +799,56 @@ def run_er_analysis(df, hospital_name=None): - Detect mismatches between demand and capacity - Generate simple visualizations """ - + required_cols = [facility_col, year_col, visits_col, visits_per_station_col] + missing = [col for col in required_cols if col not in df.columns] + if missing: + raise ValueError(f"Missing required columns: {missing}") - df = df.sort_values(["oshpd_id", "year"]).copy() + df = clean_data(df) + df = df.sort_values([facility_col, year_col]).copy() - df["YoY_Visits"] = df.groupby("oshpd_id")["Tot_ED_NmbVsts"].pct_change() + df[visits_col] = pd.to_numeric(df[visits_col], errors="coerce") + df[visits_per_station_col] = pd.to_numeric(df[visits_per_station_col], errors="coerce") - df["Utilization"] = df["Visits_Per_Station"] + df["YoY_Visits"] = ( + df.groupby(facility_col)[visits_col] + .transform(lambda x: x.ffill().pct_change()) + ) - df["Utilization_change"] = df.groupby("oshpd_id")["Utilization"].pct_change(fill_method=None) + df["Utilization"] = df[visits_per_station_col] + df["Utilization_change"] = ( + df.groupby(facility_col)["Utilization"] + .transform(lambda x: x.ffill().pct_change()) + ) df["Mismatch"] = ( (df["YoY_Visits"] > 0) & (df["Utilization_change"] <= 0) ) - # --- Visualization 1: Capacity vs Demand --- fig1 = plt.figure() - plt.scatter(df["visits_per_station"], df["total_ed_visits"]) + plt.scatter(df[visits_per_station_col], df[visits_col]) plt.xlabel("Capacity (Visits per Station)") plt.ylabel("Demand (Total Visits)") plt.title("Capacity vs Demand") plt.tight_layout() plt.show() - plt.close(fig1) # Close figure to prevent memory leak + plt.close(fig1) - # --- Visualization 2: Specific hospital trend --- if hospital_name: - data = df[df["facility_name"] == hospital_name] - + data = df[df[facility_col] == hospital_name] if not data.empty: fig2 = plt.figure() - plt.plot(data["year"], data["total_ed_visits"], marker="o") + plt.plot(data[year_col], data[visits_col], marker="o") plt.title(f"ER Visits Trend - {hospital_name}") plt.xlabel("Year") plt.ylabel("Visits") plt.tight_layout() plt.show() - plt.close(fig2) # Close figure to prevent memory leak + plt.close(fig2) else: print(f"[Warning] No data found for hospital: {hospital_name}") - # --- Visualization 3: Average YoY trend --- - yoy = df.groupby("year")["YoY_Visits"].mean() - + yoy = df.groupby(year_col)["YoY_Visits"].mean() fig3 = plt.figure() yoy.plot(marker="o") plt.title("Average Year-over-Year Change in ER Visits") @@ -855,7 +856,7 @@ def run_er_analysis(df, hospital_name=None): plt.ylabel("YoY Change") plt.tight_layout() plt.show() - plt.close(fig3) # Close figure to prevent memory leak + plt.close(fig3) return df @@ -866,78 +867,69 @@ def run_er_analysis(df, hospital_name=None): import pandas as pd # Urban vs rural disparity dashboard -def plot_urban_rural_map(state: str, save: bool = False) -> folium.Map: +def plot_urban_rural_map( + state: str, + save: bool = False, + latitude_col: str = "LATITUDE", + longitude_col: str = "LONGITUDE", + designation_col: str = "UrbanRuralDesi", + facility_col: str = "FacilityName2", +) -> folium.Map: """Downloads emergency data for a given state and displays hospital locations on an interactive map. Duplicate coordinates are merged to prevent overlapping issues on the map. """ - # Download the dataset print(f"Loading/Downloading dataset for state: {state}...") df = download_emergency_data(state) - # Check if required columns exist in the downloaded dataset - required_cols = ["latitude", "longitude", "urban_rural_designation", "facility_name"] + required_cols = [latitude_col, longitude_col, designation_col, facility_col] missing = [col for col in required_cols if col not in df.columns] if missing: raise ValueError( f"Downloaded dataset is missing required columns for mapping: {missing}" ) - # Drop rows where coordinates are missing - map_data = df.dropna(subset=["latitude", "longitude"]).copy() - - # Convert coordinates to numeric, handling errors - map_data["latitude"] = pd.to_numeric(map_data["latitude"], errors="coerce") - map_data["longitude"] = pd.to_numeric( - map_data["longitude"], errors="coerce" - ) - map_data = map_data.dropna(subset=["latitude", "longitude"]) + map_data = df.dropna(subset=[latitude_col, longitude_col]).copy() + map_data[latitude_col] = pd.to_numeric(map_data[latitude_col], errors="coerce") + map_data[longitude_col] = pd.to_numeric(map_data[longitude_col], errors="coerce") + map_data = map_data.dropna(subset=[latitude_col, longitude_col]) print(f"Total raw hospital records: {len(map_data)}") - - # Group by coordinates and combine hospital names and area types + map_data = ( - map_data.groupby(["latitude", "longitude"]) + map_data.groupby([latitude_col, longitude_col]) .agg( { - "facility_name": lambda x: "
".join( - x.dropna().astype(str).unique() - ), - "urban_rural_designation": "first", + facility_col: lambda x: "
".join(x.dropna().astype(str).unique()), + designation_col: "first", } ) .reset_index() ) total_unique_locations = len(map_data) - print( - f"Data processing complete. Found {total_unique_locations} unique hospital locations." - ) + print(f"Data processing complete. Found {total_unique_locations} unique hospital locations.") if total_unique_locations == 0: print("Warning: No valid hospital coordinates found to plot.") return None - # Initialize map at the mean center of all hospitals - center_lat = map_data["latitude"].mean() - center_lon = map_data["longitude"].mean() + center_lat = map_data[latitude_col].mean() + center_lon = map_data[longitude_col].mean() m = folium.Map(location=[center_lat, center_lon], zoom_start=7) marker_cluster = MarkerCluster( spiderfyOnMaxZoom=False, showCoverageOnHover=False, - disableClusteringAtZoom=9, + disableClusteringAtZoom=9, ).add_to(m) - # Iterate through each unique location and add colored markers for _, row in map_data.iterrows(): - hospital_names = row["facility_name"] - area_type = str(row["urban_rural_designation"]).strip().lower() - - # Assign colors and icons based on Urban/Rural status + hospital_names = row[facility_col] + area_type = str(row[designation_col]).strip().lower() if "urban" in area_type: marker_color = "blue" @@ -950,13 +942,15 @@ def plot_urban_rural_map(state: str, save: bool = False) -> folium.Map: marker_icon = "info-sign" folium.Marker( - location=[row["latitude"], row["longitude"]], - popup=f"Hospital(s):
{hospital_names}
Type: {row['urban_rural_designation']}", + location=[row[latitude_col], row[longitude_col]], + popup=( + f"Hospital(s):
{hospital_names}" + f"
Type: {row[designation_col]}" + ), icon=folium.Icon(color=marker_color, icon=marker_icon), ).add_to(marker_cluster) - + if save: - # Save the map to an HTML file output_dir = Path("data") try: output_dir.mkdir(parents=True, exist_ok=True) @@ -973,21 +967,46 @@ def plot_urban_rural_map(state: str, save: bool = False) -> folium.Map: return m -def summarize_by_ownership(df, - ownership_type="HospitalOwnership", - total_visits="Tot_ED_NmbVsts", - stations="EDStations", - visits_perstation="Visits_Per_Station"): +def mental_health_shortage_analysis( + df, + visits_col: str = "Tot_ED_NmbVsts", + stations_col: str = "EDStations", + mental_health_col: str = "MentalHealthShortageArea", +): + df = df.copy() + + required_cols = [visits_col, stations_col, mental_health_col] + missing = [col for col in required_cols if col not in df.columns] + if missing: + raise ValueError(f"Missing required columns: {missing}") + + df[visits_col] = pd.to_numeric(df[visits_col], errors="coerce") + df[stations_col] = pd.to_numeric(df[stations_col], errors="coerce") + + df[stations_col] = df[stations_col].replace(0, 0.0001) + df["burden_score"] = df[visits_col] / df[stations_col] + + avg_burden = df["burden_score"].mean() + df["high_risk"] = ( + (df[mental_health_col] == "Yes") & + (df["burden_score"] > avg_burden) + ) + + return df + + +def summarize_by_ownership(df, column_map: dict[str, str] | None = None): """ group hospitals by ownership type and compute summary statistics for burden, volume, & capacity insight group hospitals by ownership type and compute summary statistics for burden, volume, & capacity insights - - parameters needed: - ownership_type: column name representing ownership categories (e.g. nonprofit, government, private, etc) - total_visits: column name representing total number of emergency department encounters for the facility - stations: column name representing the number of emergency department treatment stations - visits_perstation: column name representing number of visits per station in a facility - """ + """ + cols = _resolve_columns(column_map, ['hospital_ownership', 'tot_ed_visits', 'ed_stations', 'visits_per_station']) + + ownership_type = cols['hospital_ownership'] + total_visits = cols['tot_ed_visits'] + stations = cols['ed_stations'] + visits_perstation = cols['visits_per_station'] + #ensure all required columns exist, raise column specific error if not required = [ownership_type, total_visits, stations, visits_perstation] missing = [col for col in required if col not in df.columns] @@ -1024,27 +1043,25 @@ def summarize_by_ownership(df, -def clean_growth(df): - df = df.copy() - - - # clear NAs - df = df.dropna(subset=["Tot_ED_NmbVsts", "year"]) - - - # types - df["year"] = df["year"].astype(int) - df["Tot_ED_NmbVsts"] = pd.to_numeric(df["Tot_ED_NmbVsts"], errors="coerce") +def clean_growth(df, column_map: dict[str, str] | None = None): + cols = _resolve_columns(column_map, ['tot_ed_visits', 'year', 'facility_name']) + + df = df.copy() + # clear NAs + df = df.dropna(subset=[cols['tot_ed_visits'], cols['year']]) - df = df.sort_values(by=["FacilityName2", "year"]) + # types + df[cols['year']] = df[cols['year']].astype(int) + df[cols['tot_ed_visits']] = pd.to_numeric(df[cols['tot_ed_visits']], errors="coerce") + df = df.sort_values(by=[cols['facility_name'], cols['year']]) - return df + return df -def calculate_growth(df, value_col, group_cols, time_col="year", pct=True): +def calculate_growth(df, value_col=None, group_cols=None, time_col=None, pct=True, column_map: dict[str, str] | None = None): """ Parameters: - df: Data Frame @@ -1052,36 +1069,47 @@ def calculate_growth(df, value_col, group_cols, time_col="year", pct=True): - group_cols: list of columns to group by ('oshpd_id') - time_col: time ('year') - pct: if True, returns percent growth; else raw difference + - column_map: optional mapping for column names """ - + cols = _resolve_columns(column_map, ['tot_ed_visits', 'facility_name', 'year']) + + if value_col is None: + value_col = cols['tot_ed_visits'] + if group_cols is None: + group_cols = [cols['facility_name']] + if time_col is None: + time_col = cols['year'] df = df.copy() - # Sort df = df.sort_values(by=group_cols + [time_col]) - # Calculate previous value df["prev_value"] = df.groupby(group_cols)[value_col].shift(1) - # Growth calculation if pct: df["growth"] = (df[value_col] - df["prev_value"]) / df["prev_value"] * 100 else: df["growth"] = df[value_col] - df["prev_value"] - return df def county_facility_counts( df: pd.DataFrame, - county_col: str = "CountyName", - facility_col: str = "FacilityName2" + county_col: str = None, + facility_col: str = None, + column_map: dict[str, str] | None = None ) -> pd.DataFrame: - # Ensure required columns exist before processing + cols = _resolve_columns(column_map, ['county_name', 'facility_name']) + + if county_col is None: + county_col = cols['county_name'] + if facility_col is None: + facility_col = cols['facility_name'] + required = [county_col, facility_col] missing = [col for col in required if col not in df.columns] if missing: @@ -1107,9 +1135,10 @@ def county_facility_counts( def spike_frequency_pivot( df: pd.DataFrame, threshold_pct: float = 20.0, - facility_col: str = 'FacilityName2', - category_col: str = 'Category', - visits_col: str = 'Visits_Per_Station' + facility_col: str = None, + category_col: str = None, + visits_col: str = None, + column_map: dict[str, str] | None = None ) -> pd.DataFrame: """ Build a pivot table of spike frequency aggregated by visit category. @@ -1120,11 +1149,20 @@ def spike_frequency_pivot( how often each category experiences high-growth periods across the entire dataset. """ + cols = _resolve_columns(column_map, ['facility_name', 'category', 'visits_per_station', 'year']) + + if facility_col is None: + facility_col = cols['facility_name'] + if category_col is None: + category_col = cols['category'] + if visits_col is None: + visits_col = cols['visits_per_station'] + year_col = cols['year'] df = df.copy() df['yoy_pct_change'] = ( - df.sort_values('year') + df.sort_values(year_col) .groupby([facility_col, category_col])[visits_col] .pct_change(fill_method=None) * 100 ) diff --git a/tests/test_stats.py b/tests/test_stats.py index 09b8455..28b8762 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -80,10 +80,10 @@ def test_rank_counties_by_burden(): def test_county_capacity_summary(monkeypatch): fake_df = pd.DataFrame( { - "county_name": ["Alameda", "Alameda", "Fresno"], - "total_ed_visits": [100, 200, 90], - "ed_stations": [10, 20, 0], - "licensed_bed_size": ["1-49", "50-99", "100-199"], + "CountyName": ["Alameda", "Alameda", "Fresno"], + "Tot_ED_NmbVsts": [100, 200, 90], + "EDStations": [10, 20, 0], + "LICENSED_BED_SIZE": ["1-49", "50-99", "100-199"], } ) @@ -94,16 +94,16 @@ def fake_download(state): result = stats.county_capacity_summary("California") - alameda = result[result["county_name"] == "Alameda"].iloc[0] - fresno = result[result["county_name"] == "Fresno"].iloc[0] + alameda = result[result["CountyName"] == "Alameda"].iloc[0] + fresno = result[result["CountyName"] == "Fresno"].iloc[0] - assert alameda["total_visits"] == 300 - assert alameda["total_stations"] == 30 - assert alameda["total_beds"] == 25.0 + 74.5 + assert alameda["tot_ed_visits"] == 300 + assert alameda["ed_stations"] == 30 + assert alameda["licensed_bed_size"] == 25.0 + 74.5 assert alameda["visits_per_station"] == 10 - assert fresno["total_visits"] == 90 - assert fresno["total_stations"] == 0 + assert fresno["tot_ed_visits"] == 90 + assert fresno["ed_stations"] == 0 assert np.isnan(fresno["visits_per_station"]) @@ -383,9 +383,9 @@ def test_generate_county_report_basic(): summary = pd.DataFrame( { "county_name": ["Autauga", "Baldwin"], - "total_visits": [1000, 2000], - "total_stations": [10, 20], - "total_beds": [75.0, 125.0], + "tot_ed_visits": [1000, 2000], + "ed_stations": [10, 20], + "licensed_bed_size": [75.0, 125.0], "visits_per_station": [100.0, 100.0], } ) @@ -393,15 +393,15 @@ def test_generate_county_report_basic(): assert result.shape == (1, 5) assert result.loc[0, "county_name"] == "Autauga" - assert result.loc[0, "total_visits"] == 1000 + assert result.loc[0, "tot_ed_visits"] == 1000 def test_generate_county_report_missing_county(): summary = pd.DataFrame( { "county_name": ["Autauga", "Baldwin"], - "total_visits": [1000, 2000], - "total_stations": [10, 20], - "total_beds": [75.0, 125.0], + "tot_ed_visits": [1000, 2000], + "ed_stations": [10, 20], + "licensed_bed_size": [75.0, 125.0], "visits_per_station": [100.0, 100.0], } ) @@ -479,10 +479,10 @@ def test_plot_urban_rural_map_runs(monkeypatch): """ fake_df = pd.DataFrame({ - "latitude": [34.1, 35.2], - "longitude": [-118.2, -119.3], - "urban_rural_designation": ["Urban", "Rural"], - "facility_name": ["Hospital A", "Hospital B"] + "LATITUDE": [34.1, 35.2], + "LONGITUDE": [-118.2, -119.3], + "UrbanRuralDesi": ["Urban", "Rural"], + "FacilityName2": ["Hospital A", "Hospital B"] }) def fake_download(state): @@ -696,7 +696,7 @@ def test_multiple_facilities_spikes_summed(): {'FacilityName2': 'Hospital B', 'year': 2021, 'Visits_Per_Station': 100}, {'FacilityName2': 'Hospital B', 'year': 2022, 'Visits_Per_Station': 200}, ]) - assert spike_frequency_pivot(df).loc['All ED Visits', 'spike_count'] == 2 + assert spike_frequency_pivot(df, facility_col='FacilityName2', category_col='Category', visits_col='Visits_Per_Station').loc['All ED Visits', 'spike_count'] == 2 def test_smoke_real_data(): """End-to-end smoke test: no NaN spike counts on live downloaded California data.""" @@ -719,7 +719,7 @@ def test_all_categories_present(): {'Category': 'Mental Health', 'year': 2021, 'Visits_Per_Station': 100}, {'Category': 'Mental Health', 'year': 2022, 'Visits_Per_Station': 105}, ]) - result = spike_frequency_pivot(df) + result = spike_frequency_pivot(df, facility_col='FacilityName2', category_col='Category', visits_col='Visits_Per_Station') assert set(result.index) == {'Diabetes', 'Mental Health'} def test_spike_counted(): @@ -728,7 +728,7 @@ def test_spike_counted(): {'year': 2021, 'Visits_Per_Station': 100}, {'year': 2022, 'Visits_Per_Station': 200}, ]) - assert spike_frequency_pivot(df, threshold_pct=20.0).loc['All ED Visits', 'spike_count'] == 1 + assert spike_frequency_pivot(df, threshold_pct=20.0, facility_col='FacilityName2', category_col='Category', visits_col='Visits_Per_Station').loc['All ED Visits', 'spike_count'] == 1 #pytests for summarize_by_ownership function from ertimes.stats import summarize_by_ownership @@ -832,9 +832,9 @@ def fake_download(): return pd.DataFrame({ "oshpd_id": [1, 1, 2, 2], "year": [2021, 2022, 2021, 2022], - "Tot_ED_NmbVsts": [100, 120, 200, 210], - "Visits_Per_Station": [10, 12, 20, 21], - "FacilityName2": ["A", "A", "B", "B"] + "total_ed_visits": [100, 120, 200, 210], + "visits_per_station": [10, 12, 20, 21], + "facility_name": ["A", "A", "B", "B"] })