Skip to main content

Function Reference & Data Flow

This document provides a comprehensive catalog of all functions in PVTools with complete source code and detailed explanations for developers to understand the entire optimization workflow.

Table of Contents​

  1. Core Entry Points
  2. Production Modeling
  3. Technical Utilities
  4. Financial Utilities
  5. Data Processing
  6. Profile Estimation
  7. BESS Utilities
  8. Optimization Engines
  9. Data Flow Analysis
  10. Key Data Structures

Core Entry Points​

main() - Consumption Profile Estimation​

Location: pv_tools/optimiser.py
Purpose: Initialize optimization environment and estimate consumption profile from billing data

Source Code:

def main(
    c: Dict,
    log_file: Optional[str] = None,
    typical_consumption: Optional[List[float]] = None,
) -> List[float]:
    # delete log_file if it exists
    if log_file is not None and os.path.exists(log_file):
        os.remove(log_file)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        filename=log_file,
    )

    stream_handler = logging.StreamHandler()
    stream_handler.setLevel(logging.INFO)
    stream_formatter = logging.Formatter("%(levelname)s: %(message)s")
    stream_handler.setFormatter(stream_formatter)
    logging.getLogger().addHandler(stream_handler)

    logger = logging.getLogger(__name__)

    logger.info(f"Latitude: {c['latitude']}, Longitude: {c['longitude']}")

    # assuming 30 days in a month
    if typical_consumption is None:
        logger.info(
            "No typical consumption provided. Calculating typical consumption based on yaml info."
        )
        monthly_energy = solve_pre_surcharge_bill(
            c,
            c["parameters"]["tariff_category"],
            c["parameters"]["monthly_bill"],
            c["parameters"]["peak_power_demand"],
            c["parameters"]["peak_fraction"],
        )
        logger.info(f"Monthly energy consumption: {np.round(monthly_energy,0)} kWh")
        daily_energy = monthly_energy / 30
        typical_consumption = estimate_load_profile(
            c["parameters"]["daytime_ratio"],
            daily_energy,
            c["parameters"]["peak_power_demand"],
        )
    return typical_consumption

Detailed Explanation:

  1. Logging Setup:

    • Removes existing log file if provided
    • Configures dual logging (file + console) with timestamps
    • File logging includes full details, console shows level and message only

  2. Consumption Estimation Logic:

    • If no consumption profile provided, performs reverse calculation
    • Uses solve_pre_surcharge_bill() to convert monthly bill to kWh consumption
    • Assumes 30-day month for daily energy calculation
    • Calls estimate_load_profile() to create realistic 24-hour pattern

  3. Key Dependencies:

    • solve_pre_surcharge_bill(): Reverse bill calculation
    • estimate_load_profile(): 24-hour profile generation
    • Configuration must include: tariff_category, monthly_bill, peak_power_demand, daytime_ratio

optimise() - Main Optimization Workflow​

Location: pv_tools/optimiser.py
Purpose: Execute complete optimization workflow with dynamic optimizer selection

Source Code:

def optimise(c: Dict, typical_consumption: List[float]) -> Tuple[Dict, Dict]:
    calc_type = c["parameters"]["contract_type"]
    financing_method = c["parameters"]["financing"]
    module_path = f"pv_tools.optimizers.{financing_method}.{calc_type}"
    module = __import__(module_path, fromlist=[calc_type])
    method = getattr(module, calc_type)
    result, c = method(c, typical_consumption)

    kg2tonnes = 1e-3
    months2year = 12
    W2kW = 1e-3

    # calculate some additional outputs
    result["no_of_panels"] = result["system_size"] / (
        c["standard_parameters"]["power_output_per_panel"] * W2kW
    )
    result["annual_tnb_tariff_increase"] = c["prices"][
        "TNB_avg_annual_tariff_increase"
    ]["rate"]

    # esg metrics
    result["ESG"] = {}
    result["ESG"]["annual_tonnes_of_CO2_reduced"] = (
        result["monthly_energy_production"]
        * months2year
        * c["ESG"]["grid_emission_factor"]["value"]
        * kg2tonnes
    )
    result["ESG"]["trees_planted_per_year"] = (
        result["ESG"]["annual_tonnes_of_CO2_reduced"]
        / kg2tonnes
        / c["ESG"]["mature_tree_absorption"]["value"]
    )

    return result, c

