MinerU/next_docs/en/user_guide/tutorial/pipeline.rst

Pipeline
==========


Minimal Example 
^^^^^^^^^^^^^^^^^

.. code:: python

    import os

    from magic_pdf.data.data_reader_writer import FileBasedDataWriter, FileBasedDataReader
    from magic_pdf.data.dataset import PymuDocDataset
    from magic_pdf.model.doc_analyze_by_custom_model import doc_analyze

    # args
    pdf_file_name = "abc.pdf"  # replace with the real pdf path
    name_without_suff = pdf_file_name.split(".")[0]

    # prepare env
    local_image_dir, local_md_dir = "output/images", "output"
    image_dir = str(os.path.basename(local_image_dir))

    os.makedirs(local_image_dir, exist_ok=True)

    image_writer, md_writer = FileBasedDataWriter(local_image_dir), FileBasedDataWriter(
        local_md_dir
    )

    # read bytes
    reader1 = FileBasedDataReader("")
    pdf_bytes = reader1.read(pdf_file_name)  # read the pdf content

    # proc
    ## Create Dataset Instance
    ds = PymuDocDataset(pdf_bytes)

    ds.apply(doc_analyze, ocr=True).pipe_ocr_mode(image_writer).dump_md(md_writer, f"{name_without_suff}.md", image_dir)

Running the above code will result in the following


.. code:: bash 

    output/
    ├── abc.md
    └── images


Excluding the setup of the environment, such as creating directories and importing dependencies, the actual code snippet for converting pdf to markdown is as follows


.. code:: python 

    # read bytes
    reader1 = FileBasedDataReader("")
    pdf_bytes = reader1.read(pdf_file_name)  # read the pdf content

    # proc
    ## Create Dataset Instance
    ds = PymuDocDataset(pdf_bytes)

    ds.apply(doc_analyze, ocr=True).pipe_ocr_mode(image_writer).dump_md(md_writer, f"{name_without_suff}.md", image_dir)

``ds.apply(doc_analyze, ocr=True)`` generates an ``InferenceResult`` object. The ``InferenceResult`` object, when executing the ``pipe_ocr_mode`` method, produces a ``PipeResult`` object.
The ``PipeResult`` object, upon executing ``dump_md``, generates a ``markdown`` file at the specified location.


The pipeline execution process is illustrated in the following diagram


.. image:: ../../_static/image/pipeline.drawio.svg 

.. raw:: html

    <br> </br>

Currently, the process is divided into three stages: data, inference, and processing, which correspond to the ``Dataset``, ``InferenceResult``, and ``PipeResult`` entities in the diagram.
These stages are linked together through methods like ``apply``, ``doc_analyze``, or ``pipe_ocr_mode``


.. admonition:: Tip
    :class: tip

    For more detailed information about ``Dataset``, ``InferenceResult``, and ``PipeResult``, please refer to :doc:`../../api/dataset`, :doc:`../../api/model_operators`, :doc:`../../api/pipe_operators`


Pipeline Composition
^^^^^^^^^^^^^^^^^^^^^

.. code:: python 

    class Dataset(ABC):
        @abstractmethod
        def apply(self, proc: Callable, *args, **kwargs):
            """Apply callable method which.

            Args:
                proc (Callable): invoke proc as follows:
                    proc(self, *args, **kwargs)

            Returns:
                Any: return the result generated by proc
            """
            pass

    class InferenceResult(InferenceResultBase):

        def apply(self, proc: Callable, *args, **kwargs):
            """Apply callable method which.

            Args:
                proc (Callable): invoke proc as follows:
                    proc(inference_result, *args, **kwargs)

            Returns:
                Any: return the result generated by proc
            """
            return proc(copy.deepcopy(self._infer_res), *args, **kwargs)

        def pipe_ocr_mode(
            self,
            imageWriter: DataWriter,
            start_page_id=0,
            end_page_id=None,
            debug_mode=False,
            lang=None,
            ) -> PipeResult:
            pass

    class PipeResult:
        def apply(self, proc: Callable, *args, **kwargs):
            """Apply callable method which.

            Args:
                proc (Callable): invoke proc as follows:
                    proc(pipeline_result, *args, **kwargs)

            Returns:
                Any: return the result generated by proc
            """
            return proc(copy.deepcopy(self._pipe_res), *args, **kwargs)


The ``Dataset``, ``InferenceResult``, and ``PipeResult`` classes all have an ``apply`` method, which can be used to chain different stages of the computation. 
As shown below, ``MinerU`` provides a set of methods to compose these classes.


.. code:: python 

    # proc
    ## Create Dataset Instance
    ds = PymuDocDataset(pdf_bytes)

    ds.apply(doc_analyze, ocr=True).pipe_ocr_mode(image_writer).dump_md(md_writer, f"{name_without_suff}.md", image_dir)


Users can implement their own functions for chaining as needed. For example, a user could use the ``apply`` method to create a function that counts the number of pages in a ``pdf`` file.


.. code:: python

    from magic_pdf.data.data_reader_writer import  FileBasedDataReader
    from magic_pdf.data.dataset import PymuDocDataset

    # args
    pdf_file_name = "abc.pdf"  # replace with the real pdf path

    # read bytes
    reader1 = FileBasedDataReader("")
    pdf_bytes = reader1.read(pdf_file_name)  # read the pdf content

    # proc
    ## Create Dataset Instance
    ds = PymuDocDataset(pdf_bytes)

    def count_page(ds)-> int:
        return len(ds)

    print("page number: ", ds.apply(count_page)) # will output the page count of `abc.pdf`