Skip to content

Mathpix Loader

MathpixPDFReader

Bases: BaseReader

Load PDF files using Mathpix service.

Source code in libs/kotaemon/kotaemon/loaders/mathpix_loader.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
class MathpixPDFReader(BaseReader):
    """Load `PDF` files using `Mathpix` service."""

    def __init__(
        self,
        processed_file_format: str = "md",
        max_wait_time_seconds: int = 900,
        should_clean_pdf: bool = True,
        **kwargs: Any,
    ) -> None:
        """Initialize with a file path.

        Args:
            processed_file_format: a format of the processed file. Default is   "mmd".
            max_wait_time_seconds: a maximum time to wait for the response from
                the server. Default is 500.
            should_clean_pdf: a flag to clean the PDF file. Default is False.
            **kwargs: additional keyword arguments.
        """
        self.mathpix_api_key = get_from_dict_or_env(
            kwargs, "mathpix_api_key", "MATHPIX_API_KEY", default="empty"
        )
        self.mathpix_api_id = get_from_dict_or_env(
            kwargs, "mathpix_api_id", "MATHPIX_API_ID", default="empty"
        )
        self.processed_file_format = processed_file_format
        self.max_wait_time_seconds = max_wait_time_seconds
        self.should_clean_pdf = should_clean_pdf
        super().__init__()

    @property
    def _mathpix_headers(self) -> Dict[str, str]:
        return {"app_id": self.mathpix_api_id, "app_key": self.mathpix_api_key}

    @property
    def url(self) -> str:
        return "https://api.mathpix.com/v3/pdf"

    @property
    def data(self) -> dict:
        options = {
            "conversion_formats": {self.processed_file_format: True},
            "enable_tables_fallback": True,
        }
        return {"options_json": json.dumps(options)}

    def send_pdf(self, file_path) -> str:
        with open(file_path, "rb") as f:
            files = {"file": f}
            response = requests.post(
                self.url, headers=self._mathpix_headers, files=files, data=self.data
            )
        response_data = response.json()
        if "pdf_id" in response_data:
            pdf_id = response_data["pdf_id"]
            return pdf_id
        else:
            raise ValueError("Unable to send PDF to Mathpix.")

    def wait_for_processing(self, pdf_id: str) -> None:
        """Wait for processing to complete.

        Args:
            pdf_id: a PDF id.

        Returns: None
        """
        url = self.url + "/" + pdf_id
        for _ in range(0, self.max_wait_time_seconds, 5):
            response = requests.get(url, headers=self._mathpix_headers)
            response_data = response.json()
            status = response_data.get("status", None)
            print(
                f"Processing status: {status},"
                f"Progress: {response_data.get('percent_done', 0)}%"
            )

            if status == "completed":
                return
            elif status == "error":
                raise ValueError(f"Mathpix processing error: {response_data}")
            elif status in [
                "split",
                "processing",
            ]:  # Add handling for processing states
                time.sleep(5)
                continue
            else:
                print(f"Unknown status: {response_data}")
                time.sleep(5)

        raise TimeoutError(
            f"Processing did not complete within {self.max_wait_time_seconds} seconds"
        )

    def get_processed_pdf(self, pdf_id: str) -> str:
        self.wait_for_processing(pdf_id)
        url = f"{self.url}/{pdf_id}.{self.processed_file_format}"
        response = requests.get(url, headers=self._mathpix_headers)
        if response.status_code != 200:
            raise ValueError(f"Failed to get processed PDF: {response.text}")
        content = response.content.decode("utf-8")
        print(f"Retrieved content length: {len(content)}")  # Debug print
        return content

    def clean_pdf(self, contents: str) -> str:
        """Clean the PDF file.

        Args:
            contents: a PDF file contents.

        Returns:

        """
        contents = "\n".join(
            [line for line in contents.split("\n") if not line.startswith("![]")]
        )
        # replace \section{Title} with # Title
        contents = contents.replace("\\section{", "# ")
        # replace the "\" slash that Mathpix adds to escape $, %, (, etc.

        # http:// or https:// followed by anything but a closing paren
        url_regex = "http[s]?://[^)]+"
        markup_regex = r"\[]\(\s*({0})\s*\)".format(url_regex)
        contents = (
            contents.replace(r"\$", "$")
            .replace(r"\%", "%")
            .replace(r"\(", "(")
            .replace(r"\)", ")")
            .replace("$\\begin{array}", "")
            .replace("\\end{array}$", "")
            .replace("\\\\", "")
            .replace("\\text", "")
            .replace("}", "")
            .replace("{", "")
            .replace("\\mathrm", "")
        )
        contents = re.sub(markup_regex, "", contents)
        return contents

    def parse_markdown_text_to_tables(
        self, content: str
    ) -> tuple[list[tuple[int, str]], list[tuple[int, str]]]:
        """Parse markdown text to get tables and texts separately.

        Returns:
            Tuple of (tables, texts) where each is a list of (page_num, content) tuples
        """
        print("Starting markdown parsing...")
        print(f"Content length: {len(content)}")

        # Split by page markers if present
        pages = re.split(r"(?m)^# Page \d+\n", content)

        tables: list[tuple[int, str]] = []
        texts: list[tuple[int, str]] = []

        for page_num, page_content in enumerate(pages, 1):
            if not page_content.strip():
                continue

            # Extract tables from the page
            table_matches = re.findall(r"(\|[^\n]+\|(?:\n\|[^\n]+\|)*)", page_content)
            if table_matches:
                for table in table_matches:
                    tables.append(
                        (page_num, table.strip())
                    )  # Store as tuple with page number
                # Remove tables from page content
                page_content = re.sub(
                    r"(\|[^\n]+\|(?:\n\|[^\n]+\|)*)", "", page_content
                )

            # Split remaining content into meaningful chunks
            chunks = re.split(r"\n\s*\n", page_content)
            for chunk in chunks:
                if chunk.strip():
                    texts.append(
                        (page_num, chunk.strip())
                    )  # Store as tuple with page number

        print(f"Found {len(tables)} tables and {len(texts)} text sections")
        return tables, texts

    def load_data(
        self,
        file: Union[str, List[str], Path],
        extra_info: Optional[Dict] = None,
        **load_kwargs: Any,
    ) -> List[Document]:
        """Load data from file path."""
        file_path = Path(file) if isinstance(file, str) else file

        if "response_content" in load_kwargs:
            content = load_kwargs["response_content"]
        else:
            pdf_id = self.send_pdf(file_path)
            content = self.get_processed_pdf(pdf_id)

        if self.should_clean_pdf:
            content = self.clean_pdf(content)

        tables, texts = self.parse_markdown_text_to_tables(content)
        documents = []

        # Handle tables
        for page_num, table_content in tables:
            text = strip_special_chars_markdown(table_content)
            metadata = {
                "table_origin": table_content,
                "type": "table",
                "page_label": page_num,
                "page_number": page_num,
            }
            if extra_info:
                metadata.update(extra_info)
            documents.append(
                Document(
                    text=text,
                    metadata=metadata,
                    metadata_template="",
                    metadata_seperator="",
                )
            )

        # Handle text sections
        for page_num, text_content in texts:
            if not text_content.strip():
                continue
            metadata = {
                "source": str(file_path),
                "type": "text",
                "page_label": page_num,
                "page_number": page_num,
            }
            if extra_info:
                metadata.update(extra_info)
            documents.append(Document(text=text_content, metadata=metadata))

        # Fallback if no content was parsed
        if not documents and content.strip():
            metadata = {
                "source": str(file_path),
                "type": "text",
                "page_label": 1,
                "page_number": 1,
            }
            if extra_info:
                metadata.update(extra_info)
            documents.append(Document(text=content.strip(), metadata=metadata))

        return documents

    def lazy_load_data(
        self,
        file: Union[str, List[str], Path],
        extra_info: Optional[Dict] = None,
        **load_kwargs: Any,
    ) -> Generator[Document, None, None]:
        """Lazy load data from file path."""
        file_path = Path(file) if isinstance(file, str) else file

        if "response_content" in load_kwargs:
            content = load_kwargs["response_content"]
        else:
            pdf_id = self.send_pdf(file_path)
            print(f"PDF ID: {pdf_id}")
            content = self.get_processed_pdf(pdf_id)

        if self.should_clean_pdf:
            content = self.clean_pdf(content)

        tables, texts = self.parse_markdown_text_to_tables(content)

        # Handle tables
        for page_num, table_content in tables:  # Changed variable name for clarity
            text = strip_special_chars_markdown(table_content)  # Pass just the content
            metadata = {
                "table_origin": table_content,  # Use table_content here too
                "type": "table",
                "page_label": page_num,
                "page_number": page_num,
            }
            if extra_info:
                metadata.update(extra_info)
            yield Document(
                text=text,
                metadata=metadata,
                metadata_template="",
                metadata_seperator="",
            )

        # Handle text sections
        for page_num, text_content in texts:  # Changed variable name for clarity
            if not text_content.strip():
                continue
            metadata = {
                "source": str(file_path),
                "type": "text",
                "page_label": page_num,
                "page_number": page_num,
            }
            if extra_info:
                metadata.update(extra_info)
            yield Document(
                text=text_content, metadata=metadata
            )  # Use text_content directly

        # Fallback if no content was parsed
        if not (tables or texts) and content.strip():
            metadata = {
                "source": str(file_path),
                "type": "text",
                "page_label": 1,
                "page_number": 1,
            }
            if extra_info:
                metadata.update(extra_info)
            yield Document(text=content.strip(), metadata=metadata)

        print(f"Completed processing PDF: {file_path}")