Detailed Explanation:

  1. Dynamic Optimizer Selection:

    • Constructs module path: pv_tools.optimizers.{financing_method}.{calc_type}
    • Examples: Tokenization.PPA, TermLoan.Instalment
    • Uses __import__() for runtime module loading
    • Gets callable function from module

  2. Result Enrichment:

    • Panel Count: system_size / (power_output_per_panel * 1e-3)
    • Tariff Escalation: Annual TNB tariff increase rate
    • ESG Metrics: CO2 reduction and tree equivalents

  3. ESG Calculations:

    • CO2 Reduction: monthly_production * 12 * emission_factor * 1e-3
    • Tree Equivalents: CO2_reduction / tree_absorption_rate

Production Modeling​

modelSetup() - Configuration Enhancement​

Location: pv_tools/production_modeling.py
Purpose: Load external data files and resolve geographic coordinates

Source Code:

def modelSetup(c: dict) -> dict:
    # load in all necessary data
    with open(c["paths"]["tariffs"], "r") as f:
        grid_tariffs = json.load(f)

    with open(c["paths"]["prices"], "r") as f:
        prices = json.load(f)

    with open(c["paths"]["ESG"], "r") as f:
        ESG = json.load(f)

    with open(c["paths"]["inverters"], "r") as f:
        inverters = json.load(f)

    with open(c["paths"]["CT_rating"], "r") as f:
        CT_rating_data = json.load(f)

    # get longitude and latitude if not provided
    if (
        c["parameters"]["location"]["longitude"] is None
        or c["parameters"]["location"]["latitude"] is None
    ):
        latitude, longitude = getCoordinates(
            c["parameters"]["location"]["street_address"]
        )
    else:
        latitude = c["parameters"]["location"]["latitude"]
        longitude = c["parameters"]["location"]["longitude"]

    # set export rate to 0 for SELCO scheme
    if c["parameters"]["scheme"] == "SELCO":
        c["standard_parameters"]["SMP_export"]["rate"] = 0

    c["grid_tariffs"] = grid_tariffs
    c["prices"] = prices
    c["ESG"] = ESG
    c["inverters"] = inverters
    c["latitude"] = latitude
    c["longitude"] = longitude
    c["CT_rating_data"] = CT_rating_data

    return c

Detailed Explanation:

  1. Data File Loading:

    • Loads 5 critical JSON files: tariffs, prices, ESG, inverters, CT_rating
    • Each file contains structured data for calculations
    • Files are loaded synchronously (could be optimized for parallel loading)

  2. Geographic Resolution:

    • Checks if lat/lon provided in config
    • If missing, calls getCoordinates() for geocoding
    • Uses OpenStreetMap Nominatim service

  3. Scheme-Specific Adjustments:

    • SELCO scheme: Sets export rate to 0 (no grid export)
    • NOVA/SARE: Uses configured export rates

Technical Utilities​

calculate_bill() - TNB Bill Calculation Engine​

Location: pv_tools/utilities/technical.py
Purpose: Calculate comprehensive TNB electricity bills with all surcharges

Source Code:

