Python API usage

Public entry points are in chronaxis. Use load_analysis_plan for a configuration file or AnalysisPlan.from_dict for a programmatically constructed configuration. AnalysisRunner executes it; AnalysisResult reads and publishes the output. Complete installation before running the following examples.

Basic single-source ASC/BLF configurations set execution.process_count: 4; the run(plan) call below inherits it. To override it for one call, use AnalysisRunner().run(plan, process_count=4). The instance's validate and explain methods accept the same keyword-only parameter. None inherits the setting without changing the plan; a positive integer sets the limit for this call. CSV, memory, and multi-source analyses use one worker setting; independent batch tasks configure parallelism at the outer level. See continuous-file multiprocessing for the supported scope.

Reading parameter tables

Reference tables distinguish actual Python names, types, required arguments, defaults, and purposes. None means an optional object is not supplied; it does not mean zero, an empty string, or an empty list. A vertical bar in a type denotes alternatives. A tuple is an ordered, immutable sequence. Class fields describe stored state; constructor arguments follow the actual signature. Generated standard methods and third-party interfaces identify their origins separately.

Load, check, and execute

Save this complete example as a .py file and run it from the repository root. Keep the if __name__ == "__main__": guard so Windows worker startup does not rerun the main program. In an offline examples directory, run api_example.py, which defaults to the adjacent analysis.yaml and also accepts a custom configuration path.

from chronaxis import AnalysisRunner, load_analysis_plan, validate_analysis_plan

def main():
    plan = load_analysis_plan("Docs/Current/examples/analysis.yaml")
    report = validate_analysis_plan(plan)
    if not report.valid or not report.execution_ready:
        raise RuntimeError(report)

    with AnalysisRunner().run(plan) as result:
        rows = result.iter_calculations()
        try:
            for row in rows:
                print(row.calculation_id, row.value, row.unit, row.status)
        finally:
            rows.close()
        result.write("new-result-directory")


if __name__ == "__main__":
    main()

The configuration permits up to four workers, but this example has one ASC and default concatenation, so it reads in the current process; explain reports read_process_limit: 0. To run two parts of the same data, load Docs/Current/examples/input-can-directory.yaml instead: read_process_limit becomes 2 and the calculation results remain the same. This is a plan limit, not a measured PID count. The synthetic example produces the following terminal output. Each row gives the calculation name, value, unit and status.

energy 38.0 J complete
sample_average 6.0 W complete
time_average 6.333333333333333 W complete

Power values of 2, 4, 8 and 10 W last 1, 2, 1 and 2 seconds, respectively. The integral is 2×1 + 4×2 + 8×1 + 10×2 = 38 J. The four readings average 6 W; the time-weighted mean is 38÷6 W. The write call saves these results to the example output directory.

load_analysis_plan reads the main configuration and its explicit, single-level shared definitions. Relative paths are resolved from the file in which they are declared. AnalysisPlan.from_dict(config, base_dir=...) accepts a Python dictionary rather than reading a main file; omitted base_dir means the current directory. Both entry points may enumerate input directories and read DBC metadata to infer CAN types, units, or enumerations. They do not scan business messages or CSV records at this stage. from_dict rejects unexpanded imports: use the file loader or chronaxis config resolve first. Both check unknown fields, name collisions, expressions, and references.

validate_analysis_plan returns a static ValidationReport. valid reports declaration validity; execution_ready reports whether the execution path supports the plan. explain_analysis_plan returns an ExplainReport containing dependencies and backend decisions. Static checks on a loaded plan do not read business records. DBC metadata read during loading is not proof that actual data is valid. Neither valid nor execution_ready certifies whole-file contents: malformed records, backwards timestamps, runtime permissions or insufficient resources may be discovered only by run.

