import math
import numpy as np
import pandas as pd
from copy import deepcopy
from typing import List, Dict, Literal, Optional
from scipy.optimize import root_scalar
[docs]
class Plant:
"""
Plant class for techno-economic analysis of industrial processing plants.
This class models capital costs, operating expenses, revenue, and
financial metrics for chemical/process plants. It supports multiple
process types (Solids, Fluids, Mixed), geographic locations with regional
cost factors, and comprehensive cash flow analysis.
Attributes:
processTypes (dict): Cost multipliers for different process categories
(OS, DE, X).
locFactors (dict): Geographic location factors for capital cost
adjustments.
Configuration Parameters:
name (str): Plant name identifier.
process_type (str): Type of process
("Solids", "Fluids", or "Mixed").
country (str): Country location for cost factor lookup.
Default: "United States".
region (str): Regional area within country. Default: "Gulf Coast".
currency (str): Currency code (e.g., "USD"). Default: "USD".
exchange_rate (float): Conversion factor to base currency.
Default: 1.0.
interest_rate (float): Discount rate for NPV calculations.
Default: 0.09.
project_lifetime (int or array): Plant operating life in years.
Default: 20.
plant_utilization (float): Capacity utilization factor (0-1).
Default: 1.
tax_rate (float): Corporate tax rate for cash flow. Default: 0.
working_capital (float or None): Working capital requirement.
Auto-calculated if None.
depreciation (dict or None): Depreciation method configuration.
Cash Flow Profiles:
capex_ramp (list or None): Fraction of fixed capital spent in each
construction year. Must be a 1-D list of non-negative numbers
that sum to 1.0. Length must be less than project_lifetime.
Working capital is drawn in the final construction year and
released at the end of the project.
Default: [0.3, 0.6, 0.1] (3-year build).
production_ramp (list or None): Nameplate capacity utilisation
fraction for each project year (0–1). Values must be between
0 and 1. If shorter than project_lifetime the remaining years
are set to 1.0 (full capacity). Length must not exceed
project_lifetime.
Default: [0, 0, 0.4, 0.8] (full capacity from year 4 on).
Capital Cost Factors:
loc_factor (float or None): Location factor applied to ISBL.
Overrides country/region lookup when
set. Default: None.
fixed_capital_factors (dict): Override the multipliers used to
calculate individual fixed capital components. Any subset of
keys may be supplied; omitted keys fall back to processTypes
defaults.
Keys and defaults (process-type dependent):
"osbl" – fraction of ISBL (e.g. 0.3 for Fluids)
"de" – fraction of ISBL+OSBL (e.g. 0.3 for Fluids)
"contingency" – fraction of ISBL+OSBL (e.g. 0.1 for all types)
fixed_capital_components (dict): Override the computed cost value
of individual fixed capital components directly. Takes
precedence over fixed_capital_factors for the same component.
Keys match attribute names:
"osbl", "dne", "contingency"
Labor & Operations:
operators_per_shift (int or None): Manual input or auto-calculated.
operators_hired (int or None): Total operators needed;
auto-calculated if None.
operator_hourly_rate (dict or float): Wage rate for operators.
working_weeks_per_year (int): Annual working weeks. Default: 49.
working_shifts_per_week (int): Shifts per week. Default: 5.
operating_shifts_per_day (int): Daily operating shifts. Default: 3.
Equipment & Economics:
equipment_list (list): Equipment objects with cost data.
variable_opex_inputs (dict): Variable operating cost inputs
(consumption, price).
plant_products (dict): Product specifications
(production rate, price).
fc (float): Fixed capital cost multiplier for installed costs.
fp (float): Fixed OPEX cost multiplier.
Fixed OPEX Customisation:
fixed_opex_factors (dict): Override the multipliers used to
calculate individual fixed OPEX components. Any subset of keys
may be supplied; omitted keys fall back to defaults.
Keys and defaults:
"supervision" – 0.25 × operating_labor_costs
"direct_salary_overhead"– 0.50 × (labor + supervision)
"laboratory_charges" – 0.10 × operating_labor_costs
"maintenance" – 0.05 × ISBL
"taxes_insurance" – 0.015 × ISBL
"rent_of_land" – 0.015 × (ISBL + OSBL)
"environmental_charges" – 0.01 × (ISBL + OSBL)
"operating_supplies" – 0.009 × ISBL
"general_plant_overhead"– 0.65 × (labor + supervision
+ direct_salary_overhead)
"working_capital" – 0.15 × fixed_capital
"patents_royalties" – 0.02 × cash cost of production
"distribution_selling" – 0.02 × cash cost of production
"rnd" – 0.03 × cash cost of production
fixed_opex_components (dict): Override the computed cost value of
individual fixed OPEX components directly. Takes precedence
over fixed_opex_factors for the same component. Downstream
components that depend on an overridden value use the
overridden value in their own calculation.
Keys match attribute names:
"supervision_costs", "direct_salary_overhead",
"laboratory_charges", "maintenance_costs",
"taxes_insurance_costs", "rent_of_land_costs",
"environmental_charges", "operating_supplies",
"general_plant_overhead", "patents_royalties",
"distribution_selling_costs", "RnD_costs"
Additional Capex:
additional_capex_years (array): Years when additional capex occurs.
additional_capex_cost (array): Corresponding capex amounts.
Monte Carlo:
project_uncertainties (dict): Per-parameter uncertainty settings
for Monte Carlo simulation. Each key maps to a sub-dict with
optional fields ``std``, ``min``, and ``max``. Omitting a key
uses the built-in default distribution. Setting ``std=0``
disables sampling for that parameter (default for
plant_utilization and tax_rate). Supported keys:
"fixed_capital_factor" – std=0.3, min=0.25, max=1.75
"fixed_opex_factor" – std=0.3, min=0.25, max=1.75
"project_lifetime" – std=5, min/max auto (±2σ, ≥5)
"interest_rate" – std=0.03, min/max auto (±2σ, ≥0.02)
"plant_utilization" – std=0 (fixed unless overridden)
"tax_rate" – std=0 (fixed unless overridden)
monte_carlo_inputs (dict or None): Stochastic input distributions
populated after running monte_carlo().
monte_carlo_metrics (dict or None): Distribution results populated
after running monte_carlo().
Example:
>>> config = {
... "plant_name": "Example Plant",
... "process_type": "Fluids",
... "country": "United States",
... "region": "Gulf Coast",
... "equipment": [equipment_obj],
... "plant_products": {"product_A": {"production": 100,
"price": 50}},
... }
>>> plant = Plant(config)
>>> plant.calculate_all(print_results=True)
>>> npv = plant.calculate_npv()
"""
processTypes = {
"Solids": {"OS": 0.4, "DE": 0.2, "X": 0.1},
"Fluids": {"OS": 0.3, "DE": 0.3, "X": 0.1},
"Mixed": {"OS": 0.4, "DE": 0.25, "X": 0.1},
}
locFactors = {
"United States": {
"Gulf Coast": 1.00,
"East Coast": 1.04,
"West Coast": 1.07,
"Midwest": 1.02,
},
"Canada": {"Ontario": 1.00, "Fort McMurray": 1.60},
"Mexico": 1.03,
"Brazil": 1.14,
"China": {"imported": 1.12, "indigenous": 0.61},
"Japan": 1.26,
"Southeast Asia": 1.12,
"Australia": 1.21,
"India": 1.02,
"Middle East": 1.07,
"France": 1.13,
"Germany": 1.11,
"Italy": 1.14,
"Netherlands": 1.19,
"Russia": 1.53,
"United Kingdom": 1.02,
}
def __init__(self, configuration: dict):
"""Initialize plant from a configuration dictionary."""
# keep a copy of the original config so code can read from it later
self.config = deepcopy(configuration)
self.name = configuration.get("plant_name")
self.process_type = configuration.get(
"process_type"
)
self.country = configuration.get(
"country", "United States"
)
self.region = configuration.get(
"region", "Gulf Coast"
)
self.currency = configuration.get(
"currency", "USD"
)
self.exchange_rate = configuration.get(
"exchange_rate", 1.0
)
self.working_capital = configuration.get(
"working_capital", None
)
self.interest_rate = configuration.get(
"interest_rate", 0.09
)
self.project_lifetime = configuration.get(
"project_lifetime", 20
)
self.plant_utilization = configuration.get(
"plant_utilization", 1
)
self.tax_rate = configuration.get("tax_rate", 0)
self.depreciation = configuration.get(
"depreciation", None
)
self.operators_per_shift = configuration.get(
"operators_per_shift", None
)
self.operators_hired = configuration.get(
"operators_hired", None
)
self.working_weeks_per_year = configuration.get(
"working_weeks_per_year", 49
)
self.working_shifts_per_week = configuration.get(
"working_shifts_per_week", 5
)
self.operating_shifts_per_day = configuration.get(
"operating_shifts_per_day", 3
)
self.additional_capex_years = configuration.get(
"additional_capex_years", None
)
self.additional_capex_cost = configuration.get(
"additional_capex_cost", None
)
self.equipment_list = configuration.get(
"equipment", []
)
self.operator_hourly_rate = configuration.get(
"operator_hourly_rate", {}
)
self.project_uncertainties = configuration.get(
"project_uncertainties", {}
)
_validate_project_uncertainties(self.project_uncertainties)
self.variable_opex_inputs = configuration.get(
"variable_opex_inputs", {}
)
self.plant_products = configuration.get(
"plant_products", {}
)
self.fc = configuration.get("fc", None)
self.fp = configuration.get("fp", None)
self.capex_ramp = configuration.get("capex_ramp", None)
self.production_ramp = configuration.get(
"production_ramp", None
)
self.loc_factor = configuration.get("loc_factor", None)
self.fixed_opex_factors = configuration.get(
"fixed_opex_factors", {}
)
self.fixed_opex_components = configuration.get(
"fixed_opex_components", {}
)
self.fixed_capital_factors = (
configuration.get("fixed_capital_factors") or {}
)
self.fixed_capital_components = (
configuration.get("fixed_capital_components") or {}
)
self.monte_carlo_inputs = None
self.monte_carlo_metrics = None
[docs]
def update_configuration(self, configuration: dict):
"""
Update plant parameters while preserving nested structures.
Top-level scalar keys overwrite existing values. Nested dicts
(``variable_opex_inputs``, ``plant_products``, ``operator_hourly_rate``,
``project_uncertainties``, ``fixed_opex_factors``,
``fixed_capital_factors``) are deep-merged rather than replaced.
Parameters
----------
configuration : dict
Partial or full plant configuration. Only supplied keys are updated.
"""
# keep the stored config up to date
if (
not hasattr(self, "config")
or self.config is None
):
self.config = {}
# shallow-merge top-level keys first
self.config.update(
{
k: v
for k, v in configuration.items()
if k
not in [
"variable_opex_inputs",
"plant_products",
"operator_hourly_rate",
]
}
)
self.name = configuration.get(
"plant_name", self.name
)
self.process_type = configuration.get(
"process_type", self.process_type
)
self.country = configuration.get(
"country", self.country
)
self.region = configuration.get(
"region", self.region
)
self.equipment_list = configuration.get(
"equipment", self.equipment_list
)
self.working_capital = configuration.get(
"working_capital", self.working_capital
)
self.interest_rate = configuration.get(
"interest_rate", self.interest_rate
)
self.project_lifetime = configuration.get(
"project_lifetime", self.project_lifetime
)
self.plant_utilization = configuration.get(
"plant_utilization", self.plant_utilization
)
self.tax_rate = configuration.get(
"tax_rate", self.tax_rate
)
self.operators_per_shift = configuration.get(
"operators_per_shift", self.operators_per_shift
)
self.operators_hired = configuration.get(
"operators_hired", self.operators_hired
)
self.working_weeks_per_year = configuration.get(
"working_weeks_per_year",
self.working_weeks_per_year,
)
self.working_shifts_per_week = configuration.get(
"working_shifts_per_week",
self.working_shifts_per_week,
)
self.operating_shifts_per_day = configuration.get(
"operating_shifts_per_day",
self.operating_shifts_per_day,
)
self.additional_capex_years = configuration.get(
"additional_capex_years",
self.additional_capex_years,
)
self.additional_capex_cost = configuration.get(
"additional_capex_cost",
self.additional_capex_cost,
)
self.fc = configuration.get("fc", self.fc)
self.fp = configuration.get("fp", self.fp)
self.loc_factor = configuration.get(
"loc_factor", self.loc_factor
)
if "fixed_capital_factors" in configuration:
self.fixed_capital_factors = {
**self.fixed_capital_factors,
**configuration["fixed_capital_factors"],
}
if "fixed_capital_components" in configuration:
self.fixed_capital_components = {
**self.fixed_capital_components,
**configuration["fixed_capital_components"],
}
self.capex_ramp = configuration.get(
"capex_ramp", self.capex_ramp
)
self.production_ramp = configuration.get(
"production_ramp", self.production_ramp
)
if "fixed_opex_factors" in configuration:
self.fixed_opex_factors = {
**self.fixed_opex_factors,
**configuration["fixed_opex_factors"],
}
if "fixed_opex_components" in configuration:
self.fixed_opex_components = {
**self.fixed_opex_components,
**configuration["fixed_opex_components"],
}
# allow updating depreciation block
if "depreciation" in configuration:
self.depreciation = configuration[
"depreciation"
]
# merge nested variable_opex_inputs without clobbering
def recursive_update(original, updates):
for key, value in updates.items():
if isinstance(value, dict) and isinstance(
original.get(key), dict
):
recursive_update(original[key], value)
else:
original[key] = value
if "variable_opex_inputs" in configuration:
if (
not hasattr(self, "variable_opex_inputs")
or self.variable_opex_inputs is None
):
self.variable_opex_inputs = {}
recursive_update(
self.variable_opex_inputs,
configuration["variable_opex_inputs"],
)
# also mirror into stored config
if "variable_opex_inputs" not in self.config:
self.config["variable_opex_inputs"] = {}
recursive_update(
self.config["variable_opex_inputs"],
configuration["variable_opex_inputs"],
)
if "plant_products" in configuration:
if (
not hasattr(self, "plant_products")
or self.plant_products is None
):
self.plant_products = {}
recursive_update(
self.plant_products,
configuration["plant_products"],
)
# also mirror into stored config
if "plant_products" not in self.config:
self.config["plant_products"] = {}
recursive_update(
self.config["plant_products"],
configuration["plant_products"],
)
if "operator_hourly_rate" in configuration:
if (
not hasattr(self, "operator_hourly_rate")
or self.operator_hourly_rate is None
):
self.operator_hourly_rate = {}
recursive_update(
self.operator_hourly_rate,
configuration["operator_hourly_rate"],
)
# also mirror into stored config
if "operator_hourly_rate" not in self.config:
self.config["operator_hourly_rate"] = {}
recursive_update(
self.config["operator_hourly_rate"],
configuration["operator_hourly_rate"],
)
if "project_uncertainties" in configuration:
if (
not hasattr(self, "project_uncertainties")
or self.project_uncertainties is None
):
self.project_uncertainties = {}
recursive_update(
self.project_uncertainties,
configuration["project_uncertainties"],
)
if "project_uncertainties" not in self.config:
self.config["project_uncertainties"] = {}
recursive_update(
self.config["project_uncertainties"],
configuration["project_uncertainties"],
)
_validate_project_uncertainties(self.project_uncertainties)
[docs]
def calculate_purchased_cost(self, print_results=False):
"""
Sum equipment purchased costs with exchange rate conversion.
Parameters
----------
print_results : bool, optional
Print a per-equipment cost breakdown. Default is False.
Returns
-------
float
Total purchased cost in plant currency.
"""
self.purchased_cost = sum(
equipment.purchased_cost
for equipment in self.equipment_list
) * self.exchange_rate
if print_results:
# Print the results
print("Purchased cost estimation")
print("===================================")
for equipment in self.equipment_list:
cost = equipment.purchased_cost * self.exchange_rate
print(
f" - {equipment.name}: {cost:,.2f} "
f"{self.currency}"
)
print("===================================")
print(
f"Total Purchased Cost: "
f"{self.purchased_cost:,.2f} {self.currency}"
)
else:
return self.purchased_cost
[docs]
def calculate_isbl(self, fc=1.0, print_results=False):
"""
Calculate Inside Battery Limits (ISBL) cost.
Sums direct equipment costs and applies the location factor and the
installed cost multiplier ``fc``.
Parameters
----------
fc : float, optional
Installed cost multiplier. Default is 1.0.
print_results : bool, optional
Print a per-equipment cost breakdown. Default is False.
Returns
-------
float
ISBL cost in plant currency.
Raises
------
ValueError
If the plant's country or region is not found in ``locFactors``
and no explicit ``loc_factor`` is set.
"""
def location_factors() -> float:
if self.loc_factor is not None:
return self.loc_factor
if self.country not in self.locFactors:
raise ValueError(
f"Country not found: {self.country}. "
f"Available countries: {list(self.locFactors.keys())}"
)
loc_factor = self.locFactors[self.country]
if isinstance(loc_factor, dict):
if self.region in loc_factor:
return loc_factor[self.region]
else:
raise ValueError(
f"Region not found: {self.region}. "
f"Available regions: {list(loc_factor.keys())}"
)
return loc_factor
self.isbl = (
sum(
equipment.direct_cost
for equipment in self.equipment_list
)
* location_factors()
* fc
* self.exchange_rate
)
if print_results:
# Print the resultS
print("ISBL cost estimation")
print("===================================")
for equipment in self.equipment_list:
print(
f" - {equipment.name}: "
f"{equipment.direct_cost*self.exchange_rate:,.2f} "
f"{self.currency}"
)
print("===================================")
print(f"Total ISBL: {self.isbl:,.2f} {self.currency}")
else:
return self.isbl
[docs]
def calculate_fixed_capital(
self,
fc=None,
additional_capex: bool = False,
print_results=False,
):
"""
Calculate total fixed capital investment.
Includes ISBL, OSBL, design & engineering, and contingency. Factors
can be overridden via ``fixed_capital_factors`` / ``fixed_capital_components``
set on the plant.
Parameters
----------
fc : float or None, optional
Installed cost multiplier. Defaults to 1.0 if None.
additional_capex : bool, optional
Include additional CAPEX items in the printed summary. Default is False.
print_results : bool, optional
Print a cost breakdown. Default is False.
Returns
-------
float
Total fixed capital cost in plant currency.
Raises
------
ValueError
If ``process_type`` is not one of the supported process types.
"""
if fc is None:
self.fc = 1.0
else:
self.fc = fc
self.calculate_isbl(self.fc)
if self.process_type not in self.processTypes:
raise ValueError(
f"Unsupported process_type '{self.process_type}'. "
f"Valid types: {list(self.processTypes)}"
)
params = self.processTypes[self.process_type]
f = {
"osbl": params["OS"],
"de": params["DE"],
"contingency": params["X"],
**{k: v for k, v in self.fixed_capital_factors.items() if v is not None},
}
c = {k: v for k, v in self.fixed_capital_components.items() if v is not None}
self.osbl = c.get("osbl", f["osbl"] * self.isbl)
self.dne = c.get("dne", f["de"] * (self.isbl + self.osbl))
self.contigency = c.get(
"contingency", f["contingency"] * (self.isbl + self.osbl)
)
self.fixed_capital = (
self.isbl
+ self.osbl
+ self.dne
+ self.contigency
)
if print_results:
if (
additional_capex
and self.additional_capex_cost is not None
):
# Print the results
print("Capital cost estimation")
print("===================================")
print(f"ISBL: {self.isbl:,.2f} {self.currency}")
print(f"OSBL: {self.osbl:,.2f} {self.currency}")
print(
f"Design and engineering: {self.dne:,.2f} {self.currency}"
)
print(
f"Contingency: {self.contigency:,.2f} {self.currency}"
)
print(
f"Additional CAPEX: "
f"{sum(self.additional_capex_cost):,.2f} {self.currency}"
)
print("===================================")
total_capex = (
self.fixed_capital
+ sum(self.additional_capex_cost)
)
print(
f"Fixed capital investment: "
f"{total_capex:,.2f} {self.currency}"
)
else:
# Print the results
print("Capital cost estimation")
print("===================================")
print(f"ISBL: {self.isbl:,.2f} {self.currency}")
print(f"OSBL: {self.osbl:,.2f} {self.currency}")
print(
f"Design and engineering: {self.dne:,.2f} {self.currency}"
)
print(
f"Contingency: {self.contigency:,.2f} {self.currency}"
)
print("===================================")
print(
f"Fixed capital investment: "
f"{self.fixed_capital:,.2f} {self.currency}"
)
else:
return self.fixed_capital
[docs]
def calculate_variable_opex(self, print_results=False):
"""
Calculate annual variable operating costs.
Iterates over ``variable_opex_inputs`` and computes cost as
consumption × price × 365 × plant_utilization for each item.
Parameters
----------
print_results : bool, optional
Print a per-item cost breakdown. Default is False.
Returns
-------
float
Total annual variable OPEX in plant currency.
"""
self.variable_production_costs = 0
self.variable_opex_breakdown = {}
for (
item,
details,
) in self.variable_opex_inputs.items():
consumption = details.get("consumption", 0)
price = details.get("price", 0)
cost = (
consumption
* price
* 365
* self.plant_utilization
)
self.variable_opex_breakdown[item] = cost
self.variable_production_costs += cost
if print_results:
print("Variable production costs estimation")
print("===================================")
for (
item,
cost,
) in self.variable_opex_breakdown.items():
item_name = item.replace(
"_", " "
).capitalize()
print(
f" - {item_name}: {cost:,.2f} {self.currency} per year"
)
print("===================================")
print(
f"Total Variable OPEX: "
f"{self.variable_production_costs:,.2f}"
f"{self.currency} per year"
)
else:
return self.variable_production_costs
[docs]
def calculate_revenue(self, print_results=False):
"""
Calculate annual revenue from plant products.
Iterates over ``plant_products`` and computes revenue as
production × price × 365 × plant_utilization for each product.
Parameters
----------
print_results : bool, optional
Print a per-product revenue breakdown. Default is False.
Returns
-------
float
Total annual revenue in plant currency.
"""
self.revenue = 0
self.revenue_breakdown = {}
self.main_product = (
next(iter(self.plant_products))
if self.plant_products
else None
)
for product, details in self.plant_products.items():
production = details.get("production", 0)
price = details.get("price", 0)
revenue = (
production
* price
* 365
* self.plant_utilization
)
self.revenue_breakdown[product] = revenue
self.revenue += revenue
if print_results:
print("Revenue estimation")
print("===================================")
for (
product,
revenue,
) in self.revenue_breakdown.items():
product_name = product.replace(
"_", " "
).capitalize()
print(
f" - {product_name}: {revenue:,.2f} "
f"{self.currency} per year"
)
print("===================================")
print(
f"Total Revenue: {self.revenue:,.2f} {self.currency} per year"
)
else:
return self.revenue
[docs]
def count_process_steps(
self,
equipments,
target_process_types,
excluded_cats=None,
):
"""
Count equipment units matching a set of process types.
Parameters
----------
equipments : list
List of Equipment objects to scan.
target_process_types : set
Process type labels to match (e.g. ``{"Fluids", "Mixed"}``).
excluded_cats : set or None, optional
Equipment categories to skip. Default is None (no exclusions).
Returns
-------
int
Number of matching equipment units.
"""
if excluded_cats is None:
excluded_cats = {}
count = 0
for equipment in equipments:
if (
equipment.process_type
in target_process_types
and equipment.category not in excluded_cats
):
count += 1
return count
[docs]
def calculate_operators_per_shift(
self, no_fluid_process=None, no_solid_process=None
):
"""
Calculate the number of operators required per shift.
Uses the empirical correlation from Turton et al. based on fluid and
solid process step counts. Returns ``operators_per_shift`` directly if
it was set manually on the plant.
Parameters
----------
no_fluid_process : int or None, optional
Number of fluid/mixed process steps. Auto-counted if None.
no_solid_process : int or None, optional
Number of solid/mixed process steps (max 2). Auto-counted if None.
Returns
-------
float
Estimated operators per shift.
Raises
------
ValueError
If ``no_solid_process`` exceeds 2.
"""
if self.operators_per_shift is not None:
return self.operators_per_shift
else:
if no_fluid_process is None:
no_fluid_process = self.count_process_steps(
self.equipment_list,
{"Fluids", "Mixed"},
{"Pumps", "Pressure vessels"},
)
if no_solid_process is None:
no_solid_process = self.count_process_steps(
self.equipment_list,
{"Solids", "Mixed"},
{"Pumps", "Pressure vessels"},
)
if no_solid_process > 2:
raise ValueError(
"Number of solid processes needs "
"to be less than or equal to 2."
)
operators_per_shifts = (
6.29
+ 31.7 * (no_solid_process**2)
+ 0.23 * no_fluid_process
) ** 0.5
return operators_per_shifts
[docs]
def calculate_operators_hired(
self, no_fluid_process=None, no_solid_process=None
):
"""
Calculate the total number of operators to hire.
Accounts for the ratio of operating shifts per year to working shifts
per year. Returns ``operators_hired`` directly if set manually.
Parameters
----------
no_fluid_process : int or None, optional
Number of fluid/mixed process steps. Passed to
``calculate_operators_per_shift`` if needed.
no_solid_process : int or None, optional
Number of solid/mixed process steps. Passed to
``calculate_operators_per_shift`` if needed.
Returns
-------
int
Total operators to hire.
"""
if self.operators_hired is not None:
return self.operators_hired
else:
operators_per_shifts = (
self.calculate_operators_per_shift(
no_fluid_process, no_solid_process
)
)
operating_shifts_per_year = (
365 * self.operating_shifts_per_day
)
working_shifts_per_year = (
self.working_weeks_per_year
* self.working_shifts_per_week
)
operators_hired = math.ceil(
operators_per_shifts
* operating_shifts_per_year
/ working_shifts_per_year
)
return operators_hired
[docs]
def calculate_operating_labor(
self, no_fluid_process=None, no_solid_process=None
):
"""
Calculate total annual operating labor costs.
Parameters
----------
no_fluid_process : int or None, optional
Number of fluid/mixed process steps. Auto-counted if None.
no_solid_process : int or None, optional
Number of solid/mixed process steps. Auto-counted if None.
Returns
-------
float
Annual operating labor cost in plant currency.
"""
operators_hired = self.calculate_operators_hired(
no_fluid_process, no_solid_process
)
working_shifts_per_year = (
self.working_weeks_per_year
* self.working_shifts_per_week
)
working_hours_per_year = working_shifts_per_year * (
24 / self.operating_shifts_per_day
)
rate_cfg = self.operator_hourly_rate
if isinstance(rate_cfg, dict):
rate = rate_cfg.get("rate", 38.11)
else:
rate = (
38.11
if rate_cfg is None
else float(rate_cfg)
)
self.operating_labor_costs = (
operators_hired * working_hours_per_year * rate
)
return self.operating_labor_costs
[docs]
def calculate_fixed_opex(
self, fp=None, print_results=False
):
"""
Calculate fixed operating expenses (OPEX).
Computes supervision, salary overhead, laboratory charges, maintenance,
taxes & insurance, rent, environmental charges, operating supplies,
general plant overhead, working capital interest, patents & royalties,
distribution & selling, and R&D costs. Factors and individual component
values can be overridden via ``fixed_opex_factors`` and
``fixed_opex_components`` set on the plant.
Parameters
----------
fp : float or None, optional
Fixed OPEX multiplier applied to the total. Defaults to 1.0 if None.
print_results : bool, optional
Print a full fixed OPEX breakdown. Default is False.
Returns
-------
float
Total annual fixed OPEX in plant currency.
"""
if fp is None:
self.fp = 1.0
else:
self.fp = fp
self.calculate_fixed_capital(fc=self.fc)
self.calculate_variable_opex()
self.calculate_operating_labor()
_defaults = {
"supervision": 0.25,
"direct_salary_overhead": 0.5,
"laboratory_charges": 0.10,
"maintenance": 0.05,
"taxes_insurance": 0.015,
"rent_of_land": 0.015,
"environmental_charges": 0.01,
"operating_supplies": 0.009,
"general_plant_overhead": 0.65,
"working_capital": 0.15,
"patents_royalties": 0.02,
"distribution_selling": 0.02,
"rnd": 0.03,
}
f = {
**_defaults,
**{k: v for k, v in self.fixed_opex_factors.items() if v is not None},
}
c = {k: v for k, v in self.fixed_opex_components.items() if v is not None}
self.supervision_costs = c.get(
"supervision_costs",
f["supervision"] * self.operating_labor_costs,
)
self.direct_salary_overhead = c.get(
"direct_salary_overhead",
f["direct_salary_overhead"] * (
self.operating_labor_costs + self.supervision_costs
),
)
self.laboratory_charges = c.get(
"laboratory_charges",
f["laboratory_charges"] * self.operating_labor_costs,
)
self.maintenance_costs = c.get(
"maintenance_costs",
f["maintenance"] * self.isbl,
)
self.taxes_insurance_costs = c.get(
"taxes_insurance_costs",
f["taxes_insurance"] * self.isbl,
)
self.rent_of_land_costs = c.get(
"rent_of_land_costs",
f["rent_of_land"] * (self.isbl + self.osbl),
)
self.environmental_charges = c.get(
"environmental_charges",
f["environmental_charges"] * (self.isbl + self.osbl),
)
self.operating_supplies = c.get(
"operating_supplies",
f["operating_supplies"] * self.isbl,
)
self.general_plant_overhead = c.get(
"general_plant_overhead",
f["general_plant_overhead"] * (
self.operating_labor_costs
+ self.supervision_costs
+ self.direct_salary_overhead
),
)
if self.working_capital is not None:
self.interest_working_capital = (
self.working_capital * self.interest_rate
)
else:
self.working_capital = f["working_capital"] * self.fixed_capital
self.interest_working_capital = (
self.working_capital * self.interest_rate
)
self.fixed_production_costs = (
self.operating_labor_costs
+ self.supervision_costs
+ self.direct_salary_overhead
+ self.laboratory_charges
+ self.maintenance_costs
+ self.taxes_insurance_costs
+ self.rent_of_land_costs
+ self.environmental_charges
+ self.operating_supplies
+ self.general_plant_overhead
+ self.interest_working_capital
)
cash_cost_markup = (
f["patents_royalties"]
+ f["distribution_selling"]
+ f["rnd"]
)
cash_cost_of_production = (
self.variable_production_costs
+ self.fixed_production_costs
) / (1 - cash_cost_markup)
self.patents_royalties = c.get(
"patents_royalties",
f["patents_royalties"] * cash_cost_of_production,
)
self.distribution_selling_costs = c.get(
"distribution_selling_costs",
f["distribution_selling"] * cash_cost_of_production,
)
self.RnD_costs = c.get(
"RnD_costs",
f["rnd"] * cash_cost_of_production,
)
self.fixed_production_costs += (
self.patents_royalties
+ self.distribution_selling_costs
+ self.RnD_costs
)
self.fixed_production_costs *= self.fp
if print_results:
# Print the results
print("Fixed production costs estimation")
print("===================================")
print(
f"Operating labor costs: "
f"{self.operating_labor_costs:,.2f} {self.currency} per year"
)
print(
f"Supervision costs: "
f"{self.supervision_costs:,.2f} {self.currency} per year"
)
print(
f"Direct salary overhead: "
f"{self.direct_salary_overhead:,.2f} {self.currency} per year"
)
print(
f"Laboratory charges: "
f"{self.laboratory_charges:,.2f} {self.currency} per year"
)
print(
f"Maintenance costs: "
f"{self.maintenance_costs:,.2f} {self.currency} per year"
)
print(
f"Taxes and insurance costs: "
f"{self.taxes_insurance_costs:,.2f} {self.currency} per year"
)
print(
f"Rent of land costs: "
f"{self.rent_of_land_costs:,.2f} {self.currency} per year"
)
print(
f"Environmental charges: "
f"{self.environmental_charges:,.2f} {self.currency} per year"
)
print(
f"Operating supplies: "
f"{self.operating_supplies:,.2f} {self.currency} per year"
)
print(
f"General plant overhead: "
f"{self.general_plant_overhead:,.2f} {self.currency} per year"
)
print(
f"Interest on working capital: "
f"{self.interest_working_capital:,.2f} "
f"{self.currency} per year"
)
print(
f"Patents and royalties: "
f"{self.patents_royalties:,.2f} {self.currency} per year"
)
print(
f"Distribution and selling costs: "
f"{self.distribution_selling_costs:,.2f} "
f"{self.currency} per year"
)
print(
f"R&D costs: {self.RnD_costs:,.2f} "
f"{self.currency} per year"
)
print("===================================")
print(
f"Fixed OPEX: {self.fixed_production_costs:,.2f} "
f"{self.currency} per year"
)
else:
return self.fixed_production_costs
[docs]
def calculate_cash_flow(
self, print_results: bool = False
):
"""
Build a year-by-year cash flow table.
Applies the CAPEX ramp, production ramp, depreciation schedule,
and tax lag to produce annual capital cost, revenue, cash cost,
gross profit, depreciation, taxable income, tax paid, and net cash
flow arrays. Supports vectorised (Monte Carlo) inputs when
``project_lifetime``, ``interest_rate``, etc. are arrays.
Parameters
----------
print_results : bool, optional
Return a formatted ``pd.DataFrame.style`` for scalar scenarios.
Default is False.
Returns
-------
pd.DataFrame.style or None
Styled cash flow table when ``print_results=True`` and inputs are
scalar; None otherwise (results stored as instance arrays).
Raises
------
ValueError
If ``project_lifetime < 3``, ``capex_ramp`` or
``production_ramp`` are invalid, or no plant products are defined.
"""
# 0) Upstream calcs (capital, opex breakdowns)
self.calculate_fixed_capital(fc=self.fc)
self.calculate_variable_opex()
self.calculate_fixed_opex(fp=self.fp)
self.calculate_revenue()
# --- Normalize shapes ---
lifetime = np.atleast_1d(
self.project_lifetime
).astype(int)
if np.any(lifetime < 3):
raise ValueError(
"All project_lifetime values must be ≥3."
)
n_samples = lifetime.shape[0]
n_years = np.max(lifetime)
fixed_capital = np.atleast_1d(
self.fixed_capital
).astype(float)
fixed_opex = np.atleast_1d(
self.fixed_production_costs
).astype(float)
var_opex = np.atleast_1d(
self.variable_production_costs
).astype(float)
interest = np.atleast_1d(self.interest_rate).astype(
float
)
# Broadcast all scalars to same length
def broadcast(x):
return np.broadcast_to(x, n_samples)
fixed_capital, fixed_opex, var_opex, interest = map(
broadcast,
(fixed_capital, fixed_opex, var_opex, interest),
)
# --- Initialize result arrays ---
shape = (n_samples, n_years)
capex = np.zeros(shape)
main_revenue = np.zeros(shape)
side_revenue = np.zeros(shape)
revenue = np.zeros(shape)
cash_cost = np.zeros(shape)
gross_profit = np.zeros(shape)
depreciation = np.zeros(shape)
taxable_income = np.zeros(shape)
tax_paid = np.zeros(shape)
cash_flow = np.zeros(shape)
prod_array = np.zeros(shape)
# --- Resolve and validate CAPEX ramp ---
if self.capex_ramp is not None:
try:
capex_ramp = np.asarray(
self.capex_ramp, dtype=float
)
except (TypeError, ValueError):
raise ValueError(
"capex_ramp must be a list or array of numbers."
)
if capex_ramp.ndim != 1 or len(capex_ramp) == 0:
raise ValueError(
"capex_ramp must be a non-empty 1-D list or array."
)
if np.any(capex_ramp < 0):
raise ValueError(
"All values in capex_ramp must be >= 0."
)
if not np.isclose(capex_ramp.sum(), 1.0, atol=1e-6):
raise ValueError(
"capex_ramp must sum to 1.0 "
f"(got {capex_ramp.sum():.6f})."
)
if len(capex_ramp) >= n_years:
raise ValueError(
f"capex_ramp has {len(capex_ramp)} entries but "
f"project_lifetime is only {n_years}; at least 1 "
"year must remain for production."
)
else:
capex_ramp = np.array([0.3, 0.6, 0.1])
# --- Resolve and validate production ramp ---
if self.production_ramp is not None:
try:
prod_ramp = np.asarray(
self.production_ramp, dtype=float
)
except (TypeError, ValueError):
raise ValueError(
"production_ramp must be a list or array of numbers."
)
if prod_ramp.ndim != 1 or len(prod_ramp) == 0:
raise ValueError(
"production_ramp must be a non-empty 1-D list "
"or array."
)
if np.any(prod_ramp < 0) or np.any(prod_ramp > 1):
raise ValueError(
"All values in production_ramp must be between "
"0 and 1."
)
if len(prod_ramp) > n_years:
raise ValueError(
f"production_ramp has {len(prod_ramp)} entries "
f"but project_lifetime is only {n_years}."
)
else:
prod_ramp = np.array([0, 0, 0.4, 0.8])
ramp = np.concatenate(
(prod_ramp, np.ones(max(0, n_years - len(prod_ramp))))
)[:n_years]
# --- CAPEX profile + WC draw/release ---
for yr, frac in enumerate(capex_ramp):
if yr < n_years:
capex[:, yr] += fixed_capital * frac
wc_year = len(capex_ramp) - 1
if wc_year < n_years:
capex[:, wc_year] += self.working_capital
capex[:, -1] -= self.working_capital
# --- Add additional CAPEX at specified years ---
if (
self.additional_capex_years is not None
and self.additional_capex_cost is not None
):
self.additional_capex_years = np.atleast_1d(
self.additional_capex_years
).astype(int)
self.additional_capex_cost = np.atleast_1d(
self.additional_capex_cost
).astype(float)
# Check if the number of years matches the number of costs
if (
self.additional_capex_years.shape[0]
!= self.additional_capex_cost.shape[0]
):
raise ValueError(
"The number of additional_capex_years must "
"match the number of additional_capex_costs."
)
for i, year in enumerate(
self.additional_capex_years
):
# Ignore invalid years
if year < 1 or year > n_years:
continue
# Apply only to samples whose lifetime includes this year
alive_mask = lifetime >= year
# Arrays are 0-indexed; NumPy will broadcast the scalar cost
capex[
alive_mask, year - 1
] += self.additional_capex_cost[i]
# --- Production ramp ---
if (
not self.plant_products
or self.main_product is None
):
raise ValueError(
"No plant_products defined; "
"cannot build cash flow / production profile."
)
self.daily_prod = self.plant_products[
self.main_product
]["production"]
nameplate = (
self.daily_prod * 365.0 * self.plant_utilization
)
# --- Revenue & cost arrays ---
for yr in range(n_years):
prod = nameplate * ramp[yr]
prod_array[:, yr] = prod
main_prod_price = self.plant_products[
self.main_product
].get("price")
if main_prod_price is None:
main_revenue[:, yr] = 0
else:
main_revenue[:, yr] = prod * main_prod_price
side_revenue[:, yr] = sum(
self.plant_products[p]["production"]
* 365.0
* self.plant_utilization
* ramp[yr]
* self.plant_products[p].get("price", 0)
for p in self.plant_products
if p != self.main_product
)
revenue[:, yr] = (
main_revenue[:, yr] + side_revenue[:, yr]
)
cash_cost[:, yr] = (
fixed_opex + var_opex * ramp[yr]
)
gross_profit[:, yr] = (
revenue[:, yr] - cash_cost[:, yr]
)
# --- Depreciation (each sample has its own config) ---
dep_cfg = getattr(self, "depreciation", None)
for i in range(n_samples):
capex_dict = {
yr: frac * fixed_capital[i]
for yr, frac in enumerate(capex_ramp)
}
depreciation[i, : lifetime[i]] = (
build_depreciation_array(
project_life=lifetime[i],
capex_by_year=capex_dict,
dep_cfg=dep_cfg,
)
)
# --- Tax and cash flow (with 1-year lag) ---
for yr in range(n_years):
taxable_income[:, yr] = (
gross_profit[:, yr] - depreciation[:, yr]
)
if yr == 0:
tax_paid[:, yr] = 0
else:
prev = taxable_income[:, yr - 1]
tax_paid[:, yr] = np.where(
prev > 0, self.tax_rate * prev, 0
)
cash_flow[:, yr] = (
gross_profit[:, yr]
- tax_paid[:, yr]
- capex[:, yr]
)
# --- Save arrays to instance ---
self.capital_cost_array = capex
self.side_revenue_array = side_revenue
self.main_revenue_array = main_revenue
self.revenue_array = revenue
self.cash_cost_array = cash_cost
self.gross_profit_array = gross_profit
self.depreciation_array = depreciation
self.taxable_income_array = taxable_income
self.tax_paid_array = tax_paid
self.cash_flow = cash_flow
self.prod_array = prod_array
# --- Optional: return formatted summary if scalar case ---
if print_results and n_samples == 1:
years = np.arange(1, n_years + 1)
data = {
"Year": years,
f"Capital cost [{self.currency}]": capex[0],
f"Revenue [{self.currency}]": revenue[0],
f"Cash cost [{self.currency}]": cash_cost[0],
f"Gross profit [{self.currency}]": gross_profit[0],
f"Depreciation [{self.currency}]": depreciation[0],
f"Taxable income [{self.currency}]": taxable_income[0],
f"Tax paid [{self.currency}]": tax_paid[0],
f"Cash flow [{self.currency}]": cash_flow[0],
}
df = pd.DataFrame(data)
fmt = {
c: "{:,.2f}"
for c in df.columns
if c not in ["Year"]
}
return df.style.format(fmt)
[docs]
def calculate_npv(self, print_results: bool = False):
"""
Calculate Net Present Value (NPV) of the project cash flows.
Discounts each year's cash flow at ``interest_rate`` and returns the
cumulative NPV at the end of the project lifetime. Supports vectorised
inputs for Monte Carlo scenarios.
Parameters
----------
print_results : bool, optional
Print a year-by-year present value and cumulative NPV table.
Default is False.
Returns
-------
float or np.ndarray
Final NPV (scalar) or array of NPVs across scenarios.
Raises
------
ValueError
If ``interest_rate`` is an array whose length does not match the
number of cash flow scenarios.
"""
self.calculate_fixed_capital(
fc=1.0 if self.fc is None else self.fc
)
self.calculate_variable_opex()
self.calculate_fixed_opex(
fp=1.0 if self.fp is None else self.fp
)
self.calculate_revenue()
self.calculate_cash_flow()
# Ensure 2D cash_flow: [n_scenarios, n_years]
cf = np.asarray(self.cash_flow, dtype=float)
if cf.ndim == 1:
cf = cf[None, :] # [1, n_years]
n_scenarios, n_years = cf.shape
years = np.arange(1, n_years + 1, dtype=float)
# Interest rate: scalar or per-scenario
r = np.atleast_1d(self.interest_rate).astype(float)
if r.size == 1:
# Same rate for all scenarios
discount_factors = (
1.0 + r[0]
) ** years # [n_years]
else:
if r.size != n_scenarios:
raise ValueError(
"interest_rate must be scalar or have length equal to "
"the number of scenarios in cash_flow."
)
# Per-scenario rates
discount_factors = (1.0 + r)[:, None] ** years[
None, :
] # [n_scenarios, n_years]
# Broadcast division: cf / discount_factors
pv_array = cf / discount_factors
npv_array = np.cumsum(pv_array, axis=-1)
self.pv_array = (
pv_array # shape [n_scenarios, n_years]
)
self.npv_array = (
npv_array # shape [n_scenarios, n_years]
)
final_npv = npv_array[:, -1]
if print_results:
print(
f"Year | "
f"Present Value [{self.currency}] |"
f" Cumulative NPV [{self.currency}]"
)
print("-------------------------------------------")
for year, pv, npv in zip(
range(1, n_years + 1),
pv_array[0],
npv_array[0],
):
print(
f"{year:4d} | {float(pv):15,.2f} | {float(npv):15,.2f}"
)
if final_npv.size == 1:
self.npv = float(final_npv[0])
return self.npv
return final_npv
[docs]
def calculate_levelized_cost(self, print_results=False):
"""
Calculate the levelized cost of production (LCOP).
Discounts capital costs, operating costs, and production over the
project lifetime at ``interest_rate``. Side-product revenues are
subtracted before dividing by discounted production.
Parameters
----------
print_results : bool, optional
Print the mean levelized cost. Default is False.
Returns
-------
float or np.ndarray
Levelized cost per unit of main product (scalar or array).
"""
self.calculate_fixed_capital(
fc=1.0 if self.fc is None else self.fc
)
self.calculate_variable_opex()
self.calculate_fixed_opex(
fp=1.0 if self.fp is None else self.fp
)
self.calculate_revenue()
self.calculate_cash_flow()
is_array = isinstance(self.project_lifetime, (list, np.ndarray))
capital_cost = self.capital_cost_array
prod = self.prod_array
cash_cost = self.cash_cost_array
side_rev = self.side_revenue_array
# ---- VECTOR CASE (Monte Carlo) ----
if is_array:
n_samples = len(self.project_lifetime)
lcop = np.zeros(n_samples)
for i in range(n_samples):
disc_capex = 0.0
disc_opex = 0.0
disc_prod = 0.0
disc_side_rev = 0.0
for year in range(len(cash_cost[i])):
discount_factor = (1 + self.interest_rate[i]) ** (year + 1)
disc_capex += capital_cost[i][year] / discount_factor
disc_opex += cash_cost[i][year] / discount_factor
disc_side_rev += side_rev[i][year] / discount_factor
disc_prod += prod[i][year] / discount_factor
value = (disc_capex + disc_opex - disc_side_rev) / disc_prod
lcop[i] = max(value, 0)
self.levelized_cost = lcop
# ---- SCALAR CASE ----
else:
n_years = int(self.project_lifetime)
disc_capex = 0.0
disc_opex = 0.0
disc_prod = 0.0
disc_side_rev = 0.0
for year in range(n_years):
discount_factor = (1 + self.interest_rate) ** (year + 1)
disc_capex += capital_cost[0][year] / discount_factor
disc_opex += cash_cost[0][year] / discount_factor
disc_side_rev += side_rev[0][year] / discount_factor
disc_prod += prod[0][year] / discount_factor
self.levelized_cost = max(
(disc_capex + disc_opex - disc_side_rev) / disc_prod,
0,
)
if print_results:
print(
f"Levelized cost: {np.mean(self.levelized_cost):,.3f} "
f"{self.currency}/unit"
)
else:
return self.levelized_cost
[docs]
def calculate_payback_time(self, additional_capex: bool = False,
print_results: bool = False):
"""
Calculate simple payback time.
Divides total fixed capital (optionally including additional CAPEX) by
the mean annual cash flow across revenue-generating years.
Parameters
----------
additional_capex : bool, optional
Include additional CAPEX in the total investment. Default is False.
print_results : bool, optional
Print the payback time. Default is False.
Returns
-------
float or np.ndarray
Payback time in years (``nan`` if no revenue-generating years exist).
"""
revenue = np.asarray(self.revenue_array, dtype=float)
cash_flow = np.asarray(self.cash_flow, dtype=float)
is_array = isinstance(self.project_lifetime, (list, np.ndarray))
if is_array:
n_samples = len(self.project_lifetime)
pbt = np.full(n_samples, np.nan, dtype=float)
if (
additional_capex
and self.additional_capex_cost is not None
):
total_fixed_capital = (
np.asarray(self.fixed_capital, dtype=float)
+ np.sum(self.additional_capex_cost)
)
else:
total_fixed_capital = np.asarray(
self.fixed_capital, dtype=float
)
for i in range(n_samples):
revenue_generating_years = cash_flow[i][revenue[i] > 0]
if len(revenue_generating_years) == 0:
pbt[i] = np.nan
else:
average_annual_cash_flow = np.mean(
revenue_generating_years
)
pbt[i] = (
total_fixed_capital[i] / average_annual_cash_flow
if average_annual_cash_flow > 0
else np.nan
)
self.payback_time = pbt
else:
revenue_generating_years = cash_flow[revenue > 0]
if len(revenue_generating_years) == 0:
self.payback_time = float("nan")
else:
if (
additional_capex
and self.additional_capex_cost is not None
):
total_fixed_capital = (
self.fixed_capital
+ sum(self.additional_capex_cost)
)
else:
total_fixed_capital = self.fixed_capital
average_annual_cash_flow = np.mean(
revenue_generating_years
)
self.payback_time = (
total_fixed_capital / average_annual_cash_flow
if average_annual_cash_flow > 0
else float("nan")
)
if print_results:
if np.ndim(self.payback_time) == 0:
print(f"Payback time: {self.payback_time:.2f} years")
else:
print(
f"Payback time: mean = "
f"{np.nanmean(self.payback_time):.2f} years"
)
else:
return self.payback_time
[docs]
def calculate_roi(self, additional_capex: bool = False,
print_results: bool = False):
"""
Calculate Return on Investment (ROI).
Computes total net profit over the project lifetime as a percentage of
total investment (fixed capital + working capital, optionally including
additional CAPEX), annualised by project lifetime.
Parameters
----------
additional_capex : bool, optional
Include additional CAPEX in total investment. Default is False.
print_results : bool, optional
Print the ROI value. Default is False.
Returns
-------
float or np.ndarray
ROI as a percentage (scalar or array across scenarios).
"""
net_profit = (
np.asarray(self.gross_profit_array, dtype=float)
- np.asarray(self.tax_paid_array, dtype=float)
)
is_array = isinstance(self.project_lifetime, (list, np.ndarray))
if is_array:
project_lifetime = np.asarray(
self.project_lifetime, dtype=float
)
fixed_capital = np.asarray(self.fixed_capital, dtype=float)
working_capital = np.asarray(
self.working_capital, dtype=float
)
if (
additional_capex
and self.additional_capex_cost is not None
):
total_investment = (
fixed_capital
+ np.sum(self.additional_capex_cost)
+ working_capital
)
else:
total_investment = fixed_capital + working_capital
annual_profit_sum = np.sum(net_profit, axis=1)
self.roi = (
annual_profit_sum * 100
/ (project_lifetime * total_investment)
)
else:
if (
additional_capex
and self.additional_capex_cost is not None
):
total_investment = (
self.fixed_capital
+ sum(self.additional_capex_cost)
+ self.working_capital
)
else:
total_investment = (
self.fixed_capital + self.working_capital
)
self.roi = (
np.sum(net_profit)
* 100
/ (self.project_lifetime * total_investment)
)
if print_results:
if np.ndim(self.roi) == 0:
print(f"Return of investment: {self.roi:.2f}%")
else:
print(
f"Return of investment: mean = {np.nanmean(self.roi):.2f}%"
)
else:
return self.roi
[docs]
def calculate_irr(self, print_results: bool = False):
"""
Calculate the Internal Rate of Return (IRR).
Finds the discount rate at which NPV equals zero using Brent's method.
Returns ``nan`` if no sign change is found (no valid IRR exists).
Parameters
----------
print_results : bool, optional
Print the IRR value. Default is False.
Returns
-------
float or np.ndarray
IRR as a fraction (e.g. 0.15 = 15%), or ``nan`` if undefined.
"""
cf = np.asarray(self.cash_flow, dtype=float)
def _irr_from_cash_flow(cf_1d):
n = cf_1d.size
if n == 0:
return float("nan")
if not (np.any(cf_1d < 0) and np.any(cf_1d > 0)):
return float("nan")
years = np.arange(n, dtype=float) + 1
def npv_at(r: float) -> float:
if r <= -1.0:
return np.inf
return float(np.sum(cf_1d / (1.0 + r) ** years))
grid = np.concatenate(
[
np.linspace(-0.95, -0.01, 120, endpoint=True),
np.array([0.0]),
np.linspace(0.01, 10.0, 240, endpoint=True),
]
)
npv_vals = np.array([npv_at(r) for r in grid])
bracket = None
for i in range(len(grid) - 1):
a, b = grid[i], grid[i + 1]
fa, fb = npv_vals[i], npv_vals[i + 1]
if not np.isfinite(fa) or not np.isfinite(fb):
continue
if fa == 0.0:
bracket = (a - 1e-6, a + 1e-6)
break
if np.sign(fa) != np.sign(fb):
bracket = (a, b)
break
if bracket is None:
a = 0.01
b = 10.0
fa = npv_at(a)
fb = npv_at(b)
while (
np.isfinite(fb)
and np.sign(fa) == np.sign(fb)
and b < 1000.0
):
b *= 1.5
fb = npv_at(b)
if np.isfinite(fb) and np.sign(fa) != np.sign(fb):
bracket = (a, b)
if bracket is None:
return float("nan")
try:
sol = root_scalar(
npv_at,
bracket=bracket,
method="brentq",
xtol=1e-10,
rtol=1e-10,
maxiter=200,
)
return (
sol.root
if sol.converged and math.isfinite(sol.root)
else float("nan")
)
except Exception:
return float("nan")
if cf.ndim == 1:
self.irr = _irr_from_cash_flow(cf)
else:
irr_vals = np.array(
[_irr_from_cash_flow(cf_i) for cf_i in cf],
dtype=float,
)
if irr_vals.size == 1:
self.irr = float(irr_vals[0])
else:
self.irr = irr_vals
if print_results:
if np.isscalar(self.irr):
if math.isfinite(self.irr):
print(
f"Internal Rate of Return: {self.irr * 100:.2f}%"
)
else:
print("Internal Rate of Return: undefined")
else:
finite_vals = self.irr[np.isfinite(self.irr)]
if len(finite_vals) > 0:
print(
f"Internal Rate of Return: mean = "
f"{np.mean(finite_vals) * 100:.2f}%"
)
else:
print("Internal Rate of Return: undefined")
else:
return self.irr
[docs]
def calculate_all(self, additional_capex=False, print_results=False):
"""
Run all financial calculations sequentially.
Calls ``calculate_fixed_capital``, ``calculate_variable_opex``,
``calculate_fixed_opex``, ``calculate_revenue``, ``calculate_cash_flow``,
``calculate_npv``, ``calculate_levelized_cost``,
``calculate_payback_time``, ``calculate_roi``, and ``calculate_irr``.
Parameters
----------
additional_capex : bool, optional
Pass through to ``calculate_fixed_capital``, ``calculate_payback_time``,
and ``calculate_roi``. Default is False.
print_results : bool, optional
Print results from each sub-calculation. Default is False.
"""
self.calculate_fixed_capital(fc=self.fc,
additional_capex=additional_capex,
print_results=print_results)
self.calculate_variable_opex(print_results=print_results)
self.calculate_fixed_opex(fp=self.fp, print_results=print_results)
self.calculate_revenue(print_results=print_results)
self.calculate_cash_flow(print_results=print_results)
self.calculate_npv(print_results=print_results)
self.calculate_levelized_cost(print_results=print_results)
self.calculate_payback_time(additional_capex=additional_capex,
print_results=print_results)
self.calculate_roi(additional_capex=additional_capex,
print_results=print_results)
self.calculate_irr(print_results=print_results)
[docs]
def to_dict(self):
"""
Serialize plant configuration and all computed metrics to a dict.
Returns
-------
dict
Nested dictionary with sections: ``plant_configuration``,
``equipment_summary``, ``capital_costs``, ``variable_opex``,
``fixed_opex``, ``revenue``, ``cash_flow``, and ``metrics``.
"""
equipment_items = []
for eq in self.equipment_list:
equipment_items.append({
"name": getattr(eq, "name", None),
"category": getattr(eq, "category", None),
"type": getattr(eq, "type", None),
"num_units": int(getattr(eq, "num_units", 1)),
"purchase_cost": float(getattr(eq, "purchase_cost", 0.0)),
"direct_cost": float(getattr(eq, "direct_cost", 0.0)),
})
plant_dict = {
"plant_configuration": {
"plant_name": self.name,
"process_type": self.process_type,
"country": self.country,
"region": self.region,
"currency": getattr(self, "currency", "USD"),
"exchange_rate": getattr(self, "exchange_rate", 1.0),
"interest_rate": self.interest_rate,
"project_lifetime": self.project_lifetime,
"plant_utilization": self.plant_utilization,
"tax_rate": self.tax_rate,
"operator_hourly_rate": deepcopy(self.operator_hourly_rate),
"operators_per_shift": self.operators_per_shift,
"operators_hired": self.operators_hired,
"working_weeks_per_year": self.working_weeks_per_year,
"working_shifts_per_week": self.working_shifts_per_week,
"operating_shifts_per_day": self.operating_shifts_per_day,
"plant_products": deepcopy(self.plant_products),
"variable_opex_inputs": deepcopy(self.variable_opex_inputs),
"working_capital": self.working_capital,
"additional_capex_cost": deepcopy(
self.additional_capex_cost.tolist()
if isinstance(self.additional_capex_cost, np.ndarray)
else self.additional_capex_cost
) if self.additional_capex_cost is not None else None,
"additional_capex_years": deepcopy(
self.additional_capex_years.tolist()
if isinstance(self.additional_capex_years, np.ndarray)
else self.additional_capex_years
) if self.additional_capex_years is not None else None,
"fc": self.fc,
"fp": self.fp,
"depreciation": deepcopy(self.depreciation),
},
"equipment_summary": {
"count": len(equipment_items),
"items": equipment_items,
},
"capital_costs": {
"isbl": float(getattr(self, "isbl", 0.0)),
"osbl": float(getattr(self, "osbl", 0.0)),
"design_and_engineering": float(getattr(self, "dne", 0.0)),
"contingency": float(getattr(self, "contigency", 0.0)),
"fixed_capital": float(getattr(self, "fixed_capital", 0.0)),
"working_capital": float(getattr(self, "working_capital", 0.0))
if self.working_capital is not None else None,
"additional_capex_cost": (
self.additional_capex_cost.tolist()
if isinstance(self.additional_capex_cost, np.ndarray)
else self.additional_capex_cost
),
"additional_capex_years": (
self.additional_capex_years.tolist()
if isinstance(self.additional_capex_years, np.ndarray)
else self.additional_capex_years
),
},
"variable_opex": {
"breakdown": deepcopy(
getattr(self, "variable_opex_breakdown", {})
),
"total": float(
getattr(self, "variable_production_costs", 0.0)
),
},
"fixed_opex": {
"operating_labor": float(
getattr(self, "operating_labor_costs", 0.0)
),
"supervision": float(getattr(self, "supervision_costs", 0.0)),
"direct_salary_overhead": float(
getattr(self, "direct_salary_overhead", 0.0)
),
"laboratory_charges": float(
getattr(self, "laboratory_charges", 0.0)
),
"maintenance": float(
getattr(self, "maintenance_costs", 0.0)
),
"taxes_insurance": float(
getattr(self, "taxes_insurance_costs", 0.0)
),
"rent_of_land": float(
getattr(self, "rent_of_land_costs", 0.0)
),
"environmental_charges": float(
getattr(self, "environmental_charges", 0.0)
),
"operating_supplies": float(
getattr(self, "operating_supplies", 0.0)
),
"general_plant_overhead": float(
getattr(self, "general_plant_overhead", 0.0)
),
"interest_working_capital": float(
getattr(self, "interest_working_capital", 0.0)
),
"patents_royalties": float(
getattr(self, "patents_royalties", 0.0)
),
"distribution_selling": float(
getattr(self, "distribution_selling_costs", 0.0)
),
"rnd": float(getattr(self, "RnD_costs", 0.0)),
"total": float(getattr(self, "fixed_production_costs", 0.0)),
},
"revenue": {
"main_product": getattr(self, "main_product", None),
"breakdown": deepcopy(getattr(self, "revenue_breakdown", {})),
"total": float(getattr(self, "revenue", 0.0)),
},
"cash_flow": {
"cash_flow": getattr(self, "cash_flow", None).tolist()
if hasattr(self, "cash_flow") and self.cash_flow is not None
else None
},
"metrics": {
"levelized_cost": float(getattr(self, "levelized_cost", 0.0))
if hasattr(self, "levelized_cost") else None,
"npv": float(self.calculate_npv())
if hasattr(self, "cash_flow") else None,
"roi": float(getattr(self, "roi", 0.0))
if hasattr(self, "roi") else None,
"payback_time": float(getattr(self, "payback_time", 0.0))
if hasattr(self, "payback_time") else None,
"irr": float(getattr(self, "irr", 0.0))
if hasattr(self, "irr") else None,
},
}
additional_capex_cost = getattr(self, "additional_capex_cost", None)
if additional_capex_cost:
self.calculate_roi(additional_capex=True)
self.calculate_payback_time(additional_capex=True)
plant_dict["metrics"]["roi_with_additional_capex"] = float(
getattr(self, "roi", None)
) if hasattr(self, "roi") else None
plant_dict["metrics"][
"payback_time_with_additional_capex"
] = (
float(getattr(self, "payback_time", None))
if hasattr(self, "payback_time")
else None
)
return plant_dict
def __str__(self):
"""Pretty string representation of all plant configuration inputs."""
# Helper for formatting dicts cleanly
import json
def fmt(obj):
if obj is None:
return "None"
if isinstance(obj, dict):
return json.dumps(obj, indent=4)
return str(obj)
# Equipment formatting
if self.equipment_list:
eq_strings = []
for i, eq in enumerate(self.equipment_list):
label = getattr(
eq,
"name",
f"{eq.__class__.__name__}({i})",
)
cost = getattr(eq, "direct_cost", "N/A")
eq_strings.append(
f" - {label}: direct_cost={cost}"
)
eq_block = "\n".join(eq_strings)
else:
eq_block = " None"
return (
f"ProcessPlant Configuration\n"
f"{'-'*40}\n"
f"Plant Name: {self.name}\n"
f"Process Type: {self.process_type}\n"
f"Country / Region: {self.country} / {self.region}\n"
f"Interest Rate: {self.interest_rate}\n"
f"Project Lifetime (years): {self.project_lifetime}\n"
f"Plant Utilization: {self.plant_utilization}\n"
f"Tax Rate: {self.tax_rate}\n"
f"Working Capital: {self.working_capital}\n"
f"Depreciation Settings: {fmt(self.depreciation)}\n"
f"\n"
f"Operator Labor Inputs\n"
f" Hourly Rate: {fmt(self.operator_hourly_rate)}\n"
f" Operators per Shift: {self.operators_per_shift}\n"
f" Operators Hired: {self.operators_hired}\n"
f" Working Weeks / Year: {self.working_weeks_per_year}\n"
f" Working Shifts / Week: {self.working_shifts_per_week}\n"
f" Operating Shifts / Day: {self.operating_shifts_per_day}\n"
f"\n"
f"Products\n"
f"{fmt(self.plant_products)}\n"
f"\n"
f"Variable OPEX Inputs:\n{fmt(self.variable_opex_inputs)}\n"
f"\n"
f"Additional CAPEX:\n"
f" Years: {self.additional_capex_years}\n"
f" Costs: {self.additional_capex_cost}\n"
f"\n"
f"Equipment List:\n{eq_block}\n"
f"\n"
f"Cost Multipliers:\n"
f" fc (installed cost factor): {self.fc}\n"
f" fp (fixed OPEX factor): {self.fp}\n"
)
# Depreciation models
DepMethod = Literal[
"straight_line", "declining_balance", "macrs"
]
# MACRS half-year convention percentage tables (IRS Pub 946).
# https://www.irs.gov/pub/irs-pdf/p946.pdf
# Values are FRACTIONS (not %). Sum to 1.0 within rounding.
_MACRS_HALF_YEAR: Dict[int, List[float]] = {
3: [0.3333, 0.4445, 0.1481, 0.0741],
5: [0.2000, 0.3200, 0.1920, 0.1152, 0.1152, 0.0576],
7: [
0.1429,
0.2449,
0.1749,
0.1249,
0.0893,
0.0892,
0.0893,
0.0446,
],
10: [
0.1000,
0.1800,
0.1440,
0.1152,
0.0922,
0.0737,
0.0655,
0.0655,
0.0656,
0.0328,
],
15: [
0.0500,
0.0950,
0.0855,
0.0770,
0.0693,
0.0623,
0.0590,
0.0590,
0.0591,
0.0590,
0.0591,
0.0590,
0.0591,
0.0590,
0.0591,
0.0295,
],
20: [
0.0375,
0.07219,
0.06677,
0.06177,
0.05713,
0.05285,
0.04888,
0.04522,
0.04462,
0.04461,
0.04462,
0.04461,
0.04462,
0.04461,
0.04462,
0.04461,
0.04462,
0.04461,
0.04462,
0.04461,
0.02231,
],
}
[docs]
class DepreciationConfig:
"""
Configuration for asset depreciation calculations.
This class defines the parameters needed to compute depreciation
using various methods.
Attributes:
method (DepMethod): The depreciation method to use.
Defaults to "straight_line".
Options: "straight_line", "declining_balance", "macrs".
life (Optional[int]): The useful life of the asset in years.
Used by straight_line and declining_balance methods.
Defaults to None.
db_factor (float): The declining balance factor (multiplier).
Only used by the declining_balance method. Defaults to 2.0.
salvage_fraction (float): The salvage value as a fraction
of the initial cost.
Used by straight_line and declining_balance methods.
Defaults to 0.0.
macrs_class (int): The MACRS property class (1-20).
Only used by the macrs method. Defaults to 7.
convention (str): The depreciation convention for MACRS.
Only used by the macrs method. Defaults to "half_year".
service_start_year (int): The year index (starting from 0)
when the asset is placed in service. Defaults to 2.
"""
method: DepMethod = "straight_line"
life: Optional[int] = (
None # straight_line / declining_balance
)
db_factor: float = 2.0 # declining_balance only
salvage_fraction: float = (
0.0 # straight_line / declining_balance only
)
macrs_class: int = 7 # macrs only
convention: str = "half_year" # macrs only
service_start_year: int = (
2 # year index when asset is placed in service
)
_UNCERTAINTY_KEYS = {
"fixed_capital_factor",
"fixed_opex_factor",
"project_lifetime",
"interest_rate",
"plant_utilization",
"tax_rate",
}
_UNCERTAINTY_SUB_KEYS = {"std", "min", "max"}
# Parameters whose values must stay within [0, 1]
_UNIT_INTERVAL_PARAMS = {"plant_utilization", "tax_rate"}
def _validate_project_uncertainties(cfg: dict) -> None:
"""
Validate the structure and values of a ``project_uncertainties`` config dict.
Parameters
----------
cfg : dict
Uncertainty configuration mapping parameter names to sub-dicts with
keys such as ``mean``, ``std``, ``min``, ``max``.
Raises
------
TypeError
If ``cfg`` is not a dict, or any sub-entry is not a dict, or any
numeric value is not an int or float.
ValueError
If unknown parameter or sub-keys are present, ``std`` is negative,
``min >= max``, or domain-specific bounds are violated (e.g. interest
rate ≤ 0, project lifetime < 1, unit-interval params outside [0, 1]).
"""
if not cfg:
return
if not isinstance(cfg, dict):
raise TypeError(
"'project_uncertainties' must be a dict, "
f"got {type(cfg).__name__}."
)
unknown = set(cfg) - _UNCERTAINTY_KEYS
if unknown:
raise ValueError(
f"Unknown key(s) in 'project_uncertainties': {sorted(unknown)}. "
f"Valid keys: {sorted(_UNCERTAINTY_KEYS)}."
)
for param, sub in cfg.items():
if not isinstance(sub, dict):
raise TypeError(
f"'project_uncertainties['{param}']' must be a dict, "
f"got {type(sub).__name__}."
)
unknown_sub = set(sub) - _UNCERTAINTY_SUB_KEYS
if unknown_sub:
raise ValueError(
f"Unknown key(s) in 'project_uncertainties['{param}']': "
f"{sorted(unknown_sub)}. "
f"Valid keys: {sorted(_UNCERTAINTY_SUB_KEYS)}."
)
for key, val in sub.items():
if not isinstance(val, (int, float)):
raise TypeError(
f"'project_uncertainties['{param}']['{key}']' must be "
f"a number, got {type(val).__name__}."
)
if "std" in sub and sub["std"] < 0:
raise ValueError(
f"'project_uncertainties['{param}']['std']' must be ≥ 0, "
f"got {sub['std']}."
)
if "min" in sub and "max" in sub and sub["min"] >= sub["max"]:
raise ValueError(
f"'project_uncertainties['{param}']': "
f"'min' ({sub['min']}) must be less than 'max' ({sub['max']})."
)
if param in ("fixed_capital_factor", "fixed_opex_factor"):
for bound in ("min", "max"):
if bound in sub and sub[bound] <= 0:
raise ValueError(
f"'project_uncertainties['{param}']['{bound}']' "
f"must be > 0, got {sub[bound]}."
)
if param == "interest_rate":
for bound in ("min", "max"):
if bound in sub and sub[bound] <= 0:
raise ValueError(
f"'project_uncertainties['interest_rate']['{bound}']' "
f"must be > 0, got {sub[bound]}."
)
if param == "project_lifetime":
for bound in ("min", "max"):
if bound in sub and sub[bound] < 1:
raise ValueError(
f"'project_uncertainties['project_lifetime']"
f"['{bound}']' must be ≥ 1, got {sub[bound]}."
)
if param in _UNIT_INTERVAL_PARAMS:
for bound in ("min", "max"):
if bound in sub and not (0 <= sub[bound] <= 1):
raise ValueError(
f"'project_uncertainties['{param}']['{bound}']' "
f"must be between 0 and 1, got {sub[bound]}."
)
def _normalize_dep_config(
project_life: int, dep_cfg: Optional[dict]
) -> DepreciationConfig:
"""
Normalize and validate a depreciation configuration.
This function creates a DepreciationConfig object from a dictionary of
configuration parameters, applies sensible defaults, and validates
the configuration based on the depreciation method and project life.
Args:
project_life (int): The expected life of the project in years.
Used to set a sensible default for the depreciation life
if not specified.
dep_cfg (Optional[dict]): A dictionary containing depreciation
configuration parameters. Keys should correspond to
DepreciationConfig attributes. If None, defaults are applied.
Returns:
DepreciationConfig: A validated depreciation configuration object
with all required parameters set.
Raises:
ValueError: If the MACRS convention is not "half_year" when
using the "macrs" depreciation method.
ValueError: If the specified MACRS class is not supported.
Only classes defined in _MACRS_HALF_YEAR are accepted.
Notes:
- If cfg.life is not specified, it defaults to the minimum of
project_life and 15 years.
- Only the "half_year" MACRS convention is currently supported.
- MACRS class validation only occurs when the depreciation
method is "macrs".
"""
cfg = DepreciationConfig()
if dep_cfg:
for k, v in dep_cfg.items():
if hasattr(cfg, k):
setattr(cfg, k, v)
# Sensible defaults
if cfg.life is None:
cfg.life = min(project_life, 15)
if cfg.method == "macrs":
if cfg.convention != "half_year":
raise ValueError(
"Only half_year MACRS convention is supported currently."
)
if cfg.macrs_class not in _MACRS_HALF_YEAR:
raise ValueError(
f"Unsupported MACRS class {cfg.macrs_class}. "
f"Choose one of {sorted(_MACRS_HALF_YEAR.keys())}."
)
return cfg
def _straight_line_schedule(
basis: float,
life: int,
salvage_frac: float,
horizon: int,
) -> np.ndarray:
"""
Calculate a straight-line depreciation schedule over a given horizon.
This function computes an annual depreciation amount based
on the asset basis, useful life, and salvage value, then
creates a schedule array that distributes this depreciation a
cross the analysis horizon.
Args:
basis: The initial cost or basis of the asset.
life: The useful life of the asset in years.
salvage_frac: The salvage value as a fraction of the basis (0 to 1).
horizon: The analysis horizon in years.
Returns:
A numpy array of shape (horizon,) containing the annual depreciation
amounts. The array is zero-filled for years beyond the asset's useful
life.
Notes:
- Depreciation is distributed equally across the asset's useful life.
- A rounding correction is applied to the final depreciation year to
ensure the sum of the schedule equals the total depreciable amount.
"""
salvage = basis * salvage_frac
dep_total = basis - salvage
annual = dep_total / life
sched = np.zeros(horizon, dtype=float)
years = min(life, horizon)
sched[:years] = annual
# Small rounding fix to ensure sum equals dep_total
diff = dep_total - sched.sum()
if abs(diff) > 1e-6 and years > 0:
sched[years - 1] += diff
return sched
def _declining_balance_schedule(
basis: float,
life: int,
factor: float,
salvage_frac: float,
horizon: int,
) -> np.ndarray:
"""
Calculate a declining balance depreciation schedule with salvage
value protection.
This function computes a depreciation schedule using a declining
balance method that switches to straight-line depreciation
when beneficial, ensuring the asset depreciates from its basis
to its salvage value over the specified life.
Parameters
----------
basis : float
The initial cost or book value of the asset.
basis : float
The initial cost or book value of the asset.
life : int
The useful life of the asset in years.
factor : float
The declining balance factor
(typically 1 or 2 for standard or double declining balance).
salvage_frac : float
The salvage value as a fraction of the basis
(e.g., 0.1 for 10% salvage).
horizon : int
The time horizon in years for which to generate the schedule.
Returns
-------
np.ndarray
A 1D array of shape (horizon,) containing the depreciation
amount for each year. Values are zero for years beyond
the asset's life.
Notes
-----
- The depreciation method automatically switches between declining balance
and straight-line when straight-line yields a higher depreciation
amount.
- The schedule respects the salvage value, preventing
depreciation below it.
- A rounding correction is applied to ensure the total depreciation equals
(basis - salvage) within numerical precision.
"""
salvage = basis * salvage_frac
remaining = basis
sched = np.zeros(horizon, dtype=float)
for y in range(min(life, horizon)):
# Candidate DB amount
db = remaining * (factor / life)
# Candidate SL amount on remaining (including salvage protection)
years_left = life - y
sl_total_left = max(0.0, remaining - salvage)
sl = (
sl_total_left / years_left
if years_left > 0
else 0.0
)
dep = max(
0.0, min(max(db, sl), remaining - salvage)
) # cannot dip below salvage
sched[y] = dep
remaining -= dep
# Tiny rounding correction
diff = (basis - salvage) - sched.sum()
if abs(diff) > 1e-6:
last = (
np.nonzero(sched)[0][-1] if sched.any() else 0
)
sched[last] += diff
return sched
def _macrs_schedule(
basis: float, macrs_class: int, horizon: int
) -> np.ndarray:
"""
Generate a MACRS depreciation schedule for an asset.
This function calculates the annual depreciation amounts using the Modified
Accelerated Cost Recovery System (MACRS) half-year convention over a
specified time horizon.
Args:
basis (float): The initial cost basis of the asset to be depreciated.
macrs_class (int): The MACRS asset class that determines the
depreciation percentages and recovery period.
horizon (int): The number of years over which to generate the
depreciation schedule.
Returns:
np.ndarray: An array of depreciation amounts for each year
in the horizon. The sum of all depreciation amounts
equals the basis (within floating-point tolerance).
Notes:
- If the standard MACRS schedule is shorter than the horizon,
the schedule is padded with zeros for remaining years.
- If the standard MACRS schedule is longer than the horizon,
it is truncated.
- A rounding adjustment is applied to the final year to ensure
the total depreciation does not exceed the basis.
"""
pct = _MACRS_HALF_YEAR[macrs_class]
sched = np.array(pct, dtype=float) * basis
if len(sched) < horizon:
sched = np.pad(sched, (0, horizon - len(sched)))
else:
sched = sched[:horizon]
# Rounding fix to make sure we don't exceed basis:
if sched.sum() - basis > 1e-6:
sched[-1] -= sched.sum() - basis
return sched
[docs]
def build_depreciation_array(
project_life: int,
capex_by_year: Dict[int, float],
dep_cfg: Optional[dict] = None,
) -> np.ndarray:
"""
Build a depreciation schedule array over the project lifecycle.
Calculates annual depreciation amounts for capital expenditures using the
specified depreciation method. Supports multiple depreciation methods
including straight-line,
declining balance, and MACRS.
Parameters
----------
project_life : int
The total duration of the project in years.
capex_by_year : Dict[int, float]
Dictionary mapping year to capital expenditure amount for that year.
dep_cfg : Optional[dict], optional
Depreciation configuration dictionary containing method, life,
salvage fraction, and method-specific parameters.
If None, uses normalized default configuration.
Default is None.
Returns
-------
np.ndarray
1D array of shape (project_life,) containing annual depreciation
amounts. Values are floats representing depreciation in each year.
Raises
------
ValueError
If the depreciation method specified in dep_cfg is not one of the
supported methods: 'straight_line', 'declining_balance', or 'macrs'.
Notes
-----
- Capital expenditures are placed in service starting at the configured
service start year.
- Depreciation schedules respect the project horizon after placement
in service.
- Zero amounts and expired horizons are skipped without error.
"""
cfg = _normalize_dep_config(project_life, dep_cfg)
dep = np.zeros(project_life, dtype=float)
for capex_year, amount in capex_by_year.items():
# place-in-service timing
start = max(cfg.service_start_year, capex_year)
horizon = max(0, project_life - start)
if horizon <= 0 or amount == 0:
continue
if cfg.method == "straight_line":
sched = _straight_line_schedule(
amount,
cfg.life,
cfg.salvage_fraction,
horizon,
)
elif cfg.method == "declining_balance":
sched = _declining_balance_schedule(
amount,
cfg.life,
cfg.db_factor,
cfg.salvage_fraction,
horizon,
)
elif cfg.method == "macrs":
sched = _macrs_schedule(
amount, cfg.macrs_class, horizon
)
else:
raise ValueError(
f"Unknown depreciation method: {cfg.method}"
)
dep[start: start + len(sched)] += sched
return dep