111 lines
3.2 KiB
Python
111 lines
3.2 KiB
Python
"""Built-in formula functions for the bitable formula engine.
|
|
|
|
v1 implements: SUM, AVG, COUNT, MIN, MAX, CONCAT, ABS, ROUND, IF, LEN.
|
|
|
|
Aggregate functions (SUM/AVG/COUNT/MIN/MAX) accept a list of values
|
|
(the entire column) and return a scalar. Non-aggregate functions
|
|
(ABS/ROUND/IF/LEN/CONCAT) operate on scalars.
|
|
|
|
The engine determines whether to pass a column (list) or scalar (row value)
|
|
based on the calling context — see :mod:`agentkit.bitable.formula.engine`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable, TypeAlias
|
|
|
|
# A formula evaluates to a scalar primitive: text, number, or nothing.
|
|
# bool is intentionally excluded — comparisons live in the parser layer
|
|
# and never reach the function registry.
|
|
FormulaResult: TypeAlias = str | int | float | None
|
|
|
|
# ── Aggregate functions (operate on lists) ────────────────
|
|
|
|
|
|
def _sum(values: list[FormulaResult]) -> float | int:
|
|
"""Sum of numeric values, ignoring None/empty."""
|
|
total = 0
|
|
for v in values:
|
|
if v is None or v == "":
|
|
continue
|
|
total += v
|
|
return total
|
|
|
|
|
|
def _avg(values: list[FormulaResult]) -> float:
|
|
"""Average of numeric values, ignoring None/empty."""
|
|
nums = [v for v in values if v is not None and v != ""]
|
|
if not nums:
|
|
return 0.0
|
|
return sum(nums) / len(nums)
|
|
|
|
|
|
def _count(values: list[FormulaResult]) -> int:
|
|
"""Count of non-empty values."""
|
|
return sum(1 for v in values if v is not None and v != "")
|
|
|
|
|
|
def _min(values: list[FormulaResult]) -> FormulaResult:
|
|
"""Minimum of numeric values, ignoring None/empty."""
|
|
nums = [v for v in values if v is not None and v != ""]
|
|
if not nums:
|
|
return 0
|
|
return min(nums)
|
|
|
|
|
|
def _max(values: list[FormulaResult]) -> FormulaResult:
|
|
"""Maximum of numeric values, ignoring None/empty."""
|
|
nums = [v for v in values if v is not None and v != ""]
|
|
if not nums:
|
|
return 0
|
|
return max(nums)
|
|
|
|
|
|
# ── Scalar functions ──────────────────────────────────────
|
|
|
|
|
|
def _abs(value: FormulaResult) -> FormulaResult:
|
|
return abs(value)
|
|
|
|
|
|
def _round(value: FormulaResult, digits: int = 0) -> float:
|
|
return round(value, digits)
|
|
|
|
|
|
def _if(
|
|
condition: FormulaResult,
|
|
true_val: FormulaResult,
|
|
false_val: FormulaResult = None,
|
|
) -> FormulaResult:
|
|
return true_val if condition else false_val
|
|
|
|
|
|
def _len(value: FormulaResult) -> int:
|
|
if value is None:
|
|
return 0
|
|
return len(str(value))
|
|
|
|
|
|
def _concat(*args: FormulaResult) -> str:
|
|
"""Concatenate all arguments as strings."""
|
|
return "".join(str(a) for a in args if a is not None)
|
|
|
|
|
|
# ── Registry ──────────────────────────────────────────────
|
|
|
|
# Functions that aggregate a column (receive a list of all column values)
|
|
AGGREGATE_FUNCTIONS: frozenset[str] = frozenset({"SUM", "AVG", "COUNT", "MIN", "MAX"})
|
|
|
|
FUNCTION_REGISTRY: dict[str, Callable[..., FormulaResult]] = {
|
|
"SUM": _sum,
|
|
"AVG": _avg,
|
|
"COUNT": _count,
|
|
"MIN": _min,
|
|
"MAX": _max,
|
|
"ABS": _abs,
|
|
"ROUND": _round,
|
|
"IF": _if,
|
|
"LEN": _len,
|
|
"CONCAT": _concat,
|
|
}
|