"""Gera as projeções nacionais, os CSVs e o painel da Projeção 2035. Premissas de operação: - sem gestão: três coortes consecutivas, 30% / 40% / 30%; - gestão: a energia flexível é redistribuída nas 24 h; - V2L: contribuição distribuída entre 18 h, 19 h e 20 h, também 30% / 40% / 30%. As bases Senatran e ABVE permanecem independentes. """ from pathlib import Path import csv import json from bases_brasil import load R = Path(__file__).resolve().parent obs, growth, fleet, reconciliation = load() profiles = json.loads((R / "perfis_carga_solar_2025_2026.json").read_text(encoding="utf-8-sig")) ARRIVAL_WEIGHTS = (0.30, 0.40, 0.30) V2L_WEIGHTS = {18: 0.30, 19: 0.40, 20: 0.30} def fill_valleys(base, energy, hours, cap, spread=4): """Distribui energia em horas de menor carga, preservando energia e potência.""" out = [0.0] * 24 if energy <= 0: return out if energy > cap * len(hours) + 1e-7: raise ValueError("Energia excede a janela de recarga disponível") lo = min(base[h] for h in hours) hi = max(base[h] for h in hours) + cap * (1 + spread) for _ in range(90): level = (lo + hi) / 2 total = sum(min(cap, max(0, (level - base[h]) / (1 + spread))) for h in hours) if total < energy: lo = level else: hi = level for h in hours: out[h] = min(cap, max(0, ((lo + hi) / 2 - base[h]) / (1 + spread))) return out def add_cohort(curve, start, power_gw, duration_h): """Converte uma sessão curta em potência média por hora, inclusive na virada do dia.""" for hour in range(24): for day_shift in (-1, 0, 1): st = start + 24 * day_shift overlap = max(0, min(hour + 1, st + duration_h) - max(hour, st)) curve[hour] += power_gw * overlap def simulate(n, h, share=.8, power=50, energy=20, flex=1., vshare=.1, house=1, eff=.85, base=None, spread=4): if not (0 <= share <= 1 and 0 <= vshare <= 1 - share + 1e-9): raise ValueError("As participações de recarga e V2L devem somar no máximo 100%") base = base if base is not None else [v / 1000 for v in profiles["2025-SIN"]["net"]] mobility_energy = n * share * energy / 1e6 # GWh power_cap = n * share * power / 1e6 # GW se todos os participantes carregassem juntos duration = energy / power # horas por sessão red = [0.0] * 24 for offset, weight in enumerate(ARRIVAL_WEIGHTS): add_cohort(red, h + offset, power_cap * weight, duration) unmanaged_instant_peak = power_cap * max(ARRIVAL_WEIGHTS) fixed = [v * (1 - flex) for v in red] allocation = fill_valleys( [base[i] + fixed[i] for i in range(24)], mobility_energy * flex, list(range(24)), power_cap * flex, spread, ) blue = [fixed[i] + allocation[i] for i in range(24)] discharge_energy = n * vshare * house / 1e6 # GWh equivalentes a 1 h por participante v2l_curve = [0.0] * 24 for hour, weight in V2L_WEIGHTS.items(): v2l_curve[hour] = discharge_energy * weight replacement = discharge_energy / eff if eff else 0 topup = fill_valleys( [base[i] + blue[i] for i in range(24)], replacement, list(range(18)), n * vshare * power / 1e6, spread, ) green = [blue[i] + topup[i] - v2l_curve[i] for i in range(24)] tol_m = max(1e-8, mobility_energy * 1e-10) tol_v = max(1e-8, discharge_energy * 1e-10) assert abs(sum(red) - mobility_energy) < tol_m assert abs(sum(blue) - mobility_energy) < tol_m assert abs(sum(v2l_curve) - discharge_energy) < tol_v assert abs(sum(green) - (mobility_energy + replacement - discharge_energy)) < tol_m return { "energy": mobility_energy, "peak": unmanaged_instant_peak, "power_cap": power_cap, "managed": allocation, "discharge": max(v2l_curve), "discharge_energy": discharge_energy, "replacement": replacement, "curves": {"Sem gestão": red, "Inteligente": blue, "Inteligente + V2L": green}, "instant": [unmanaged_instant_peak, max(blue), max(green)], "v2l_hours": "18-20", "v2l_curve": v2l_curve, "topup": topup, } def write_semicolon(path, rows, fieldnames): with path.open("w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=fieldnames, delimiter=";") w.writeheader() w.writerows(rows) # Frota anual: ordenação e nomes voltados à leitura humana sem perder estrutura tabular. fleet_rows = [] for r in fleet: fleet_rows.append({ "fonte": r["fonte"], "recorte": r["recorte"], "tecnologia": r["tecnologia"], "cenario": r["cenario"], "ano": r["ano"], "taxa_anual": r["taxa"], "frota_veiculos": r["frota"], "entradas_veiculos": r["entradas"] if r["entradas"] is not None else "", "mes_referencia": r["mes_referencia"], "natureza": r["natureza"], "taxa_aplicada_a": r["taxa_sobre"], }) write_semicolon(R / "brasil_frota_crescimento.csv", fleet_rows, list(fleet_rows[0])) # Cenários-padrão. scenario_rows = [] for r in fleet: for h in (12, 18): s = simulate(r["frota"], h) for i, (strategy, curve) in enumerate(s["curves"].items()): scenario_rows.append({ "fonte": r["fonte"], "recorte": r["recorte"], "tecnologia": r["tecnologia"], "cenario": r["cenario"], "ano": r["ano"], "taxa_anual": r["taxa"], "frota_veiculos": r["frota"], "mes_referencia": r["mes_referencia"], "natureza": r["natureza"], "taxa_aplicada_a": r["taxa_sobre"], "hora_evento": h, "estrategia": strategy, "perfil_sem_gestao": "30%-40%-30% em três horas consecutivas", "pico_instantaneo_adicional_GW": s["instant"][i], "delta_medio_hora_inicial_GW": curve[h], "delta_medio_hora_mais_1_GW": curve[(h + 1) % 24], "delta_medio_hora_mais_2_GW": curve[(h + 2) % 24], "energia_mobilidade_GWh": s["energy"], "energia_liquida_24h_GWh": sum(curve), "pico_horario_delta_24h_GW": max(curve), "contribuicao_V2L_max_GW": s["discharge"] if i == 2 else 0, "energia_V2L_GWh": s["discharge_energy"] if i == 2 else 0, "reposicao_V2L_GWh": s["replacement"] if i == 2 else 0, "janela_V2L": s["v2l_hours"] if i == 2 else "", "referencia_carga_2025_GW": profiles["2025-SIN"]["load"][h] / 1000, "referencia_menos_solar_2025_GW": profiles["2025-SIN"]["net"][h] / 1000, }) write_semicolon(R / "brasil_cenarios_12h_18h.csv", scenario_rows, list(scenario_rows[0])) # Sensibilidade V2L sobre o pico histórico. base = [v / 1000 for v in profiles["2025-SIN"]["net"]] historical = max(base) v2l_rows = [] for r in fleet: for participation in (0, .05, .10, .15, .20): s = simulate(r["frota"], 18, vshare=participation) smart = [base[i] + s["curves"]["Inteligente"][i] for i in range(24)] total = [base[i] + s["curves"]["Inteligente + V2L"][i] for i in range(24)] peak = max(total) v2l_rows.append({ **r, "participacao_V2L_pct": participation * 100, "pico_historico_GW": historical, "pico_sem_V2L_GW": max(smart), "contribuicao_V2L_max_GW": s["discharge"], "energia_V2L_GWh": s["discharge_energy"], "contribuicao_18h_GW": s["v2l_curve"][18], "contribuicao_19h_GW": s["v2l_curve"][19], "contribuicao_20h_GW": s["v2l_curve"][20], "janela_V2L": "18-20", "pico_resultante_GW": peak, "hora_pico": total.index(peak), "reducao_historico_GW": historical - peak, "reducao_historico_pct": (historical - peak) / historical * 100, "reposicao_V2L_GWh": s["replacement"], }) write_semicolon(R / "brasil_v2l_pico_historico.csv", v2l_rows, list(v2l_rows[0])) # JSON consumido pelo painel. data = { "growth": growth, "observed": list(obs.values()), "fleet": fleet, "profiles": profiles, "reconciliation": reconciliation, "model": { "arrival_weights": list(ARRIVAL_WEIGHTS), "v2l_weights": {str(k): v for k, v in V2L_WEIGHTS.items()}, "updated": "2026-09-26", }, } (R / "brasil_dados.json").write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") # Validação compacta. def validation_case(source="ABVE", kind="BEV", scenario="Intermediário", year=2035, vshare=.10): r = next(x for x in fleet if x["fonte"] == source and x["tecnologia"] == kind and x["cenario"] == scenario and x["ano"] == year) s = simulate(r["frota"], 18, vshare=vshare) total = [base[i] + s["curves"]["Inteligente + V2L"][i] for i in range(24)] return { "fonte": source, "tecnologia": kind, "cenario": scenario, "ano": year, "frota": r["frota"], "participacao_V2L_pct": vshare * 100, "sem_gestao_18_20_GW": {str(h): s["curves"]["Sem gestão"][h] for h in (18, 19, 20)}, "V2L_18_20_GW": {str(h): s["v2l_curve"][h] for h in (18, 19, 20)}, "energia_mobilidade_GWh": s["energy"], "energia_V2L_GWh": s["discharge_energy"], "reposicao_V2L_GWh": s["replacement"], "pico_resultante_GW": max(total), "hora_pico": total.index(max(total)), } validation = { "modelo": "recarga sem gestão distribuída em três coortes + gestão 24h + V2L distribuído 18-20h", "perfil_sem_gestao": {"pesos": list(ARRIVAL_WEIGHTS), "horas_evento_noturno": [18, 19, 20]}, "perfil_V2L": {"pesos": [V2L_WEIGHTS[h] for h in (18, 19, 20)], "horas": [18, 19, 20]}, "energia_preservada": True, "caso_ABVE_2035": validation_case(), "caso_SENATRAN_2035": validation_case(source="SENATRAN"), } (R / "brasil_validacao.json").write_text(json.dumps(validation, ensure_ascii=False, indent=2), encoding="utf-8") # HTML com dados incorporados para funcionar como arquivo estático. template = R / "brasil-cenarios.template.html" if template.exists(): html = template.read_text(encoding="utf-8-sig").replace("__DATA__", json.dumps(data, ensure_ascii=False)) (R / "brasil-cenarios.html").write_text(html, encoding="utf-8") print(json.dumps({ "frota_registros": len(fleet), "cenarios_registros": len(scenario_rows), "validacao": validation, }, ensure_ascii=False, indent=2))