95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
|
|
"""
|
|
Cleaned version of the CCICollaborationAnalyzer script.
|
|
This script is structured and corrected for proper exception handling and visualization generation.
|
|
"""
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
import seaborn as sns
|
|
import logging
|
|
import warnings
|
|
from pathlib import Path
|
|
from sklearn.preprocessing import StandardScaler
|
|
from scipy import stats
|
|
|
|
from cci_analyzer import CCIDataAnalyzer
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
|
logger = logging.getLogger("cci_collaboration_analysis")
|
|
|
|
# Suppress pandas warnings
|
|
warnings.filterwarnings("ignore")
|
|
|
|
|
|
class CCICollaborationAnalyzer:
|
|
def __init__(self, data_path, output_path="./output/collaboration"):
|
|
self.data_path = Path(data_path)
|
|
self.output_path = Path(output_path)
|
|
self.output_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.base_analyzer = CCIDataAnalyzer(data_path, output_path=str(self.output_path))
|
|
if not self.base_analyzer.load_data():
|
|
logger.error("Failed to load data through base analyzer")
|
|
return
|
|
|
|
self.data = self.base_analyzer.data
|
|
self.collaboration_metrics = {}
|
|
self.temporal_analysis = {}
|
|
self.regional_analysis = {}
|
|
self.ev_vouchers_analysis = {}
|
|
self._separate_ev_vouchers()
|
|
|
|
def _separate_ev_vouchers(self):
|
|
if 'cci_projects' not in self.data:
|
|
logger.error("No project data available to separate EV vouchers")
|
|
return
|
|
df = self.data['cci_projects']
|
|
try:
|
|
ev_mask = ((df['agency_name'].str.contains('Air Resources Board', case=False, na=False)) &
|
|
(df['program_name'].str.contains('Low Carbon Transportation', case=False, na=False)) &
|
|
(df['sub_program_name'].str.contains('Clean Cars 4 All|CVRP|Financing Assistance',
|
|
case=False, na=False)))
|
|
self.data['ev_vouchers'] = df[ev_mask].copy()
|
|
self.data['non_ev_projects'] = df[~ev_mask].copy()
|
|
logger.info(f"Separated {len(self.data['ev_vouchers'])} EV vouchers from {len(self.data['non_ev_projects'])} other projects")
|
|
except Exception as e:
|
|
logger.error(f"Error separating EV vouchers: {e}")
|
|
|
|
def _generate_visualizations(self):
|
|
"""Generate visualizations of key findings."""
|
|
logger.info("Generating visualizations")
|
|
try:
|
|
self._plot_collaboration_impact()
|
|
self._plot_temporal_trends()
|
|
self._plot_regional_analysis()
|
|
self._plot_ev_vouchers_analysis()
|
|
self._plot_efficiency_equity_tradeoff()
|
|
logger.info(f"All visualizations completed and saved to {self.output_path}")
|
|
except Exception as e:
|
|
logger.error(f"Error generating visualizations: {e}")
|
|
|
|
# Define stubs for the required plotting methods
|
|
def _plot_collaboration_impact(self):
|
|
logger.info("Plotting collaboration impact...")
|
|
# Implementation goes here
|
|
|
|
def _plot_temporal_trends(self):
|
|
logger.info("Plotting temporal trends...")
|
|
# Implementation goes here
|
|
|
|
def _plot_regional_analysis(self):
|
|
logger.info("Plotting regional analysis...")
|
|
# Implementation goes here
|
|
|
|
def _plot_ev_vouchers_analysis(self):
|
|
logger.info("Plotting EV vouchers analysis...")
|
|
# Implementation goes here
|
|
|
|
def _plot_efficiency_equity_tradeoff(self):
|
|
logger.info("Plotting efficiency-equity tradeoff...")
|
|
# Implementation goes here
|