AnalysisRunner(backend=None, *, extensions=None) uses the plan's setting. Omitting backend in the configuration selects auto, which chooses reference or numpy from the whole plan. Reference uses the independent numerical implementation; NumPy uses point or array execution as appropriate. Explicit auto, reference or numpy overrides that run. See execution settings for the selection scope. run(plan, work_dir=None, *, process_count=None) executes and returns an AnalysisResult. Without work_dir, it allocates independent temporary storage; an explicit working directory must be new. Results are not published automatically: call write with an output directory, which may already exist. Each publication uses timestamped filenames.

Omit extensions for built-in operations. A Python host that registers custom functions or rolling operators must pass the same extension registry when loading the plan and constructing AnalysisRunner. See Custom analysis operators. An asterisk in a signature marks the following arguments as keyword-only.

Read rows and close resources

iter_events, iter_segments, iter_sequences, iter_calculations, iter_selections, and iter_input_trace return cursors for events, windows, pairs, calculations, selected samples, and input tracing respectively. iter_aggregations reads grouped statistics over calculation rows. iter_contributions reads explicitly requested per-message contributions. Unrequested details are not generated by default. Rows use the fields described under results.

Close a cursor whether you exhaust it or stop early. Manage the result with with, or call close in finally. Reading a closed result raises ResultClosedError. Resource failures and cancellation are not empty results; failed publication or cleanup must not be reported as completion.

Main entry points

See the Python API index for precise signatures. The developer reference describes internal modules and private implementations in Chinese.

Build configuration in a program

This complete example reads the supplied JSON equivalent into a dictionary. base_dir establishes the origin of relative input paths. This entry point does not read imports; use the file loader for shared definitions.

import json
from pathlib import Path
from chronaxis import AnalysisPlan, explain_analysis_plan

directory = Path("Docs/Current/examples")
config = json.loads((directory / "analysis.json").read_text(encoding="utf-8"))
plan = AnalysisPlan.from_dict(config, base_dir=directory)
print(explain_analysis_plan(plan))

Exceptions and resource ownership

Import AnalysisError and ResultClosedError from chronaxis. Other result exceptions are defined in chronaxis.results.errors, a Chinese implementation reference. Results and cursors own resources independently: use with for the result and finally for cursors. After closing a result, its run_manifest still provides a small diagnostic snapshot, but data cannot be read or published again.

If cleanup fails, retain the original exception and the result object. For an ordinary run failure, inspect getattr(exc, "runner_cleanup_owner", None) on the caught exception; publication or result-close failures may provide analysis_result_cleanup_owner. Resolve the resource fault, then call close() on an available recovery object or retry closing the result you still hold. Recovery releases this invocation's resources; it does not rerun calculations or undo published results. An attribute may still refer to an already closed object, so its presence alone does not establish that resources remain open.

Time and preset entry points

from chronaxis import parse_duration, checked_add_ns, convert_time

duration_ns = parse_duration("500ms").time_ns  # Integer nanoseconds in the conversion result
end_ns = checked_add_ns(1_000_000_000, duration_ns)
converted = convert_time(1.5, "s")
print(duration_ns, end_ns, converted)

Usually, use configuration presets; see business presets for role and parameter examples. To compose them in Python, build PresetSpec and SignalDefinition objects, check roles with resolve_preset_roles, and obtain a PlanFragment with build_preset. The fragment must be merged into a complete AnalysisPlan before execution. Batch tasks includes complete loading, execution, and report-cleanup code.

Build and execute a preset from typed signals

For programmatic role checks or composition, the complete call sequence is SignalDefinition → PresetSpec → resolve_preset_roles → build_preset → PlanFragment.to_dict → AnalysisPlan.from_dict → AnalysisRunner.run.

Import SignalDefinition from chronaxis.core.binding; it is not exported under the same name from the root package. resolve_preset_roles returns binding checks, not samples. build_preset also checks roles itself. A fragment contains rules, windows, derived expressions, and calculations; supply dataset, sources, and signals separately. Do not merge a fragment and also place the same preset in presets, which would expand it twice.

