cf-pipeline-engine 0.2.0

Cogniflow pipeline engine (C++ core) with a thin Python package wrapper.

cf-pipeline-engine

The Cogniflow pipeline engine implemented in C++ (compiler, scheduler, runtime).

This folder is structured as a Python package for consistency with the other cf_* components, even though the core implementation is native C++ and built via CMake.

Build (CMake)

cmake -S . -B build
cmake --build build

Python package wrapper

The Python package is intentionally thin and provides access to the packaged native engine assets:

• cf_pipeline_engine.cf_pipeline_v2_path()
• cf_pipeline_engine.cf_siggen_path()
• cf_pipeline_engine.cf_engine_include_path()
• cf_pipeline_engine.cf_type_registry_path()
• cf_pipeline_engine.resolve_cf_pipeline_v2_executable()

Published distribution name:

pip install cf-pipeline-engine

The published wheel installs:

• bin/cf_pipeline_v2(.exe)
• bin/cf_siggen(.exe)
• bin/type_registry.v0.json
• include/*.h

The native sink path no longer embeds cf_datahive_cpp sources into the engine build. Instead, cf-pipeline-engine links against the packaged native consumer surface exported by cf-datahive.

cfio:OpcuaReaderStep is now runner-owned as well. The engine executes that step through the installed cf-opcua-server owner surface instead of loading protocol code from cf-basic-io. If the console script is not on PATH, set CF_OPCUA_SERVER_CMD to the absolute script or executable path.

Python CLI

The legacy standalone Python wrapper package has been retired. Its CLI surface now lives in this cf_pipeline_engine package.

Direct invocation remains policy-gated by CF_ALLOW_DIRECT_ENGINE:

CF_ALLOW_DIRECT_ENGINE=1 python -m cf_pipeline_engine.cli \
  --pipeline sandcastle/cf_pipeline/cf_pipeline_engine/examples/opcua_fifo_avg_to_duckdb_parquet_triggered.nq \
  --interval 1 \
  --duration 10

Equivalent module entrypoint:

CF_ALLOW_DIRECT_ENGINE=1 python -m cf_pipeline_engine --help

Publishing

cf_pipeline_engine is published with the dedicated Windows workflow:

• Workflow: .github/workflows/cf_pipeline_engine_windows_publish.yml
• Package directory: sandcastle/cf_pipeline/cf_pipeline_engine
• PyPI tag: cf-pipeline-engine-v<version>
• TestPyPI tag: cf-pipeline-engine-v<version>-test

Local preflight:

powershell -ExecutionPolicy Bypass -File scripts/mimic_windows_python_publish_workflow.ps1 `
  -WorkflowFile .github/workflows/cf_pipeline_engine_windows_publish.yml `
  -PackageDir sandcastle/cf_pipeline/cf_pipeline_engine `
  -PythonExe py `
  -PythonVersion 3.13

Queue a dry-run dispatch:

powershell -ExecutionPolicy Bypass -File scripts/queue_windows_python_publish_workflow.ps1 `
  -WorkflowFile .github/workflows/cf_pipeline_engine_windows_publish.yml `
  -PackageDir sandcastle/cf_pipeline/cf_pipeline_engine `
  -PublishTarget testpypi `
  -Ref main `
  -RequireLocalPass `
  -DryRun

OPC UA Demo Pipeline Sink

The existing demo pipeline examples/opcua_fifo_avg_to_duckdb_parquet_triggered.nq now uses cfsink:DataHiveParquetSinkStep from cf_basic_sinks.

Its cfio:OpcuaReaderStep ingress remains in cf-basic-io as a declarative step definition, while the runner performs the actual OPC UA snapshot through cf-opcua-server.

The step stays declarative in cf_basic_sinks, while the runner executes the sink through the packaged C++ gatekeeper library exported by cf-datahive, producing one committed data hive run with 20 rows (cycle_id 0..19) and no archive.jsonl.

Run the one-click demo:

.\scripts\fresh_install_v2.ps1 -Clean -RunDemo -ShowDetails

Expected output layout after one demo session:

• workspace/<data_hive>/opcua_fifo_avg/latest.txt
• workspace/<data_hive>/opcua_fifo_avg/runs/<run_id>/manifest.json
• workspace/<data_hive>/opcua_fifo_avg/runs/<run_id>/tables/measurements/part-*.parquet