import math BASE_YEAR = 2024 END_YEAR = 2050 ELECTRIC_HEAT = {'resistance', 'existing_heat_pump', 'existing_electric_unsplit'} FUELS = {'oil', 'gas', 'coal', 'wood', 'other'} def interpolate(start, end, fraction): return start + (end - start) * fraction def adoption(year, path): if len(path) < 2 or any(not 0 <= share <= 1 for _, share in path) or any(b[0] <= a[0] or b[1] < a[1] for a, b in zip(path, path[1:])): raise ValueError('Adoption path requires increasing years and nondecreasing shares between zero and one') if path[0] != [BASE_YEAR, 0] or path[-1][0] != END_YEAR: raise ValueError('Adoption path must begin at [2024, 0] and end in 2050') for (left_year, left), (right_year, right) in zip(path, path[1:]): if left_year <= year <= right_year: return interpolate(left, right, (year - left_year) / (right_year - left_year)) raise ValueError('Year is outside adoption path') def parameters(config, scenario='central', overrides=None): if scenario not in config['scenarios']: raise ValueError(f'Unknown scenario: {scenario}') values = {key: row['value'] for key, row in config['parameters'].items()} values.update(config['scenarios'][scenario]) for key, value in (overrides or {}).items(): if key not in values: raise ValueError(f'Unknown parameter: {key}') values[key] = value for key, value in values.items(): lower, upper = config['parameters'][key]['allowed'] if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or not lower <= value <= upper: raise ValueError(f'Invalid {key}: {value}; allowed [{lower}, {upper}]') return values def heat_replacement(carrier, p): if carrier in {'oil', 'gas', 'coal'}: return p['fossil_heat_conversion'] return p.get(f'{carrier}_heat_conversion', 0) def fuel_efficiency(carrier, p): return p.get(f'{carrier}_boiler_efficiency', p['solid_boiler_efficiency']) def calculate(baseline, config, year, scenario='central', overrides=None): if isinstance(year, bool) or not isinstance(year, int) or not BASE_YEAR <= year <= END_YEAR: raise ValueError(f'Year must be an integer from {BASE_YEAR} through {END_YEAR}') p = parameters(config, scenario, overrides) elapsed = (year - BASE_YEAR) / (END_YEAR - BASE_YEAR) replacement = adoption(year, config['adoption_path']) activities = { 'households': interpolate(1, p['population_million'] / baseline['population_million'], elapsed), 'services': interpolate(1, p['services_activity'], elapsed), 'industry': interpolate(1, p['industry_activity'], elapsed), 'transport': interpolate(1, p['other_transport_activity'], elapsed), } rows = [] retained_carriers = [] def add(sector, component, base, future, source, detail=None): if future < -1e-10: raise ValueError(f'Negative electricity in {component}: {future}') rows.append({'sector': sector, 'component': component, 'baseline_twh': base, 'electricity_twh': future, 'change_twh': future - base, 'source': source, 'calculation_details': detail or {}}) for sector, total in baseline['sector_electricity_twh'].items(): current_heat = sum(row['final_energy_twh'] for row in baseline['heat'] if row['sector'] == sector and row['carrier'] in ELECTRIC_HEAT) cooling = baseline['space_cooling_electricity_twh'].get(sector, 0) aux = baseline['household_heating_auxiliary_electricity_twh'] if sector == 'households' else 0 road = sum(row['existing_electricity_twh'] for row in baseline['road_transport']) if sector == 'transport' else 0 retained = total - current_heat - cooling - aux - road intensity_key = f'{sector}_electric_intensity' if sector != 'transport' else 'other_transport_electric_intensity' add(sector, 'retained_electricity', retained, retained * activities[sector] * interpolate(1, p[intensity_key], elapsed), 'BFE sector electricity minus explicitly modeled electric components') if cooling: future = cooling * activities[sector] * interpolate(1, p['cooling_service_multiplier'], elapsed) * interpolate(1, p['cooling_electric_intensity'], elapsed) add(sector, 'space_cooling', cooling, future, 'Tabelle52 climate cooling electricity; excludes process refrigeration') if aux: add(sector, 'legacy_heating_auxiliaries', aux, aux * activities[sector] * (1 - replacement * p['household_auxiliary_retirement']), 'Tabelle17!AA12; retirement allocation is a scenario assumption') for row in baseline['heat']: sector, carrier, end_use = row['sector'], row['carrier'], row['end_use'] energy = row['final_energy_twh'] space_fraction = 1 if end_use == 'space' else 0 if end_use == 'water' else baseline['nonresidential_space_fraction_proxy'][sector] space_intensity = interpolate(1, p['space_heat_intensity'], elapsed) * interpolate(1, p['space_heat_weather_multiplier'], elapsed) water_intensity = interpolate(1, p['water_heat_intensity'], elapsed) heat_scale = space_fraction * space_intensity + (1 - space_fraction) * water_intensity electric_per_heat = space_fraction * space_intensity / p['space_heat_spf'] + (1 - space_fraction) * water_intensity / p['water_heat_spf'] activity = activities[sector] detail = {'activity': activity, 'space_fraction_assumed_for_combined_carriers': space_fraction, 'heat_service_scale': heat_scale} if carrier in ELECTRIC_HEAT: if carrier == 'resistance': fraction = replacement * p['resistance_conversion'] future = energy * activity * ((1 - fraction) * heat_scale + fraction * electric_per_heat) detail['replacement_fraction'] = fraction else: factor = p['existing_hp_intensity'] if carrier == 'existing_heat_pump' else p['unsplit_existing_heat_intensity'] future = energy * activity * heat_scale * interpolate(1, factor, elapsed) add(sector, f'{end_use}_{carrier}', energy, future, row['source_cell'], detail) elif carrier in FUELS: fraction = replacement * heat_replacement(carrier, p) efficiency = fuel_efficiency(carrier, p) future = energy * activity * fraction * efficiency * electric_per_heat detail.update(displaced_input_before_activity_twh=energy * fraction, existing_generator_efficiency=efficiency, conversion_fraction=fraction) add(sector, f'{end_use}_new_electric_from_{carrier}', 0, future, row['source_cell'], detail) retained_carriers.append({'sector': sector, 'end_use': end_use, 'carrier': carrier, 'remaining_final_energy_twh': energy * activity * heat_scale * (1 - fraction)}) else: retained_carriers.append({'sector': sector, 'end_use': end_use, 'carrier': carrier, 'remaining_final_energy_twh': energy * activity * heat_scale, 'note': 'Existing ambient/solar and district heat are retained reference flows, not added electricity or full future renewable-heat accounting'}) converted_process = [] for row in baseline['process']: if row['sector'] != 'industry' or row['carrier'] not in FUELS: continue carrier = row['carrier'] target = p['process_fossil_conversion'] if carrier in {'oil', 'gas', 'coal'} else p[f'process_{carrier}_conversion'] converted_process.append((row, target * row['final_energy_twh'])) endpoint_converted = sum(value for _, value in converted_process) low_temperature = baseline['industry_process_below_100c_final_twh'] total_temperature = baseline['industry_total_process_final_twh'] overlap_lower = max(0, endpoint_converted + low_temperature - total_temperature) overlap_upper = min(endpoint_converted, low_temperature) assumed_overlap = endpoint_converted * p['process_heatpump_share'] if not overlap_lower - 1e-10 <= assumed_overlap <= overlap_upper + 1e-10: raise ValueError('Industrial heat-pump allocation exceeds feasible carrier/temperature marginal bounds') process_intensity = interpolate(1, p['process_heat_intensity'], elapsed) for row, converted in converted_process: useful = converted * replacement * activities['industry'] * process_intensity * p['process_existing_efficiency'] future = useful * (p['process_heatpump_share'] / p['process_heatpump_cop'] + (1 - p['process_heatpump_share']) / p['process_direct_efficiency']) add('industry', f'new_process_heat_from_{row["carrier"]}', 0, future, row['source_cell'], {'converted_2024_input_at_full_adoption_twh': converted, 'useful_heat_twh': useful, 'heatpump_share_assumption': p['process_heatpump_share']}) retained_carriers.append({'sector': 'industry', 'end_use': 'process', 'carrier': row['carrier'], 'remaining_final_energy_twh': (row['final_energy_twh'] - converted * replacement) * activities['industry'] * process_intensity}) for row in baseline['process']: if row['sector'] == 'households' and row['carrier'] in {'gas', 'wood'}: fraction = replacement * p['fossil_heat_conversion'] add('households', f'new_cooking_from_{row["carrier"]}', 0, row['final_energy_twh'] * activities['households'] * fraction * p['cooking_conversion_ratio'], row['source_cell']) road_activity = interpolate(1, p['road_activity'], elapsed) for row in baseline['road_transport']: mode = row['mode'] extra_activity = 1 + p['foreign_van_activity_addition'] if mode == 'vans' else 1 full_electric = row['million_vehicle_km'] / 1000 * extra_activity * p[f'{mode}_battery_kwh_per_km'] / p['charging_efficiency'] * p['domestic_charging_multiplier'] base = row['existing_electricity_twh'] future = road_activity * ((1 - replacement) * base + replacement * full_electric) add('transport', f'road_{mode}', base, future, row['electricity_source_cell'], {'million_vehicle_km_2024': row['million_vehicle_km'], 'full_electric_at_2024_activity_twh': full_electric, 'replacement_fraction': replacement, 'activity_multiplier': road_activity, 'charging_efficiency': p['charging_efficiency'], 'domestic_charging_multiplier': p['domestic_charging_multiplier']}) retained_carriers.append({'sector': 'transport', 'end_use': mode, 'carrier': 'non_electric_transport_energy', 'remaining_final_energy_twh': (row['existing_total_energy_twh'] - base) * road_activity * (1 - replacement), 'note': 'Covered road modes only; no claim of zero aviation, navigation or non-road fossil demand'}) base_residual = baseline['accounting']['model_baseline_unallocated_residual_twh'] add('unallocated', 'explicit_residual', base_residual, interpolate(base_residual, p['residual_twh'], elapsed), 'Detailed official final electricity minus published end-use sector sum') for key in ['data_centres_increment_twh', 'hydrogen_and_efuels_increment_twh', 'district_heat_increment_twh']: add('explicit_additions', key.removesuffix('_twh'), 0, p[key] * elapsed, 'Editable additional electricity assumption; not an observed future requirement') sectors = {} for row in rows: sectors[row['sector']] = sectors.get(row['sector'], 0) + row['electricity_twh'] total = sum(sectors.values()) base_loss = baseline['accounting']['published_losses_twh'] / baseline['accounting']['country_consumption_twh'] loss_fraction = interpolate(base_loss, p['effective_loss_fraction'], elapsed) return { 'year': year, 'scenario': scenario, 'evidence': 'illustrative_scenario_calculation' if year != BASE_YEAR else 'published_baseline_reconstruction', 'boundary': baseline['boundary'], 'technology_replacement_progress': replacement, 'parameters_2050': p, 'sector_electricity_twh': sectors, 'electricity_demand_twh': total, 'change_from_2024_twh': total - baseline['accounting']['final_consumption_twh'], 'annual_supply_accounting_before_storage_and_exports_twh': total / (1 - loss_fraction), 'effective_loss_fraction': loss_fraction, 'components': rows, 'remaining_non_electric_reference_flows': retained_carriers, 'industrial_low_temperature_allocation': {'assumed_twh_2024_basis': assumed_overlap, 'feasible_lower_twh': overlap_lower, 'feasible_upper_twh': overlap_upper, 'status': 'feasible_assumption_not_identified_joint_distribution'}, 'omissions': ['International aviation and synthetic fuels unless explicit additional electricity entered', 'Domestic non-road and navigation fuel conversion', 'Foreign van activity unless explicit addition entered', 'District heat production electrification unless explicit addition entered', 'National hourly load, winter peaks, storage, generation outages and cross-border adequacy'], }