app.transformation_manager
==========================

.. py:module:: app.transformation_manager

.. autoapi-nested-parse::

   Provides the ore logic of the backend.



Submodules
----------

.. toctree::
   :maxdepth: 1

   /autoapi/app/transformation_manager/bpmn_templates/index
   /autoapi/app/transformation_manager/frontend_graph/index
   /autoapi/app/transformation_manager/graph/index
   /autoapi/app/transformation_manager/merge/index
   /autoapi/app/transformation_manager/optimize/index
   /autoapi/app/transformation_manager/post/index
   /autoapi/app/transformation_manager/pre/index
   /autoapi/app/transformation_manager/utils/index






Package Contents
----------------

.. py:class:: 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 :class:`~app.transformation_manager.frontend_graph.FrontendGraph`.

   :param enricher: The enricher to use to get node implementations
   :frontend_graph: The graph to process
   :optimize_settings: Specify how to optimize the result


.. py:class:: 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.


   .. py:method:: process_nodes() -> None
      :async:


      Process graph by enriching and preprocessing the nodes.



   .. py:method:: process() -> str
      :async:


      Process the :class:`~app.model.CompileRequest`.

      #. Enrich frontend nodes.
      #. :meth:`~app.transformation_manager.pre.preprocess` frontend nodes.
      #. Optionally :meth:`~app.transformation_manager.optimize.optimize` graph width.
      #. :meth:`~app.transformation_manager.merge.merge_nodes` and
      #. :meth:`~app.transformation_manager.post.postprocess` into final program.

      :return: The final QASM program as a string.



.. py:class:: 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.


   .. py:method:: enrich() -> collections.abc.AsyncIterator[app.model.CompileRequest.ImplementationNode]
      :async:


      Yield enrichment of nodes.



   .. py:method:: enrich_all() -> list[app.model.CompileRequest.ImplementationNode]
      :async:


      Get list of all enrichments.



.. py:class:: 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.


   .. py:method:: process() -> tuple[str, bytes]
      :async:


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



   .. py:method:: identify_quantum_groups() -> dict[str, list[app.model.CompileRequest.ImplementationNode]]
      :async:


      Returns a dictionary of quantum groups keyed by group ID.



   .. py:method:: generate_service_zips(composite_nodes: list[str], node_metadata: dict[str, dict[str, Any]]) -> bytes
      :async:


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



.. py:class:: 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.


   .. py:method:: insert_all() -> None
      :async:


      Insert all enrichments.



.. py:function:: generate_qrms(quantum_groups: dict[str, list[app.model.CompileRequest.ImplementationNode]]) -> bytes
   :async:


   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)