def calculate_bill(
    c,
    tariff_category,
    total_consumption,
    peak_power_demand=None,
    peak_fraction=None,
):
    # first check if total_consumption is negative
    if total_consumption < 0:
        # use SMP export rate if negative
        total_energy_charge = (
            total_consumption * c["standard_parameters"]["SMP_export"]["rate"]
        )
        return total_energy_charge

    tariff_json = c["grid_tariffs"]
    tariff_data = tariff_json.get(tariff_category)
    if not tariff_data:
        raise ValueError(f"Tariff category {tariff_category} not found in tariff JSON.")

    charge_type = tariff_data.get("charge_type", "energy_only")

    if charge_type == "energy_only":
        # This covers tariffs that use a tiered pricing structure.
        tiers = tariff_data.get("tiers")
        if tiers is None:
            # Fall back to a single energy_charge if provided.
            energy_info = tariff_data.get("energy_charge")
            if energy_info is None:
                raise ValueError(
                    "No tier or energy_charge information available for tariff category."
                )
            effective_rate = energy_info["rate"] / 100.0  # Convert sen to RM
            total_energy_charge = total_consumption * effective_rate

        remaining_consumption = total_consumption
        total_energy_charge = 0.0
        for tier in tiers:
            rate_rm = tier["rate"] / 100.0  # convert sen/kWh to RM/kWh
            limit = tier.get(
                "limit"
            )  # kWh limit for this tier; None indicates no upper bound
            if limit is not None:
                if remaining_consumption >= limit:
                    total_energy_charge += limit * rate_rm
                    remaining_consumption -= limit
                else:
                    total_energy_charge += remaining_consumption * rate_rm
                    remaining_consumption = 0
                    break
            else:
                # Unlimited tier: all remaining consumption converts to charge
                total_energy_charge += remaining_consumption * rate_rm
                remaining_consumption = 0
                break

    elif charge_type == "energy_and_demand":
        total_energy_charge = 0.0
        if peak_power_demand is not None:
            total_energy_charge += (
                peak_power_demand * tariff_data["demand_charge"]["rate"]
            )
        # For tariffs that include both energy and demand charges, we assume the
        # total_energy_charge provided is for energy consumption only.
        if "energy_charge" in tariff_data:
            # Use a single energy charge rate.
            effective_rate = tariff_data["energy_charge"]["rate"] / 100.0
            total_energy_charge += total_consumption * effective_rate
        elif (
            "energy_charge_peak" in tariff_data
            and "energy_charge_offpeak" in tariff_data
        ):
            # Use separate peak and off-peak rates.
            if peak_fraction is None:
                peak_fraction = 0.5  # default 50% if not provided
            peak_rate = tariff_data["energy_charge_peak"]["rate"]
            offpeak_rate = tariff_data["energy_charge_offpeak"]["rate"]
            total_energy_charge += (
                peak_fraction * total_consumption * peak_rate
                + (1 - peak_fraction) * total_consumption * offpeak_rate
            ) / 100.0
        else:
            raise ValueError("Energy charge information missing for tariff category.")
    else:
        raise ValueError(
            f"Unknown charge_type '{charge_type}' for tariff category {tariff_category}."
        )

    # add in ICPT charges
    ICPT_charge = total_consumption * c["standard_parameters"]["ICPT"]["rate"]
    # add in KWTBB charges
    KWTBB_charge = total_energy_charge * c["standard_parameters"]["KWTBB"]["rate"]

    total_energy_charge += ICPT_charge
    total_energy_charge += KWTBB_charge

    return total_energy_charge

Detailed Explanation:

  1. Export Handling:

    • Negative consumption = solar export
    • Uses SMP export rate for credits
    • Returns negative value (credit to customer)

  2. Tariff Structure Support:

    • energy_only: Tiered pricing (e.g., Domestic B)
    • energy_and_demand: Commercial tariffs with demand charges
    • Time-of-Use: Peak/off-peak rate structures

  3. Tiered Billing Logic:

    • Processes tiers sequentially
    • Each tier has rate and limit (None = unlimited)
    • Accumulates charges until consumption exhausted

  4. Surcharge Application:

    • ICPT: Applied to total consumption
    • KWTBB: Applied to energy charge (not demand)

Data Flow Analysis​

Main Optimization Workflow​

YAML Config
    ↓
modelSetup() - Load data files & resolve coordinates
    ↓
Enhanced Config (with tariffs, prices, ESG, inverters, lat/lon)
    ↓
main() - Estimate consumption profile
    ↓
Consumption Profile (24 hourly values)
    ↓
optimise() - Dynamic optimizer selection
    ↓
Financing Optimizer (Tokenization.PPA, TermLoan.Instalment, etc.)
    ↓
power_and_energy() - Core energy balance
    ↓
Optimization Loop (SLSQP)
    ↓
Results + ESG Metrics

Energy Balance Process Flow​

daily_consumption + config
    ↓
Scheme Constraints (NOVA/SELCO/SARE)
    ↓
Load Historical Production Data
    ↓
Scale by System Size
    ↓
Calculate Power Delta (consumption - production)
    ↓
BESS Sizing (if enabled)
    ↓
