"""Illustrative water-supply screen. Standard library only. All money is assumed real 2026 USD. The service target is held fixed. Lower utilization increases installed capacity; it does NOT reduce annual delivered volume or annual electricity in this model. This is annual arithmetic, not an hourly reliability or engineering model. """ import argparse import json from pathlib import Path DEFAULTS = {"electricity_price": 0.12, "utilization": 0.90, "grid_intensity": 0.35} TARGET_M3_DAY = 100_000 DAYS = 365 REFERENCE_CAPACITY_M3_DAY = 100_000 RATE = 0.04 LIFE_YEARS = 30 FIXED_OM_FRACTION = 0.02 TECHNOLOGIES = { "desal": {"name": "Seawater reverse osmosis", "plant_kwh_m3": 3.5, "delivery_kwh_m3": 0.3, "reference_capex_usd": 350_000_000, "variable_nonenergy_usd_m3": 0.15}, "reuse": {"name": "Advanced potable reuse", "plant_kwh_m3": 0.84, "delivery_kwh_m3": 0.16, "reference_capex_usd": 200_000_000, "variable_nonenergy_usd_m3": 0.15}, } def capital_recovery_factor(rate, years): if rate < 0 or years <= 0: raise ValueError("Rate must be nonnegative and years positive.") return 1 / years if rate == 0 else rate * (1 + rate) ** years / ((1 + rate) ** years - 1) def calculate(electricity_price=0.12, utilization=0.90, grid_intensity=0.35): if not (0 < utilization <= 1): raise ValueError("Utilization must be greater than 0 and no more than 1.") if electricity_price < 0 or grid_intensity < 0: raise ValueError("Price and grid intensity must be nonnegative.") if not all(__import__('math').isfinite(x) for x in (electricity_price, utilization, grid_intensity)): raise ValueError("Inputs must be finite.") annual_volume = TARGET_M3_DAY * DAYS capacity = TARGET_M3_DAY / utilization crf = capital_recovery_factor(RATE, LIFE_YEARS) result = {"inputs": {"electricity_price": electricity_price, "utilization": utilization, "grid_intensity": grid_intensity}, "annual_volume_m3": annual_volume, "capacity_m3_day": capacity, "crf": crf, "technologies": {}} for key, tech in TECHNOLOGIES.items(): energy = tech["plant_kwh_m3"] + tech["delivery_kwh_m3"] capex = tech["reference_capex_usd"] * capacity / REFERENCE_CAPACITY_M3_DAY annual_capital = capex * crf fixed_om = capex * FIXED_OM_FRACTION variable_om = annual_volume * tech["variable_nonenergy_usd_m3"] electricity = annual_volume * energy electric_cost = electricity * electricity_price annual_cost = annual_capital + fixed_om + variable_om + electric_cost result["technologies"][key] = { "energy_kwh_m3": energy, "capex_usd": capex, "annual_capital_usd": annual_capital, "annual_fixed_om_usd": fixed_om, "annual_variable_om_usd": variable_om, "annual_electricity_kwh": electricity, "annual_electricity_gwh": electricity / 1_000_000, "average_power_mw": electricity / (DAYS * 24) / 1000, "online_power_mw": capacity * energy / 24 / 1000, "annual_electricity_cost_usd": electric_cost, "annual_cost_usd": annual_cost, "unit_cost_usd_m3": annual_cost / annual_volume, "annual_electricity_emissions_tco2e": electricity * grid_intensity / 1000, "emissions_kgco2e_m3": energy * grid_intensity, "unit_capital_usd_m3": annual_capital / annual_volume, "unit_fixed_om_usd_m3": fixed_om / annual_volume, "unit_variable_om_usd_m3": tech["variable_nonenergy_usd_m3"], "unit_electricity_usd_m3": energy * electricity_price, } return result def physical_balance(product_m3_day=100_000, recovery=0.45, feed_salinity_g_l=35): if not (0 < recovery < 1): raise ValueError("Recovery must be between 0 and 1, exclusive.") intake = product_m3_day / recovery brine = intake - product_m3_day return {"product_m3_day": product_m3_day, "intake_m3_day": intake, "brine_m3_day": brine, "brine_salinity_g_l": feed_salinity_g_l / (1-recovery), "note": "Simplified average balance; negligible product salt, no pretreatment losses, no dilution water."} def portfolio(savings_m3_day=20_000, eligible_effluent_m3_day=75_000, reuse_recovery=0.80): if min(savings_m3_day, eligible_effluent_m3_day) < 0 or not (0 <= reuse_recovery <= 1): raise ValueError("Invalid portfolio inputs.") reuse = min(eligible_effluent_m3_day * reuse_recovery, max(0, TARGET_M3_DAY-savings_m3_day)) return {"savings_m3_day": savings_m3_day, "reuse_m3_day": reuse, "residual_m3_day": max(0, TARGET_M3_DAY - savings_m3_day - reuse), "effluent_needed_to_avoid_desal_m3_day": (TARGET_M3_DAY - savings_m3_day) / reuse_recovery if reuse_recovery else None} def supplemental(): base = calculate() low_desal = (250_000_000/.9*(base['crf']+.02)/36_500_000)+.10+3*.12 high_desal = (600_000_000/.9*(base['crf']+.02)/36_500_000)+.30+5*.12 low_reuse = (150_000_000/.9*(base['crf']+.02)/36_500_000)+.10+.7*.12 high_reuse = (400_000_000/.9*(base['crf']+.02)/36_500_000)+.25+2*.12 return { "target_us_mgd": TARGET_M3_DAY / 3785.411784, "annual_target_acre_feet": TARGET_M3_DAY * DAYS / 1233.48183754752, "additional_liters_per_resident_day": TARGET_M3_DAY * 1000 / 300_000, "conservation_unit_cost_usd_m3": 3_650_000 / (20_000*DAYS), "conservation_half_yield_unit_cost_usd_m3": 3_650_000 / (10_000*DAYS), "desal_cost_stress_range_usd_m3": [low_desal, high_desal], "reuse_cost_stress_range_usd_m3": [low_reuse, high_reuse], "grid_threshold_for_half_kg_m3": .5 / 3.8, "full_nameplate_100k_at_90pct_annual_m3": 100_000*365*.9, "annual_deficit_at_100k_nameplate_90pct_m3": 100_000*365*.1, "three_day_storage_m3": 100_000*3, "portfolio_energy_gwh": (60_000*1.0+20_000*3.8)*365/1_000_000, "portfolio_energy_reduction_vs_desal_percent": 100*(1-(60_000+20_000*3.8)/(100_000*3.8)), "elevated_reuse_energy_countercase_gwh": 100_000*365*3/1_000_000, "pump_energy_100m_head_80pct_kwh_m3": 1000*9.81*100/(.8*3_600_000), "ocwd_m3_day_from_130_mgd": 130*3785.411784, "carlsbad_m3_day_from_50_mgd": 50*3785.411784, "carlsbad_annual_m3_purchase_range": [48_000*1233.48183754752,56_000*1233.48183754752], } if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--price',type=float,default=.12) parser.add_argument('--utilization',type=float,default=.9) parser.add_argument('--grid',type=float,default=.35) parser.add_argument('--output', type=Path) args=parser.parse_args() result={"scenario":calculate(args.price,args.utilization,args.grid),"physical_balance":physical_balance(),"portfolio":portfolio(),"supplemental":supplemental()} text=json.dumps(result,indent=2,allow_nan=False)+'\n' if args.output: args.output.write_text(text) else: print(text,end='')