wait_for_processing 

1
wait_for_processing(pdf_id)

Wait for processing to complete.

Parameters:

Name Type Description Default
pdf_id str

a PDF id.

 required

Returns: None

Source code in libs/kotaemon/kotaemon/loaders/mathpix_loader.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def wait_for_processing(self, pdf_id: str) -> None:
    """Wait for processing to complete.

    Args:
        pdf_id: a PDF id.

    Returns: None
    """
    url = self.url + "/" + pdf_id
    for _ in range(0, self.max_wait_time_seconds, 5):
        response = requests.get(url, headers=self._mathpix_headers)
        response_data = response.json()
        status = response_data.get("status", None)
        print(
            f"Processing status: {status},"
            f"Progress: {response_data.get('percent_done', 0)}%"
        )

        if status == "completed":
            return
        elif status == "error":
            raise ValueError(f"Mathpix processing error: {response_data}")
        elif status in [
            "split",
            "processing",
        ]:  # Add handling for processing states
            time.sleep(5)
            continue
        else:
            print(f"Unknown status: {response_data}")
            time.sleep(5)

    raise TimeoutError(
        f"Processing did not complete within {self.max_wait_time_seconds} seconds"
    )

clean_pdf 

1
clean_pdf(contents)

Clean the PDF file.

Parameters:

Name Type Description Default
contents str

a PDF file contents.

 required

Returns:

Source code in libs/kotaemon/kotaemon/loaders/mathpix_loader.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def clean_pdf(self, contents: str) -> str:
    """Clean the PDF file.

    Args:
        contents: a PDF file contents.

    Returns:

    """
    contents = "\n".join(
        [line for line in contents.split("\n") if not line.startswith("![]")]
    )
    # replace \section{Title} with # Title
    contents = contents.replace("\\section{", "# ")
    # replace the "\" slash that Mathpix adds to escape $, %, (, etc.

    # http:// or https:// followed by anything but a closing paren
    url_regex = "http[s]?://[^)]+"
    markup_regex = r"\[]\(\s*({0})\s*\)".format(url_regex)
    contents = (
        contents.replace(r"\$", "$")
        .replace(r"\%", "%")
        .replace(r"\(", "(")
        .replace(r"\)", ")")
        .replace("$\\begin{array}", "")
        .replace("\\end{array}$", "")
        .replace("\\\\", "")
        .replace("\\text", "")
        .replace("}", "")
        .replace("{", "")
        .replace("\\mathrm", "")
    )
    contents = re.sub(markup_regex, "", contents)
    return contents

parse_markdown_text_to_tables 

1
parse_markdown_text_to_tables(content)

Parse markdown text to get tables and texts separately.

Returns:

Type Description
tuple[list[tuple[int, str]], list[tuple[int, str]]]

Tuple of (tables, texts) where each is a list of (page_num, content) tuples
Source code in libs/kotaemon/kotaemon/loaders/mathpix_loader.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def parse_markdown_text_to_tables(
    self, content: str
) -> tuple[list[tuple[int, str]], list[tuple[int, str]]]:
    """Parse markdown text to get tables and texts separately.

    Returns:
        Tuple of (tables, texts) where each is a list of (page_num, content) tuples
    """
    print("Starting markdown parsing...")
    print(f"Content length: {len(content)}")

    # Split by page markers if present
    pages = re.split(r"(?m)^# Page \d+\n", content)

    tables: list[tuple[int, str]] = []
    texts: list[tuple[int, str]] = []

    for page_num, page_content in enumerate(pages, 1):
        if not page_content.strip():
            continue

        # Extract tables from the page
        table_matches = re.findall(r"(\|[^\n]+\|(?:\n\|[^\n]+\|)*)", page_content)
        if table_matches:
            for table in table_matches:
                tables.append(
                    (page_num, table.strip())
                )  # Store as tuple with page number
            # Remove tables from page content
            page_content = re.sub(
                r"(\|[^\n]+\|(?:\n\|[^\n]+\|)*)", "", page_content
            )

        # Split remaining content into meaningful chunks
        chunks = re.split(r"\n\s*\n", page_content)
        for chunk in chunks:
            if chunk.strip():
                texts.append(
                    (page_num, chunk.strip())
                )  # Store as tuple with page number

    print(f"Found {len(tables)} tables and {len(texts)} text sections")
    return tables, texts