SOC Simulation (hour-by-hour)
    ↓
Monthly Aggregation
    ↓
Bill Calculations (with/without solar)
    ↓
Export Credits & Grid Draw
    ↓
Return monthly_data + storage_size

Data Dependencies​

External APIs:
- PVGIS (solar production data)
- OpenStreetMap (geocoding)

Local Data Files:
- tariffs.json (TNB tariff structures)
- prices.json (equipment & service costs)
- ESG.json (emission factors, tree absorption)
- inverters.json (available inverter specs)
- CT_rating.json (connection capacity limits)

Function Dependencies:
- calculate_bill() ← tariffs.json
- size_inverters() ← inverters.json
- get_fuse_limitation() ← CT_rating.json
- calculate_project_cost() ← prices.json
- ESG calculations ← ESG.json

Key Data Structures​

Configuration Dictionary (c)​

{
    "parameters": {
        "location": {
            "latitude": float,      # Decimal degrees
            "longitude": float,     # Decimal degrees
            "street_address": str   # For geocoding if lat/lon missing
        },
        "tariff_category": str,     # "B", "C1", "C2", etc.
        "monthly_bill": float,      # Current monthly bill (RM)
        "peak_power_demand": float, # Peak demand (kW)
        "scheme": str,              # "NOVA", "SELCO", "SARE"
        "contract_type": str,       # "PPA", "DirectPurchase", "Instalment"
        "financing": str,           # "Tokenization", "TermLoan"
        "daytime_ratio": float,     # Daytime consumption ratio
        "efficiency_scaling": float # System efficiency factor
    },
    "decision_variables": {
        "system_size": float,       # Solar system size (kWp)
        "declination": float,       # Panel tilt angle
        "azimuth": float           # Panel orientation
    },
    "paths": {
        "tariffs": str,             # Path to tariffs.json
        "prices": str,              # Path to prices.json
        "ESG": str,                 # Path to ESG.json
        "inverters": str,           # Path to inverters.json
        "CT_rating": str            # Path to CT_rating.json
    },
    "standard_parameters": {
        "SMP_export": {"rate": float},  # Export rate (RM/kWh)
        "power_output_per_panel": float, # Panel power (W)
        "ICPT": {"rate": float},        # ICPT surcharge rate
        "KWTBB": {"rate": float}        # KWTBB surcharge rate
    }
}

Monthly Data DataFrame​

{
    "year": int,                    # Year
    "month": int,                   # Month (1-12)
    "E_consumed": float,            # Monthly consumption (kWh)
    "E_produced": float,            # Monthly production (kWh)
    "E_grid_draw": float,           # Grid import (kWh)
    "exported_kWh": float,          # Grid export (kWh)
    "export_credit_RM": float,      # Export earnings (RM)
    "peak_power_demand": float,     # Peak demand with solar (kW)
    "bill_excluding_solar": float,  # Grid bill minus export credits (RM)
    "bill_without_solar": float,    # Total bill without solar (RM)
    "bill_equivalent_solar": float, # Cost of solar at grid rates (RM)
    "peak_sun_hours": float         # Equivalent peak sun hours
}

Optimization Results Dictionary​

{
    "system_size": float,           # Optimal system size (kWp)
    "PPA_tariff": float,            # Optimal PPA rate (RM/kWh)
    "monthly_energy_consumption": float, # Monthly consumption (kWh)
    "monthly_energy_production": float,  # Monthly production (kWh)
    "total_savings": float,         # Total savings percentage
    "savings_on_solar_portion": float,   # Solar portion savings percentage
    "total_monthly_bill_with_solar": float, # Total bill with solar (RM)
    "total_monthly_bill_without_solar": float, # Total bill without solar (RM)
    "monthly_data": pd.DataFrame,   # Complete monthly data
    "daily_consumption": List[float], # 24-hour consumption profile
    "no_of_panels": int,            # Number of solar panels
    "ESG": {
        "annual_tonnes_of_CO2_reduced": float, # CO2 reduction
        "trees_planted_per_year": float        # Tree equivalents
    }
}

This comprehensive function reference provides developers with complete source code, detailed explanations, and data flow analysis for understanding and extending PVTools' optimization capabilities.