# healthcare_analysis.py import pandas as pd import numpy as np from typing import Dict, List, Any, Optional, Tuple import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class HealthcareAnalyzer: def __init__(self, data_registry): self.data_registry = data_registry self.analysis_results = {} self.scenario_text = "" # Store scenario text for context def comprehensive_analysis(self, scenario_text: str) -> Dict[str, Any]: """Perform comprehensive healthcare scenario analysis""" logger.info("Starting comprehensive healthcare analysis") # Store scenario text for use in other methods self.scenario_text = scenario_text # Extract tasks and requirements tasks = self._extract_tasks(scenario_text) requirements = self._extract_requirements(scenario_text) # Identify relevant datasets relevant_data = self._identify_relevant_data(scenario_text) # Perform analyses based on tasks results = {} if "facility_distribution" in tasks: results["facility_distribution"] = self.analyze_facility_distribution(relevant_data, requirements) if "capacity_analysis" in tasks: results["capacity_analysis"] = self.analyze_capacity(relevant_data, requirements) if "resource_allocation" in tasks: results["resource_allocation"] = self.analyze_resource_allocation(relevant_data) if "trends" in tasks: results["trends"] = self.analyze_trends(relevant_data) # Generate recommendations results["recommendations"] = self.generate_recommendations(results, requirements) # Future integration opportunities results["future_integration"] = self.identify_integration_opportunities(results) logger.info("Comprehensive analysis completed") return results def _extract_tasks(self, scenario_text: str) -> List[str]: """Extract specific tasks from scenario text""" tasks = [] task_keywords = { "facility_distribution": ["facility", "distribution", "location", "sites"], "capacity_analysis": ["capacity", "beds", "occupancy", "utilization"], "resource_allocation": ["resource", "allocation", "staffing", "equipment"], "trends": ["trend", "change", "growth", "decline", "pattern"] } for task_type, keywords in task_keywords.items(): if any(kw in scenario_text.lower() for kw in keywords): tasks.append(task_type) return tasks def _extract_requirements(self, scenario_text: str) -> Dict[str, Any]: """Extract specific requirements from scenario text""" return { "geographic_scope": self._extract_geographic_scope(scenario_text), "time_period": self._extract_time_period(scenario_text), "facility_types": self._extract_facility_types(scenario_text), "metrics_needed": self._extract_metrics(scenario_text), "regions": self._extract_regions(scenario_text) } def _extract_regions(self, scenario_text: str) -> List[str]: """Extract specific regions mentioned in the scenario""" # Look for region names in the scenario regions = [] # Common region patterns - this could be expanded region_patterns = [ r'([A-Z][a-z]+ (Zone|Region|Area|District))', r'(North|South|East|West|Central|Calgary|Edmonton|Toronto|Vancouver|Montreal)', r'(Alberta|British Columbia|Ontario|Quebec|Manitoba|Saskatchewan|Nova Scotia|New Brunswick|PEI|Newfoundland|Yukon|NWT|Nunavut)' ] import re for pattern in region_patterns: matches = re.findall(pattern, scenario_text) for match in matches: if isinstance(match, tuple): regions.append(match[0]) else: regions.append(match) # Remove duplicates while preserving order seen = set() unique_regions = [r for r in regions if not (r in seen or seen.add(r))] return unique_regions def analyze_facility_distribution(self, relevant_data: List[str], requirements: Dict[str, Any]) -> Dict[str, Any]: """Enhanced facility distribution analysis""" results = {} geographic_scope = requirements.get("geographic_scope", "Unknown") regions = requirements.get("regions", []) for data_name in relevant_data: df = self.data_registry.get(data_name) if df is None or df.empty: continue # Filter data based on geographic scope filtered_df = self._filter_by_geography(df, geographic_scope, regions) if filtered_df.empty: continue # Facility type distribution type_col = self._find_column(filtered_df, ['type', 'category', 'class', 'facility_type', 'odhf_facility_type']) if type_col: # Ensure we're working with string data filtered_df[type_col] = filtered_df[type_col].astype(str) type_dist = filtered_df[type_col].value_counts().to_dict() results["facility_type_distribution"] = type_dist # Calculate diversity index diversity = self._calculate_diversity_index(type_dist) results["facility_diversity"] = diversity # Geographic distribution geo_col = self._find_column(filtered_df, ['province', 'state', 'region', 'zone', 'area']) if geo_col: # Ensure we're working with string data filtered_df[geo_col] = filtered_df[geo_col].astype(str) geo_dist = filtered_df[geo_col].value_counts().to_dict() results["geographic_distribution"] = geo_dist # Calculate Gini coefficient for inequality gini = self._calculate_gini(list(geo_dist.values())) results["geographic_inequality"] = gini # City distribution city_col = self._find_column(filtered_df, ['city', 'municipality', 'town']) if city_col: # Ensure we're working with string data filtered_df[city_col] = filtered_df[city_col].astype(str) city_counts = filtered_df[city_col].value_counts().head(5) top_cities = city_counts.index.tolist() # Breakdown by facility type for top cities city_breakdown = {} for city in top_cities: city_data = filtered_df[filtered_df[city_col] == city] if not city_data.empty and type_col in city_data.columns: city_breakdown[city] = city_data[type_col].value_counts().to_dict() results["top_cities"] = top_cities results["city_breakdown"] = city_breakdown # Total facilities count results["total_facilities"] = len(filtered_df) return results def analyze_capacity(self, relevant_data: List[str], requirements: Dict[str, Any]) -> Dict[str, Any]: """Enhanced capacity analysis""" results = {} geographic_scope = requirements.get("geographic_scope", "Unknown") regions = requirements.get("regions", []) for data_name in relevant_data: df = self.data_registry.get(data_name) if df is None or df.empty: continue # Filter data based on geographic scope filtered_df = self._filter_by_geography(df, geographic_scope, regions) if filtered_df.empty: continue # Current capacity capacity_col = self._find_column(filtered_df, ['capacity', 'beds', 'current_capacity', 'beds_current']) if capacity_col: # Ensure we're working with numeric data filtered_df[capacity_col] = pd.to_numeric(filtered_df[capacity_col], errors='coerce') total_capacity = filtered_df[capacity_col].sum() results["total_capacity"] = total_capacity # Capacity by facility type type_col = self._find_column(filtered_df, ['type', 'facility_type']) if type_col and type_col in filtered_df.columns: capacity_by_type = filtered_df.groupby(type_col)[capacity_col].sum().to_dict() results["capacity_by_type"] = capacity_by_type # Capacity utilization utilization_col = self._find_column(filtered_df, ['utilization', 'occupancy', 'occupancy_rate']) if utilization_col: # Ensure we're working with numeric data filtered_df[utilization_col] = pd.to_numeric(filtered_df[utilization_col], errors='coerce') avg_utilization = filtered_df[utilization_col].mean() results["average_utilization"] = avg_utilization # Utilization by facility type if type_col and type_col in filtered_df.columns: utilization_by_type = filtered_df.groupby(type_col)[utilization_col].mean().to_dict() results["utilization_by_type"] = utilization_by_type # Capacity trends time_cols = [col for col in filtered_df.columns if any(year in col.lower() for year in ['2020', '2021', '2022', '2023', '2024'])] if len(time_cols) >= 2: trend_data = {} for col in time_cols: # Ensure we're working with numeric data filtered_df[col] = pd.to_numeric(filtered_df[col], errors='coerce') trend_data[col] = filtered_df[col].sum() results["capacity_trends"] = trend_data # Calculate growth rate if len(time_cols) >= 2: latest = time_cols[-1] earliest = time_cols[0] if trend_data[earliest] > 0: # Avoid division by zero growth_rate = (trend_data[latest] - trend_data[earliest]) / trend_data[earliest] * 100 results["capacity_growth_rate"] = growth_rate # Bed change analysis prev_col = self._find_column(filtered_df, ['prev', 'previous', '2022', 'beds_prev', 'previous_beds']) current_col = self._find_column(filtered_df, ['current', '2023', '2024', 'beds_current', 'staffed_beds', 'capacity']) if prev_col and current_col: # Ensure we're working with numeric data filtered_df[prev_col] = pd.to_numeric(filtered_df[prev_col], errors='coerce') filtered_df[current_col] = pd.to_numeric(filtered_df[current_col], errors='coerce') # Calculate bed change filtered_df['bed_change'] = filtered_df[current_col] - filtered_df[prev_col] # Calculate percentage change filtered_df['percent_change'] = filtered_df.apply( lambda row: (row['bed_change'] / row[prev_col] * 100) if row[prev_col] != 0 else 0, axis=1 ) # Zone/Region-level analysis zone_col = self._find_column(filtered_df, ['zone', 'region', 'area', 'district']) if zone_col: # Ensure we're working with string data filtered_df[zone_col] = filtered_df[zone_col].astype(str) zone_summary = filtered_df.groupby(zone_col).agg({ current_col: 'sum', prev_col: 'sum', 'bed_change': 'sum' }).reset_index() zone_summary['percent_change'] = zone_summary.apply( lambda row: (row['bed_change'] / row[prev_col] * 100) if row[prev_col] != 0 else 0, axis=1 ) results["zone_summary"] = zone_summary.to_dict('records') # Find zones with largest changes if not zone_summary.empty: # Get zone with largest absolute decrease if zone_summary['bed_change'].notna().any(): max_abs_decrease_idx = zone_summary['bed_change'].idxmin() max_abs_decrease = zone_summary.loc[max_abs_decrease_idx] results["max_absolute_decrease"] = max_abs_decrease.to_dict() # Get zone with largest percentage decrease if zone_summary['percent_change'].notna().any(): max_pct_decrease_idx = zone_summary['percent_change'].idxmin() max_pct_decrease = zone_summary.loc[max_pct_decrease_idx] results["max_percentage_decrease"] = max_pct_decrease.to_dict() # Identify facilities with largest declines facilities_decline = filtered_df.sort_values('bed_change').head(5) if not facilities_decline.empty: results["facilities_with_largest_declines"] = facilities_decline.to_dict('records') return results def _filter_by_geography(self, df: pd.DataFrame, geographic_scope: str, regions: List[str]) -> pd.DataFrame: """Filter dataframe based on geographic scope and regions""" if geographic_scope == "Unknown" and not regions: return df.copy() # Try to find a geographic column geo_col = self._find_column(df, ['province', 'state', 'region', 'zone', 'area', 'district']) if geo_col is None: return df.copy() # Ensure we're working with string data df[geo_col] = df[geo_col].astype(str) # Create filters filters = [] # Add geographic scope filter if geographic_scope != "Unknown": # Create a list of possible values for the geographic scope scope_values = [geographic_scope.lower()] # Add common abbreviations abbreviations = { "alberta": "ab", "british columbia": "bc", "ontario": "on", "quebec": "qc", "manitoba": "mb", "saskatchewan": "sk", "nova scotia": "ns", "new brunswick": "nb", "prince edward island": "pe", "newfoundland": "nl", "yukon": "yt", "northwest territories": "nt", "nunavut": "nu" } if geographic_scope.lower() in abbreviations: scope_values.append(abbreviations[geographic_scope.lower()]) scope_filter = df[geo_col].str.lower().isin(scope_values) filters.append(scope_filter) # Add region filters if regions: region_filter = df[geo_col].str.lower().isin([r.lower() for r in regions]) filters.append(region_filter) # Apply filters if filters: combined_filter = filters[0] for f in filters[1:]: combined_filter = combined_filter | f return df[combined_filter].copy() return df.copy() def analyze_resource_allocation(self, relevant_data: List[str]) -> Dict[str, Any]: """Analyze resource allocation patterns""" results = {} for data_name in relevant_data: df = self.data_registry.get(data_name) if df is None or df.empty: continue # Staff analysis staff_col = self._find_column(df, ['staff', 'employees', 'fte']) if staff_col: # Ensure we're working with numeric data df[staff_col] = pd.to_numeric(df[staff_col], errors='coerce') total_staff = df[staff_col].sum() results["total_staff"] = total_staff # Staff per bed ratio capacity_col = self._find_column(df, ['capacity', 'beds']) if capacity_col and capacity_col in df.columns: # Ensure we're working with numeric data df[capacity_col] = pd.to_numeric(df[capacity_col], errors='coerce') df['staff_per_bed'] = df[staff_col] / df[capacity_col].replace(0, np.nan) # Avoid division by zero avg_staff_per_bed = df['staff_per_bed'].mean() results["staff_per_bed_ratio"] = avg_staff_per_bed # Equipment analysis equipment_cols = [col for col in df.columns if 'equipment' in col.lower()] if equipment_cols: equipment_summary = {} for col in equipment_cols: # Ensure we're working with numeric data df[col] = pd.to_numeric(df[col], errors='coerce') equipment_summary[col] = df[col].sum() results["equipment_summary"] = equipment_summary return results def analyze_trends(self, relevant_data: List[str]) -> Dict[str, Any]: """Analyze trends in healthcare data""" results = {} for data_name in relevant_data: df = self.data_registry.get(data_name) if df is None or df.empty: continue # Find time-based columns time_cols = [col for col in df.columns if any(year in col.lower() for year in ['2020', '2021', '2022', '2023', '2024'])] if len(time_cols) >= 2: trends = {} # Calculate year-over-year changes for i in range(1, len(time_cols)): prev_year = time_cols[i-1] curr_year = time_cols[i] # Ensure we're working with numeric data df[prev_year] = pd.to_numeric(df[prev_year], errors='coerce') df[curr_year] = pd.to_numeric(df[curr_year], errors='coerce') prev_total = df[prev_year].sum() curr_total = df[curr_year].sum() if prev_total > 0: # Avoid division by zero change_pct = (curr_total - prev_total) / prev_total * 100 trends[f"{prev_year}_to_{curr_year}"] = { "absolute_change": curr_total - prev_total, "percentage_change": change_pct } results["year_over_year_trends"] = trends return results def generate_recommendations(self, analysis_results: Dict[str, Any], requirements: Dict[str, Any]) -> List[Dict[str, str]]: """Generate data-driven operational recommendations""" recommendations = [] geographic_scope = requirements.get("geographic_scope", "the region") # Capacity-related recommendations if "capacity_analysis" in analysis_results: capacity = analysis_results["capacity_analysis"] # Low utilization recommendations if "average_utilization" in capacity and capacity["average_utilization"] < 0.7: recommendations.append({ "title": "Optimize Underutilized Capacity", "description": f"Average utilization is {capacity['average_utilization']:.1%} in {geographic_scope}. Consider repurposing underutilized facilities or consolidating services.", "priority": "Medium", "data_source": "Capacity utilization analysis" }) # Capacity growth recommendations if "capacity_growth_rate" in capacity and capacity["capacity_growth_rate"] < 2: recommendations.append({ "title": "Expand Capacity Strategically", "description": f"Capacity growth rate is only {capacity['capacity_growth_rate']:.1f}% in {geographic_scope}. Invest in new facilities or expand existing ones to meet demand.", "priority": "High", "data_source": "Capacity trend analysis" }) # Zone-specific recommendations if "max_percentage_decrease" in capacity and isinstance(capacity["max_percentage_decrease"], dict): zone_col = capacity.get("columns_used", {}).get("zone") zone = capacity["max_percentage_decrease"].get(zone_col, 'a zone') if zone_col else 'a zone' decrease = capacity["max_percentage_decrease"].get("percent_change", 0) if zone and decrease: recommendations.append({ "title": f"Address Capacity Decline in {zone}", "description": f"{zone} shows a {decrease:.1f}% decrease in bed capacity. Investigate causes and implement recovery strategies.", "priority": "High", "data_source": "Zone capacity analysis" }) # Geographic distribution recommendations if "facility_distribution" in analysis_results: dist = analysis_results["facility_distribution"] if "geographic_inequality" in dist and dist["geographic_inequality"] > 0.4: recommendations.append({ "title": "Address Geographic Inequity", "description": f"High geographic inequality (Gini: {dist['geographic_inequality']:.2f}) in {geographic_scope}. Consider targeted investments in underserved areas.", "priority": "High", "data_source": "Geographic distribution analysis" }) # Resource allocation recommendations if "resource_allocation" in analysis_results: resources = analysis_results["resource_allocation"] if "staff_per_bed_ratio" in resources and resources["staff_per_bed_ratio"] < 1.5: recommendations.append({ "title": "Increase Staffing Levels", "description": f"Staff per bed ratio is {resources['staff_per_bed_ratio']:.2f} in {geographic_scope}, which may be insufficient. Consider hiring additional staff.", "priority": "High", "data_source": "Resource allocation analysis" }) # Sort by priority priority_order = {"High": 0, "Medium": 1, "Low": 2} recommendations.sort(key=lambda x: priority_order.get(x["priority"], 3)) return recommendations def identify_integration_opportunities(self, analysis_results: Dict[str, Any]) -> Dict[str, Any]: """Identify opportunities for AI integration and data enhancement""" opportunities = { "data_integration": [], "ai_applications": [], "enhanced_metrics": [] } # Data integration opportunities opportunities["data_integration"].append({ "opportunity": "Integrate real-time occupancy data", "description": "Combine current facility data with real-time occupancy monitoring systems", "benefit": "Enable dynamic resource allocation and surge planning" }) opportunities["data_integration"].append({ "opportunity": "Incorporate demographic data", "description": "Add population demographics and health needs data", "benefit": "Improve demand forecasting and service planning" }) # AI application opportunities opportunities["ai_applications"].append({ "opportunity": "Predictive capacity modeling", "description": "Use ML to forecast capacity needs based on trends and external factors", "benefit": "Proactive resource planning and reduced wait times" }) opportunities["ai_applications"].append({ "opportunity": "Optimization algorithms", "description": "Implement AI for staff scheduling and resource allocation", "benefit": "Improved efficiency and reduced operational costs" }) # Enhanced metrics opportunities["enhanced_metrics"].append({ "metric": "Patient flow efficiency", "description": "Measure time from admission to discharge across facilities", "benefit": "Identify bottlenecks and improve patient experience" }) opportunities["enhanced_metrics"].append({ "metric": "Resource utilization index", "description": "Composite metric combining staff, equipment, and space utilization", "benefit": "Holistic view of operational efficiency" }) return opportunities # Helper methods def _find_column(self, df, patterns): """Find the first column matching any pattern""" if df is None or df.empty: return None for col in df.columns: if any(pattern.lower() in col.lower() for pattern in patterns): return col return None def _calculate_gini(self, values): """Calculate Gini coefficient for inequality measurement""" if not values or len(values) < 2: return 0 values = sorted(values) n = len(values) index = np.arange(1, n + 1) total = np.sum(values) if total == 0: return 0 gini = (np.sum((2 * index - n - 1) * values)) / (n * total) return gini def _calculate_diversity_index(self, distribution): """Calculate Shannon diversity index""" if not distribution: return 0 total = sum(distribution.values()) if total == 0: return 0 proportions = [count/total for count in distribution.values() if count > 0] if not proportions: return 0 return -sum(p * np.log(p) for p in proportions) def _extract_geographic_scope(self, text): """Extract geographic scope from text""" # Look for province/state names provinces = [ "alberta", "british columbia", "ontario", "quebec", "manitoba", "saskatchewan", "nova scotia", "new brunswick", "prince edward island", "newfoundland", "yukon", "northwest territories", "nunavut" ] states = [ "alabama", "alaska", "arizona", "arkansas", "california", "colorado", "connecticut", "delaware", "florida", "georgia", "hawaii", "idaho", "illinois", "indiana", "iowa", "kansas", "kentucky", "louisiana", "maine", "maryland", "massachusetts", "michigan", "minnesota", "mississippi", "missouri", "montana", "nebraska", "nevada", "new hampshire", "new jersey", "new mexico", "new york", "north carolina", "north dakota", "ohio", "oklahoma", "oregon", "pennsylvania", "rhode island", "south carolina", "south dakota", "tennessee", "texas", "utah", "vermont", "virginia", "washington", "west virginia", "wisconsin", "wyoming" ] text_lower = text.lower() # Check for provinces for province in provinces: if province in text_lower: return province.title() # Check for states for state in states: if state in text_lower: return state.title() # Check for countries if "canada" in text_lower: return "Canada" if "usa" in text_lower or "united states" in text_lower: return "United States" return "Unknown" def _extract_time_period(self, text): """Extract time period from text""" # Look for year patterns import re years = re.findall(r'\b(20\d{2})\b', text) if len(years) >= 2: return f"{min(years)}-{max(years)}" return "Unknown" def _extract_facility_types(self, text): """Extract facility types from text""" types = [] if "hospital" in text.lower(): types.append("Hospitals") if "nursing" in text.lower() or "long-term" in text.lower(): types.append("Nursing homes") if "clinic" in text.lower(): types.append("Clinics") return types def _extract_metrics(self, text): """Extract required metrics from text""" metrics = [] if "bed" in text.lower(): metrics.append("Bed capacity") if "occupancy" in text.lower(): metrics.append("Occupancy rates") if "staff" in text.lower(): metrics.append("Staffing levels") return metrics def _identify_relevant_data(self, text): """Identify relevant datasets for the scenario""" # Use data registry's find_related_datasets method keywords = ["facility", "bed", "capacity", "healthcare", "hospital"] return [item["name"] for item in self.data_registry.find_related_datasets(keywords)]