diff --git a/src/masterdata/__pycache__/service.cpython-311.pyc b/src/masterdata/__pycache__/service.cpython-311.pyc index 5860f01..047d614 100644 Binary files a/src/masterdata/__pycache__/service.cpython-311.pyc and b/src/masterdata/__pycache__/service.cpython-311.pyc differ diff --git a/src/masterdata/service.py b/src/masterdata/service.py index 87da8c7..4821b63 100644 --- a/src/masterdata/service.py +++ b/src/masterdata/service.py @@ -36,7 +36,7 @@ async def get_all( *, db_session: DbSession, items_per_page: int, search: str = None, common ) -> list[MasterData]: """Returns all documents.""" - query = Select(MasterData).order_by(MasterData.description.asc()) + query = Select(MasterData) if search: query = query.filter(MasterData.name.ilike(f"%{search}%")) diff --git a/src/modules/equipment/formula.py b/src/modules/equipment/formula.py new file mode 100644 index 0000000..849d964 --- /dev/null +++ b/src/modules/equipment/formula.py @@ -0,0 +1,193 @@ +""" +Collected formulas and helpers extracted from module algorithms. + +This file consolidates the core mathematical/financial formulas used across: +- `Eac.py` (NPV, PMT, discounted projections, EAC calculation) +- `insert_actual_data.py` (aggregation formulas, man-hour conversion) +- `Prediksi.py` (future value / fv wrappers) + +Keep these functions pure and well-documented to make debugging and +comparisons easier. +""" +from typing import Iterable, List, Sequence +import math +import numpy as np +try: + import numpy_financial as npf +except Exception: # numpy_financial may not be installed in all envs + npf = None + + +def npv_from_cash_flows(cash_flows: Sequence[float], rate: float, offset: int = 0) -> float: + """Compute discounted sum (NPV) of `cash_flows` at `rate` with optional time offset. + + Formula used in `Eac.py`: + NPV = Σ [Ct / (1 + r)^(offset + t)] + where t is 1-based index of flows in `cash_flows`. + + Args: + cash_flows: sequence of cash flows (Ct) for consecutive periods. + rate: periodic discount/inflation rate (e.g., 0.05 for 5%). + offset: integer number of periods to offset (e.g., last_seq from actuals). + + Returns: + Discounted present value as float. + """ + if not cash_flows: + return 0.0 + total = 0.0 + for i, value in enumerate(cash_flows): + exp = offset + i + 1 + total += float(value) / ((1.0 + rate) ** exp) + return float(total) + + +def pmt_formula(rate: float, nper: int, pv: float) -> float: + """Compute level payment PMT for given rate, periods and present value. + + Uses the standard annuity payment formula (same as numpy_financial.pmt but + returned value uses same sign-convention as mathematical formula): + + PMT = PV * [r (1+r)^n] / [(1+r)^n - 1] + + Special-case when rate == 0 (equal division): PMT = PV / nper + + Note: many code places use sign flips (e.g. `-npf.pmt(...)`) because of + numpy_financial's cashflow sign policy. Use whichever sign convention your + caller expects. + + Args: + rate: periodic rate (decimal) + nper: number of periods (int) + pv: present value to amortize + + Returns: + payment per period (float) + """ + if nper == 0: + raise ValueError("nper must be > 0") + if abs(rate) < 1e-12: + return float(pv) / float(nper) + r = float(rate) + n = int(nper) + numerator = pv * r * ((1 + r) ** n) + denominator = ((1 + r) ** n) - 1 + return float(numerator / denominator) + + +def pmt(rate: float, nper: int, pv: float) -> float: + """Wrapper to compute PMT using `numpy_financial` when available, else fall back to formula. + + Returns the value that `numpy_financial.pmt` would return (i.e. typically + negative for positive PV by that library's sign convention). The caller + can negate the result to get a positive payment if desired (as seen in + `Eac.py` where they do `-npf.pmt(...)`). + """ + if npf is not None: + return float(npf.pmt(rate, nper, pv)) + # numpy_financial returns payments with sign convention; our fallback + # returns payment with the same absolute magnitude but as negative (to + # match npf behavior for positive PV) so callers that negate will get the + # same sign semantics. + res = pmt_formula(rate, nper, pv) + return -res + + +def fv(rate: float, nper: int, pmt: float, pv: float) -> float: + """Future value wrapper to `numpy_financial.fv` when available. + + Used in `Prediksi.py` as `npf.fv(rate, nper + i, pmt, pv)` to compute + forecasting values. + """ + if npf is not None: + return float(npf.fv(rate, nper, pmt, pv)) + # fallback: compute iteratively + r = float(rate) + n = int(nper) + fv_val = -pv + # apply periods + for _ in range(n): + fv_val = fv_val * (1 + r) + (-pmt) + return float(fv_val) + + +def discounted_projection_sum(values: Sequence[float], inflation_rate: float, last_seq: int = 0) -> float: + """Discount a sequence of projected `values` using `inflation_rate` with an initial offset `last_seq`. + + This mirrors the calculation in `Eac.py` where projected flows are discounted + starting at `last_seq + 1`. + """ + return npv_from_cash_flows(values, inflation_rate, offset=last_seq) + + +def rc_labor_cost(interval: float, labor_time: float, labor_human: float, man_hour_rate: float) -> float: + """Compute labor cost for corrective/pm/oh rows. + + Formula from `Prediksi.py` / `insert_actual_data.py`: + rc_xxx_labor_cost = (interval * labor_time * labor_human * man_hour_rate) + """ + return float(interval) * float(labor_time) * float(labor_human) * float(man_hour_rate) + + +def rc_lost_cost(raw_loss_output_mw: float, raw_loss_output_price: float, cm_interval: float) -> float: + """Compute lost production cost as in `Prediksi.py`. + + Formula used: + rc_lost_cost = (raw_loss_output_MW * raw_loss_output_price * raw_cm_interval) * 1000 + The multiplication by 1000 appears in source and is preserved here. + """ + return float(raw_loss_output_mw) * float(raw_loss_output_price) * float(cm_interval) * 1000.0 + + +def rc_total_cost(**components) -> float: + """Sum all provided components into a total RC cost. + + The `Eac.py` / `Prediksi.py` code builds a `rc_total_cost` as the sum of many + parts; this helper simply sums values passed as keyword args (coalescing + None to 0.0) to make testing/comparison easier. + """ + total = 0.0 + for k, v in components.items(): + try: + total += 0.0 if v is None else float(v) + except Exception: + # ignore non-numeric components + continue + return float(total) + + +def seconds_to_hours(seconds: float, ndigits: int = 2) -> float: + """Convert seconds to hours and round to `ndigits` decimals. + + This mirrors usage like: + ROUND(SUM(EXTRACT(EPOCH FROM (actfinish - actstart)) / 3600), 2) + in SQL queries seen in `insert_actual_data.py`. + """ + hours = float(seconds) / 3600.0 + return float(round(hours, ndigits)) + + +def man_count_from_labtrans(count: int) -> int: + """Return man_count as per SQL CASE: if count == 0 -> 3 else count. + + Implemented to match the SQL "CASE WHEN COUNT(b.laborcode) = 0 THEN 3 ELSE COUNT(b.laborcode)". + """ + try: + c = int(count) + except Exception: + c = 0 + return 3 if c == 0 else c + + +__all__ = [ + "npv_from_cash_flows", + "pmt_formula", + "pmt", + "fv", + "discounted_projection_sum", + "rc_labor_cost", + "rc_lost_cost", + "rc_total_cost", + "seconds_to_hours", + "man_count_from_labtrans", +] diff --git a/src/plant_transaction_data/__pycache__/service.cpython-311.pyc b/src/plant_transaction_data/__pycache__/service.cpython-311.pyc index 3c5ab9d..d032b24 100644 Binary files a/src/plant_transaction_data/__pycache__/service.cpython-311.pyc and b/src/plant_transaction_data/__pycache__/service.cpython-311.pyc differ diff --git a/src/plant_transaction_data/service.py b/src/plant_transaction_data/service.py index 29f5717..f811a09 100644 --- a/src/plant_transaction_data/service.py +++ b/src/plant_transaction_data/service.py @@ -4,8 +4,8 @@ import logging from subprocess import PIPE from sqlalchemy import Select, Delete, cast, String -from .model import PlantTransactionData -from .schema import PlantTransactionDataCreate, PlantTransactionDataUpdate +from src.plant_transaction_data.model import PlantTransactionData +from src.plant_transaction_data.schema import PlantTransactionDataCreate, PlantTransactionDataUpdate from src.database.service import search_filter_sort_paginate from typing import Optional @@ -67,12 +67,12 @@ async def get_charts( for idx, item in enumerate(chart_data): total_cost = ( - item.chart_capex_annualized - + item.chart_oem_annualized - + item.chart_fuel_cost_annualized - + item.cost_disposal_cost + float(item.chart_capex_annualized) + + float(item.chart_oem_annualized) + + float(item.chart_fuel_cost_annualized) + + float(item.cost_disposal_cost) ) - revenue = item.chart_revenue_annualized + revenue = float(item.chart_revenue_annualized) if previous_total_cost is not None and previous_revenue is not None: prev_diff = previous_total_cost - previous_revenue