load_data 

1
load_data(file, extra_info=None, **load_kwargs)

Load data from file path.

Source code in libs/kotaemon/kotaemon/loaders/mathpix_loader.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def load_data(
    self,
    file: Union[str, List[str], Path],
    extra_info: Optional[Dict] = None,
    **load_kwargs: Any,
) -> List[Document]:
    """Load data from file path."""
    file_path = Path(file) if isinstance(file, str) else file

    if "response_content" in load_kwargs:
        content = load_kwargs["response_content"]
    else:
        pdf_id = self.send_pdf(file_path)
        content = self.get_processed_pdf(pdf_id)

    if self.should_clean_pdf:
        content = self.clean_pdf(content)

    tables, texts = self.parse_markdown_text_to_tables(content)
    documents = []

    # Handle tables
    for page_num, table_content in tables:
        text = strip_special_chars_markdown(table_content)
        metadata = {
            "table_origin": table_content,
            "type": "table",
            "page_label": page_num,
            "page_number": page_num,
        }
        if extra_info:
            metadata.update(extra_info)
        documents.append(
            Document(
                text=text,
                metadata=metadata,
                metadata_template="",
                metadata_seperator="",
            )
        )

    # Handle text sections
    for page_num, text_content in texts:
        if not text_content.strip():
            continue
        metadata = {
            "source": str(file_path),
            "type": "text",
            "page_label": page_num,
            "page_number": page_num,
        }
        if extra_info:
            metadata.update(extra_info)
        documents.append(Document(text=text_content, metadata=metadata))

    # Fallback if no content was parsed
    if not documents and content.strip():
        metadata = {
            "source": str(file_path),
            "type": "text",
            "page_label": 1,
            "page_number": 1,
        }
        if extra_info:
            metadata.update(extra_info)
        documents.append(Document(text=content.strip(), metadata=metadata))

    return documents

lazy_load_data 

1
lazy_load_data(file, extra_info=None, **load_kwargs)

Lazy load data from file path.

Source code in libs/kotaemon/kotaemon/loaders/mathpix_loader.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def lazy_load_data(
    self,
    file: Union[str, List[str], Path],
    extra_info: Optional[Dict] = None,
    **load_kwargs: Any,
) -> Generator[Document, None, None]:
    """Lazy load data from file path."""
    file_path = Path(file) if isinstance(file, str) else file

    if "response_content" in load_kwargs:
        content = load_kwargs["response_content"]
    else:
        pdf_id = self.send_pdf(file_path)
        print(f"PDF ID: {pdf_id}")
        content = self.get_processed_pdf(pdf_id)

    if self.should_clean_pdf:
        content = self.clean_pdf(content)

    tables, texts = self.parse_markdown_text_to_tables(content)

    # Handle tables
    for page_num, table_content in tables:  # Changed variable name for clarity
        text = strip_special_chars_markdown(table_content)  # Pass just the content
        metadata = {
            "table_origin": table_content,  # Use table_content here too
            "type": "table",
            "page_label": page_num,
            "page_number": page_num,
        }
        if extra_info:
            metadata.update(extra_info)
        yield Document(
            text=text,
            metadata=metadata,
            metadata_template="",
            metadata_seperator="",
        )

    # Handle text sections
    for page_num, text_content in texts:  # Changed variable name for clarity
        if not text_content.strip():
            continue
        metadata = {
            "source": str(file_path),
            "type": "text",
            "page_label": page_num,
            "page_number": page_num,
        }
        if extra_info:
            metadata.update(extra_info)
        yield Document(
            text=text_content, metadata=metadata
        )  # Use text_content directly

    # Fallback if no content was parsed
    if not (tables or texts) and content.strip():
        metadata = {
            "source": str(file_path),
            "type": "text",
            "page_label": 1,
            "page_number": 1,
        }
        if extra_info:
            metadata.update(extra_info)
        yield Document(text=content.strip(), metadata=metadata)

    print(f"Completed processing PDF: {file_path}")