app.transformation_manager

Provides the ore logic of the backend.

Submodules

• app.transformation_manager.bpmn_templates
• app.transformation_manager.frontend_graph
• app.transformation_manager.graph
• app.transformation_manager.merge
• app.transformation_manager.optimize
• app.transformation_manager.post
• app.transformation_manager.pre
• app.transformation_manager.utils

Package Contents

class app.transformation_manager.CommonProcessor(enricher: app.enricher.Enricher, frontend_graph: frontend_graph.FrontendGraph, optimize_settings: app.model.CompileRequest.OptimizeSettings, qiskit_compat: bool = False, result: str | None = None, original_request: app.model.CompileRequest.CompileRequest | None = None)

Process a FrontendGraph.

Parameters:

• enricher (app.enricher.Enricher) – The enricher to use to get node implementations

• frontend_graph (frontend_graph.FrontendGraph)

• optimize_settings (app.model.CompileRequest.OptimizeSettings)

• qiskit_compat (bool)

• result (str | None)

• original_request (app.model.CompileRequest.CompileRequest | None)

Frontend_graph:

The graph to process

Optimize_settings:

Specify how to optimize the result

class app.transformation_manager.MergingProcessor(enricher: app.enricher.Enricher, frontend_graph: frontend_graph.FrontendGraph, optimize_settings: app.model.CompileRequest.OptimizeSettings, qiskit_compat: bool = False, result: str | None = None, original_request: app.model.CompileRequest.CompileRequest | None = None)

Process request with the whole pipeline.

Parameters:

• enricher (app.enricher.Enricher)

• frontend_graph (frontend_graph.FrontendGraph)

• optimize_settings (app.model.CompileRequest.OptimizeSettings)

• qiskit_compat (bool)

• result (str | None)

• original_request (app.model.CompileRequest.CompileRequest | None)

async process_nodes() → None

Process graph by enriching and preprocessing the nodes.

Return type:

None

async process() → str

Process the CompileRequest.

1. Enrich frontend nodes.

2. preprocess() frontend nodes.

3. Optionally optimize() graph width.

4. merge_nodes() and

5. postprocess() into final program.

Returns:

The final QASM program as a string.

Return type:

str

class app.transformation_manager.EnrichingProcessor(enricher: app.enricher.Enricher, frontend_graph: frontend_graph.FrontendGraph, optimize_settings: app.model.CompileRequest.OptimizeSettings, qiskit_compat: bool = False, result: str | None = None, original_request: app.model.CompileRequest.CompileRequest | None = None)

Return enrichment for all nodes.

Parameters:

• enricher (app.enricher.Enricher)

• frontend_graph (frontend_graph.FrontendGraph)

• optimize_settings (app.model.CompileRequest.OptimizeSettings)

• qiskit_compat (bool)

• result (str | None)

• original_request (app.model.CompileRequest.CompileRequest | None)

async enrich() → collections.abc.AsyncIterator[app.model.CompileRequest.ImplementationNode]

Yield enrichment of nodes.

Return type:

collections.abc.AsyncIterator[app.model.CompileRequest.ImplementationNode]

async enrich_all() → list[app.model.CompileRequest.ImplementationNode]

Get list of all enrichments.

Return type:

list[app.model.CompileRequest.ImplementationNode]

class app.transformation_manager.WorkflowProcessor(enricher: app.enricher.Enricher, frontend_graph: frontend_graph.FrontendGraph, optimize_settings: app.model.CompileRequest.OptimizeSettings, qiskit_compat: bool = False, result: str | None = None, original_request: app.model.CompileRequest.CompileRequest | None = None)

Process a request to a workflow representation.

Parameters:

• enricher (app.enricher.Enricher)

• frontend_graph (frontend_graph.FrontendGraph)

• optimize_settings (app.model.CompileRequest.OptimizeSettings)

• qiskit_compat (bool)

• result (str | None)

• original_request (app.model.CompileRequest.CompileRequest | None)

async process() → tuple[str, bytes]

Run enrichment, classify nodes, group quantum nodes, and return BPMN XML.

Return type:

tuple[str, bytes]

async identify_quantum_groups() → dict[str, list[app.model.CompileRequest.ImplementationNode]]

Returns a dictionary of quantum groups keyed by group ID.

Return type:

dict[str, list[app.model.CompileRequest.ImplementationNode]]

async generate_service_zips(composite_nodes: list[str], node_metadata: dict[str, dict[str, Any]]) → bytes

Generate one ZIP per node. Each ZIP contains a single service with logic ONLY for that node.

Parameters:

• composite_nodes (list[str])

• node_metadata (dict[str, dict[str, Any]])

Return type:

bytes

class app.transformation_manager.EnrichmentInserter(inserts: list[app.model.CompileRequest.SingleInsert], enricher: app.enricher.Enricher, engine: sqlalchemy.ext.asyncio.AsyncEngine)

Insert enrichment implementations for frontend-nodes into the Enricher.

Parameters:

• inserts (list[app.model.CompileRequest.SingleInsert])

• enricher (app.enricher.Enricher)

• engine (sqlalchemy.ext.asyncio.AsyncEngine)

async insert_all() → None

Insert all enrichments.

Return type:

None

async app.transformation_manager.generate_qrms(quantum_groups: dict[str, list[app.model.CompileRequest.ImplementationNode]]) → bytes

Generate QRM BPMN files for each quantum node in the provided groups.

For each quantum node: - Create a detector BPMN (quantumCircuitLoadingTask) - Create a replacement BPMN (serviceTask with deploymentModelUrl) - Package both into a ZIP file named after the node (e.g. node_id.zip)

Returns:

dict[str, bytes]: mapping of node_id -> zip file content (bytes)

Parameters:

quantum_groups (dict[str, list[app.model.CompileRequest.ImplementationNode]])

Return type:

bytes