ObjCalculation is the core calculation engine for the Axion platform. It loads
SQL-based calculation definitions from the database and executes them against a
target GUID. Calculations can perform lookups, transformations, and multi-step
computations that enrich workflow context or application state.
This module was extracted from factory.service/package.core/ObjServiceCalculation.py
so that the calculation engine is available directly in factory.core without
going through the service layer. The service (ObjServiceCalculation.ObjServiceApi)
remains as a thin subclass providing service-framework integration.
Objects.Object
→ ObjDataKey.ObjDataKey
→ ObjDataSql.ObjDataSql
→ ObjDataDDL.ObjDataDDL
→ ObjData.ObjData
→ ObjSimulation.ObjSimulation
→ ObjApi.ObjApi
→ ObjCalculation.ObjCalculation ← this module
→ ObjServiceCalculation.ObjServiceApi ← service wrapper
| Table | Purpose |
|---|---|
def_calculations |
One row per named calculation step: SQL, rank, flags |
def_calculation |
One row per calculation group: metadata, threewords label |
Process-lifetime caches prevent repeated DB queries:
| Variable | Key | Purpose |
|---|---|---|
BUFFER_CALC |
"group:name:pkg:sim" |
Caches a single calculation row from def_calculations |
BUFFER_CALCSET |
group_name |
Caches the ordered list of names for a group |
BUFFER_CALC_DETAIL |
— | Declared, reserved for future use |
_NOTIFY_CHECKED |
bool | Ensures CALC_FAIL notify entry check runs once per process |
_COLUMN_CHECKED |
"table:col" |
Ensures add_column for SimulationSql runs once per table |
_VERSION_CHECKED |
"group:package" |
Ensures _ensure_versioned runs once per group+package |
These are module globals shared across all ObjCalculation instances within the
same Python process.
run_calculation_set(group_name, guid_val, simul_guid, context_dict)Runs all calculation steps for a group in Rank order. Entry point used by
ObjWorkflowCalc and ObjScheduler.
run_calculation(table_name, group_name, calculation_name, simul_guid, context_dict)Executes a single named calculation step. Handles three execution modes:
Operation field set) — fetches a dict row and callsexecute_operation() (e.g. transfer(...)).HasResult = 'Y') — runs SQL, captures return value, sets_calc_* attributes on the instance.sql_execute() for UPDATE/INSERT steps.In UAT/simulation mode the SimulationSql column is preferred over
CalculationSql when present.
map_calculation(calculation_text)Replaces all $placeholder$ tokens in a SQL string before execution.
Uses a single-pass dict substitution instead of chained .replace() calls.
Early-exits when no $ tokens are present. Handles:
$guid$, $id$, $setguid$, $setvalue$$sourcetable$, $basetable$, $targettable$$sourceguid$, $targetguid$, $mapnames$$MapP2$–$MapP9$ parameter aliases → $param2$–$param9$patch_param_calc)ObjData.patch_param)patch_param_calc(text)Replaces $attribute_name$ and &attribute_name& tokens using the instance's
runtime attributes. Uses a reverse-lookup strategy: extracts placeholder tokens
from the SQL first, then finds matching attrs — avoids iterating all 100+
__dict__ entries when only a few placeholders are present.
Skips internal _Uppercase attributes. Delegates remaining $…$ tokens to
ObjData.ObjData.patch_param.
Prefix handling: Attributes whose name starts with a recognized non-empty
context prefix (_form_, _calc_, _api_, _sys_, _service_, sys_)
are matched by both full name and suffix form (e.g. _form_name matches
both $_form_name$ and $name$). All other attributes use simple
$attr_lower$ matching.
load_calculation_set(group_name)Returns the ordered list of CalculationName values for a group. Queries
def_calculations filtered by active package and Active = 'Y'.
compile_calculation(group_name)Analyses the SQL statements for a group and identifies adjacent UPDATE steps
targeting the same table and WHERE clause that could be merged. Used for
optimisation and reporting.
Visualise(state_name, calculation_group)Generates a Mermaid state diagram fragment for a calculation group. Used by
ObjServiceClassification and ObjServiceSegmentation to visualise pipelines.
_strip_sql_comments(sql) (static)Strips SQL comments before execution. Handles:
-- single-line comments (preserving content inside single-quoted strings)/* ... */ block comments\\n escape sequences to real newlines firstAdded to prevent comments from eating SQL keywords (e.g. -- comment
followed by end end on the same line consumed the closing keywords).
_generate_threewords() → strCalled once per calculation group run to generate a human-readable three-word
label stored in def_calculation.Threewords. Returns "" by default, which
is safe (the column is just a display label).
ObjServiceCalculation.ObjServiceApi overrides this to call ObjServiceGuidName,
keeping the factory.service dependency out of factory.core.
Key attributes populated during run_calculation:
| Attribute | Source | Purpose |
|---|---|---|
Guid |
Caller | Primary record being computed |
SetGuid |
set_sql result |
Iterator GUID within a loop |
MapGuid |
DB | GUID alias for placeholder substitution |
MapP2–MapP9 |
DB | Parameter alias mappings |
source_table |
DB | $sourcetable$ placeholder value |
target_table |
DB | $targettable$ placeholder value |
set_sql |
DB | SQL that drives the SET iterator |
tracking_context |
Runtime | Snapshot of calc inputs/outputs per step |
send_email(group_name, recipients) — Send a branded HTML email with the calculation steps table (rank, name, result flag, SQL preview), version history, RACI matrix, and signoff status. Falls back to RACI contacts if no recipients specified.The def_calculation table has RACI columns: RaciSimulation, RaciProduction, RaciOutcome, RaciDevelopment. These are used by send_email() to render governance information and resolve default recipients. Package supplementation fills empty slots from def_package_roles.
All CLI commands use typer.
# Run a calculation group directly against a GUID
python factory.core/ObjCalculation.py direct \
GROUP_NAME GUID_VALUE
# List all calculation groups
python factory.core/ObjCalculation.py list
python factory.core/ObjCalculation.py list \
--package CORE
# Run benchmark (store outcomes as baseline)
python factory.core/ObjCalculation.py benchmark \
GROUP_NAME source_table \
--guid-column Guid --limit 500
# Compare current outcomes against stored baseline
python factory.core/ObjCalculation.py compare \
GROUP_NAME source_table \
--guid-column Guid --limit 500
# Send calculation group report email
python factory.core/ObjCalculation.py send-email \
GROUP_NAME --recipients "user@example.com"
| Command | Purpose |
|---|---|
direct |
Run a group against a single GUID |
list |
Show all groups with step counts |
benchmark |
Run group against a table, store to bloom_calc_{group} |
compare |
Re-run and diff against the stored baseline |
send-email |
Branded HTML report with steps, RACI, version history |
_CalcOptimizer)A private helper class that analyses compiled SQL
at execution time and applies safe transformations
without modifying stored definitions.
_get_optimization_plan(group_name) builds a
per-process-lifetime plan cached in
BUFFER_OPTIMIZATION. Each step is classified:
| Plan type | Trigger | Behaviour |
|---|---|---|
literal |
SELECT 'val' AS alias |
Sets attr directly, no DB round-trip |
constants |
SELECT col AS alias FROM constants |
Cached lookup, once per key |
merge_leader |
Adjacent UPDATEs, same table/WHERE | Leads a merged SET clause |
merge_skip |
Subsequent UPDATE in merge group | SETs folded into leader |
None |
Everything else | Executed normally |
| Variable | Key | Purpose |
|---|---|---|
BUFFER_CALC |
"group:name:pkg:sim" |
Single calculation row |
BUFFER_CALCSET |
group_name |
Ordered step names |
BUFFER_CALC_DETAIL |
— | Reserved for future use |
BUFFER_CALC_GROUP_INIT |
set |
Groups already initialised |
BUFFER_OPTIMIZATION |
"group:package" |
Optimization plans per group |
_NOTIFY_CHECKED |
bool | CALC_FAIL notify entry checked |
_COLUMN_CHECKED |
"table:col" |
add_column guard |
_VERSION_CHECKED |
"group:package" |
_ensure_versioned guard |
_flush_merge_buffer() accumulates adjacent
UPDATE statements that target the same table and
WHERE clause. When flushed, they are combined
into a single UPDATE with merged SET clauses via
_CalcOptimizer.merge_updates().
When get_deployment() returns DEV, each
calculation step runs explain_sql() (from
ObjDataSql) before execution. Results are
stored in track_calculations with three
additional columns:
| Column | Type | Content |
|---|---|---|
ExplainKey |
VARCHAR(255) | Index name used (or empty) |
ExplainWarning |
TEXT | Human-readable warning for full scans |
ExplainRows |
INT | Estimated rows scanned |
Columns are added lazily via self.add_column().
_ai_explain_analysis(group_name) reads the
EXPLAIN tracking data and sends it to an LLM
(default llm:qwen3:8b) for analysis. The
response identifies bottlenecks, missing indexes,
merge candidates, and quick wins. The HTML result
is included in the send_email() report.
run_benchmark(group_name, source_table, guid_column, limit)Runs a calculation group against every GUID in
source_table and stores outcomes in
bloom_calc_{group_name}. Each row captures:
FinalResult — return value of the groupOutcomes — JSON dict of all _calc_* attrsElapsedMs — per-record execution timeReturns a summary dict with processed,
elapsed_s, avg_ms, output_table,
set_guid, and distribution.
compare_benchmark(group_name, source_table, guid_column, limit)Loads the stored baseline from
bloom_calc_{group_name}, re-runs the group for
each GUID, and compares FinalResult and
Outcomes JSON. Returns:
total, matches, mismatches, match_pctdiffs — first 50 differing records with bothNine optimisations applied to reduce per-calculation overhead:
| # | Fix | Before | After |
|---|---|---|---|
| 1 | _ensure_calc_fail_notify |
2 SQL per init | Once per process |
| 2 | _ensure_versioned |
Schema check per run | Once per group+package |
| 3 | add_column(SimulationSql) |
Column check per SIM/UAT run | Once per table |
| 4 | map_calculation |
13+ chained .replace() | Single-pass dict |
| 5 | patch_param_calc |
Iterates all dict | Reverse lookup from tokens |
| 6 | _strip_sql_comments |
Called every time | Skip when no -- or /* |
| 7 | Tracking dict scan | Per guid iteration | Pre-compute attr list |
| 8 | copy.copy(string) |
Unnecessary copy | Removed (strings immutable) |
| 9 | Operation comment strip | Called every time | Same inline skip as #6 |
self.IsMySql(self.DB) at line ~430 uses the old capitalized method name.Objects.py method is is_my_sql. This is a pre-existing issueObjServiceCalculation and does not affect runtime in mostDO_TRACKING flag is False by default).