The complete script is preset_api_example.py. From the examples directory, run python preset_api_example.py --output preset-api-result with a new output directory. This example script creates that parent directory for both its temporary work and published results. It produces heat_duration=9.0s, complete. Dictionaries here are Python function arguments; users need not rewrite YAML as JSON.

This API example supplies in-memory records and explicitly uses process_count: 1. For your ASC/BLF volumes, use the file-loading example above and its execution.process_count: 4 configuration.

"""Build a preset from typed signals and run it. --output selects a new directory for work and published results."""
from pathlib import Path
import argparse
from chronaxis import (PresetSpec, resolve_preset_roles,
                       build_preset, AnalysisPlan, AnalysisRunner)
from chronaxis.core.binding import SignalDefinition
from chronaxis.core.types import SignalDType


def run(output):
    """Return calculation rows; propagate role, plan, and execution errors and close results."""
    # Typed validity uses integer nanoseconds; the definition contains no samples.
    temperature=SignalDefinition(name='temperature',source='table',field='temperature',
        dtype=SignalDType.FLOAT64,unit='degC',valid_for_ns=10_000_000_000)
    signals=(temperature,)
    spec=PresetSpec.from_dict(dict(name='heat',preset_type='threshold_excursion',version='1',
        role_bindings={'signal':'temperature'},parameters={'unit':'degC','enter_threshold':80,'exit_threshold':75}))
    roles=resolve_preset_roles(spec,signals)
    assert roles[0].definition is temperature and roles[0].resolution=='explicit'
    fragment=build_preset(spec,signals)  # Also checks roles; returns a PlanFragment.
    nodes=fragment.to_dict()            # Nodes only; no sources, signals, or results.
    raw=dict(config_version=4,dataset=dict(name='preset_api'),
        execution=dict(process_count=1),
        sources=[dict(name='table',source_type='memory',
            rows=[dict(time=f'{t}s',values={'temperature':v}) for t,v in ((0,85),(3,77),(9,75),(10,75))])],
        signals=[dict(name=s.name,source=s.source,field=s.field,dtype=s.dtype.value,unit=s.unit,valid_for=f'{s.valid_for_ns}ns') for s in signals])
    # This base has no node lists yet. Existing lists must instead be merged individually.
    # Do not also add spec to raw['presets'], which would expand the same preset twice.
    raw.update(nodes)
    plan=AnalysisPlan.from_dict(raw)
    with AnalysisRunner().run(plan,work_dir=output/'work') as result:
        rows=list(result.iter_calculations())
        assert len(rows)==1 and rows[0].value==9.0 and rows[0].unit=='s'
        result.write(output/'published')
        return rows


def main():
    """Read a CLI result directory; propagate validation, execution, and I/O errors."""
    parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--output',type=Path,required=True)
    output=parser.parse_args().output;output.mkdir(parents=True,exist_ok=False)
    for row in run(output):print(row.calculation_id,row.value,row.unit,row.status)


if __name__=='__main__':main()

When merging into an existing plan, merge derived by name and append each node list, then let AnalysisPlan check names and references. Do not use update to overwrite existing business lists. This example uses it because the base has no such nodes. See the public API for complete signatures and parameters.

Python example: result aggregation

Run this code from the repository root. The plan first measures mean power during each load. AnalysisResult.iter_aggregations() then reads P95 across the load means, one row per group. Publishing to CSV is optional. Power test and downloadable inputs

Save the code as a .py file and run it from the repository root. The main guard prevents Windows reader processes from running the script again. The first example prints load_mean_p95 19.5 W complete 2 2. The two activity means are 10 W and 20 W, whose linear-interpolated P95 is 19.5 W.

from chronaxis import AnalysisRunner, load_analysis_plan

def main():
    plan = load_analysis_plan("Docs/Current/examples/power-statistics.yaml")
    with AnalysisRunner().run(plan) as result:
        for row in result.iter_aggregations():
            print(row.aggregation_id, row.value, row.unit, row.status,
                  row.included_count, row.candidate_count)


if __name__ == "__main__":
    main()