# application/inverter/utils.py from application.energy.queries import EnergyQueries from datetime import datetime as date from collections import defaultdict class Utilities: def __init__(self): self.queries = EnergyQueries() def get_daily_energy(self,user_id): try: result = self.queries.get_daily_energy(user_id) if not result: return { 'energyGenerated': 0, 'co2Saved': 0, 'amountSaved': 0, 'date': None } else: return { 'energyGenerated': round(result['generation_kwh'],2), 'co2Saved': round(self.calculate_co2_saved(result['generation_kwh']), 2), 'amountSaved': round(self.calculate_amount_saved(result["generation_kwh"]),2), 'date': result['report_date'].strftime("%Y-%m-%d") } except Exception as e: # log error raise e def get_monthly_energy(self,user_id,month, year): try: result = self.queries.get_monthly_energy(user_id,month, year) if not result: return { "dailyData": [], "totalEnergy": round(0, 2), "avgDaily": 0, "totalCO2Saved": round(0, 2) } else: daily_data = [] total_energy = 0.0 total_co2 = 0.0 for record in result: # Extract date and energy report_date = record.get("report_date") generation_kwh = float(record.get("generation_kwh", 0)) # Convert date to datetime if it's a string if isinstance(report_date, str): report_date = date.strptime(report_date, "%Y-%m-%d") day_num = report_date.day date_label = report_date.strftime("%b %-d") # e.g., Jan 1 co2_saved = self.calculate_co2_saved(generation_kwh) daily_data.append({ "day": day_num, "date": date_label, "energy": round(generation_kwh, 2), "co2Saved": round(co2_saved, 2) }) total_energy += generation_kwh total_co2 += co2_saved # Calculate averages days_count = len(daily_data) avg_daily = round(total_energy / days_count, 2) if days_count else 0 return { "dailyData": daily_data, "totalEnergy": round(total_energy, 2), "avgDaily": avg_daily, "totalCO2Saved": round(total_co2, 2) } except Exception as e: # log error raise e def get_annual_energy(self, user_id, year): try: result = self.queries.get_yearly_energy(user_id, year) if not result: return { "monthlyData": [], "totalEnergy": round(0, 2), "totalCO2Saved": round(0, 2), "avgEnergyPerMonth": 0, "bestMonth": {} } else: month_data = self.init_month_data() total_energy, total_co2 = 0.0, 0.0 for record in result: energy, co2 = self.aggregate_daily_record(month_data, record) total_energy += energy total_co2 += co2 month_order = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] monthly_data_list = self.build_monthly_data_list(month_data, month_order) best_month_entry = self.find_best_month(monthly_data_list) avg_energy_per_month = round(total_energy / 12, 2) if monthly_data_list else 0 return { "monthlyData": monthly_data_list, "totalEnergy": round(total_energy, 2), "totalCO2Saved": round(total_co2, 2), "avgEnergyPerMonth": round(avg_energy_per_month, 2), "bestMonth": best_month_entry } except Exception as e: # log error raise e @staticmethod def calculate_co2_saved(kwh_generated): return float(kwh_generated) * 0.82 @staticmethod def calculate_amount_saved(kwh_generated): return float(kwh_generated) * 9.0 def update_credentials(self, username, password, inverter_name, user_id): try: return self.queries.update_inverter_credentials(username, password, inverter_name, user_id) except Exception as e: # log error raise e @staticmethod def parse_report_date(report_date): """Convert string date to datetime object if needed.""" if isinstance(report_date, str): return date.strptime(report_date, "%Y-%m-%d") return report_date @staticmethod def init_month_data(): """Initialize month data dictionary with default energy and CO₂.""" return defaultdict(lambda: {"energy": 0.0, "co2Saved": 0.0}) def aggregate_daily_record(self,month_data, record): """Add a day's energy and CO₂ data into the correct month bucket.""" date = self.parse_report_date(record.get("report_date")) generation_kwh = float(record.get("generation_kwh", 0)) co2_saved = generation_kwh * 0.82 month_short = date.strftime("%b") month_data[month_short]["energy"] += generation_kwh month_data[month_short]["co2Saved"] += co2_saved return generation_kwh, co2_saved @staticmethod def build_monthly_data_list(month_data, month_order): """Create sorted list of monthly energy data, filling empty months.""" return [ { "month": m, "energy": round(month_data[m]["energy"], 2) if m in month_data else 0.0, "co2Saved": round(month_data[m]["co2Saved"], 2) if m in month_data else 0.0 } for m in month_order ] @staticmethod def find_best_month(monthly_data): """Return the month with highest energy, or None if all are zero.""" return max(monthly_data, key=lambda x: x["energy"]) if any(m["energy"] > 0 for m in monthly_data) else None def aggregate_by_key(self,daily_records, key_func): """ Generic aggregator that groups energy by a key (day, date, or month). key_func: function to extract the grouping key from datetime """ data = defaultdict(float) for record in daily_records: date = self.parse_report_date(record.get("report_date")) energy = float(record.get("generation_kwh", 0)) key = key_func(date) data[key] += energy return data @staticmethod def build_output_list(aggregated_data, sort_key_func): """Convert aggregated data dict into sorted list of {period, energy}.""" return [ {"period": str(period), "energy": round(energy, 2)} for period, energy in sorted(aggregated_data.items(), key=lambda x: sort_key_func(x[0])) ] def calculate_weekly_energy(self,user_id): """ Returns list of 7 entries: [ {"period": "2024-01-15", "energy": 45.2}, ... ] """ results = self.queries.get_weekly_energy(user_id) aggregated_data = self.aggregate_by_key(results, key_func=lambda d: d.strftime("%Y-%m-%d")) return self.build_output_list(aggregated_data, sort_key_func=lambda x: date.strptime(x, "%Y-%m-%d")) def calculate_monthly_energy(self,user_id): """ Returns list for each day of the month: [ {"period": "1", "energy": 45.2}, ... ] """ results = self.queries.get_monthly_energy(user_id, date.now().month, date.now().year) aggregated_data = self.aggregate_by_key(results, key_func=lambda d: d.day) return self.build_output_list(aggregated_data, sort_key_func=lambda x: int(x)) def calculate_yearly_energy(self,user_id): """ Returns list for each month of the year: [ {"period": "1", "energy": 1245.8}, ... ] """ results = self.queries.get_yearly_energy(user_id, date.now().year) aggregated_data = self.aggregate_by_key(results, key_func=lambda d: d.month) return self.build_output_list(aggregated_data, sort_key_func=lambda x: int(x))