menekis/OCRmyPDF
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor pdfinfo: split info.py into focused modules

Browse Source 
Split the 1288-line info.py into smaller, single-responsibility modules:
- _types.py: Enums, type aliases, lookup dictionaries
- _contentstream.py: PDF content stream parsing, DPI calculation
- _image.py: ImageInfo class and image finding functions
- _worker.py: Concurrency/worker process handling
- info.py: PageInfo, PdfInfo classes (reduced to ~530 lines)

Public API unchanged - all existing imports continue to work.
This commit is contained in:
James R. Barlow
2025-12-22 01:27:23 -08:00
parent b4f9673364
commit be425e7405
8 changed files with 853 additions and 773 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ocrmypdf/_pipeline.py
+1 -2
View File
@@ -39,8 +39,7 @@ from ocrmypdf.helpers import IMG2PDF_KWARGS, Resolution, safe_symlink
from ocrmypdf.hocrtransform import DebugRenderOptions, HocrTransform
from ocrmypdf.hocrtransform._font import Courier
from ocrmypdf.pdfa import generate_pdfa_ps
from ocrmypdf.pdfinfo import Colorspace, Encoding, PageInfo, PdfInfo
from ocrmypdf.pdfinfo.info import FloatRect
from ocrmypdf.pdfinfo import Colorspace, Encoding, FloatRect, PageInfo, PdfInfo
from ocrmypdf.pluginspec import OrientationConfidence
try:
src/ocrmypdf/pdfinfo/__init__.py
+3 -2
View File
@@ -6,6 +6,7 @@
from __future__ import annotations
from ocrmypdf.pdfinfo.info import Colorspace, Encoding, PageInfo, PdfInfo
from ocrmypdf.pdfinfo._types import Colorspace, Encoding, FloatRect
from ocrmypdf.pdfinfo.info import PageInfo, PdfInfo
__all__ = ["Colorspace", "Encoding", "PageInfo", "PdfInfo"]
__all__ = ["Colorspace", "Encoding", "FloatRect", "PageInfo", "PdfInfo"]
src/ocrmypdf/pdfinfo/_contentstream.py
+231
View File
@@ -0,0 +1,231 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""PDF content stream interpretation."""
from __future__ import annotations
import re
from collections import defaultdict
from collections.abc import Mapping
from math import hypot, inf, isclose
from typing import NamedTuple
from warnings import warn
from pikepdf import Matrix, Object, PdfInlineImage, parse_content_stream
from ocrmypdf.exceptions import InputFileError
from ocrmypdf.helpers import Resolution
from ocrmypdf.pdfinfo._types import UNIT_SQUARE
class XobjectSettings(NamedTuple):
"""Info about an XObject found in a PDF."""
name: str
shorthand: tuple[float, float, float, float, float, float]
stack_depth: int
class InlineSettings(NamedTuple):
"""Info about an inline image found in a PDF."""
iimage: PdfInlineImage
shorthand: tuple[float, float, float, float, float, float]
stack_depth: int
class ContentsInfo(NamedTuple):
"""Info about various objects found in a PDF."""
xobject_settings: list[XobjectSettings]
inline_images: list[InlineSettings]
found_vector: bool
found_text: bool
name_index: Mapping[str, list[XobjectSettings]]
class TextboxInfo(NamedTuple):
"""Info about a text box found in a PDF."""
bbox: tuple[float, float, float, float]
is_visible: bool
is_corrupt: bool
class VectorMarker:
"""Sentinel indicating vector drawing operations were found on a page."""
class TextMarker:
"""Sentinel indicating text drawing operations were found on a page."""
def _is_unit_square(shorthand):
"""Check if the shorthand represents a unit square transformation."""
values = map(float, shorthand)
pairwise = zip(values, UNIT_SQUARE)
return all(isclose(a, b, rel_tol=1e-3) for a, b in pairwise)
def _normalize_stack(graphobjs):
"""Convert runs of qQ's in the stack into single graphobjs."""
for operands, operator in graphobjs:
operator = str(operator)
if re.match(r'Q*q+$', operator): # Zero or more Q, one or more q
for char in operator: # Split into individual
yield ([], char) # Yield individual
else:
yield (operands, operator)
def _interpret_contents(contentstream: Object, initial_shorthand=UNIT_SQUARE):
"""Interpret the PDF content stream.
The stack represents the state of the PDF graphics stack. We are only
interested in the current transformation matrix (CTM) so we only track
this object; a full implementation would need to track many other items.
The CTM is initialized to the mapping from user space to device space.
PDF units are 1/72". In a PDF viewer or printer this matrix is initialized
to the transformation to device space. For example if set to
(1/72, 0, 0, 1/72, 0, 0) then all units would be calculated in inches.
Images are always considered to be (0, 0) -> (1, 1). Before drawing an
image there should be a 'cm' that sets up an image coordinate system
where drawing from (0, 0) -> (1, 1) will draw on the desired area of the
page.
PDF units suit our needs so we initialize ctm to the identity matrix.
According to the PDF specification, the maximum stack depth is 32. Other
viewers tolerate some amount beyond this. We issue a warning if the
stack depth exceeds the spec limit and set a hard limit beyond this to
bound our memory requirements. If the stack underflows behavior is
undefined in the spec, but we just pretend nothing happened and leave the
CTM unchanged.
"""
stack = []
ctm = Matrix(initial_shorthand)
xobject_settings: list[XobjectSettings] = []
inline_images: list[InlineSettings] = []
name_index = defaultdict(lambda: [])
found_vector = False
found_text = False
vector_ops = set('S s f F f* B B* b b*'.split())
text_showing_ops = set("""TJ Tj " '""".split())
image_ops = set('BI ID EI q Q Do cm'.split())
operator_whitelist = ' '.join(vector_ops | text_showing_ops | image_ops)
for n, graphobj in enumerate(
_normalize_stack(parse_content_stream(contentstream, operator_whitelist))
):
operands, operator = graphobj
if operator == 'q':
stack.append(ctm)
if len(stack) > 32: # See docstring
if len(stack) > 128:
raise RuntimeError(
f"PDF graphics stack overflowed hard limit at operator {n}"
)
warn("PDF graphics stack overflowed spec limit")
elif operator == 'Q':
try:
ctm = stack.pop()
except IndexError:
# Keeping the ctm the same seems to be the only sensible thing
# to do. Just pretend nothing happened, keep calm and carry on.
warn("PDF graphics stack underflowed - PDF may be malformed")
elif operator == 'cm':
try:
ctm = Matrix(operands) @ ctm
except ValueError:
raise InputFileError(
"PDF content stream is corrupt - this PDF is malformed. "
"Use a PDF editor that is capable of visually inspecting the PDF."
)
elif operator == 'Do':
image_name = operands[0]
settings = XobjectSettings(
name=image_name, shorthand=ctm.shorthand, stack_depth=len(stack)
)
xobject_settings.append(settings)
name_index[str(image_name)].append(settings)
elif operator == 'INLINE IMAGE': # BI/ID/EI are grouped into this
iimage = operands[0]
inline = InlineSettings(
iimage=iimage, shorthand=ctm.shorthand, stack_depth=len(stack)
)
inline_images.append(inline)
elif operator in vector_ops:
found_vector = True
elif operator in text_showing_ops:
found_text = True
return ContentsInfo(
xobject_settings=xobject_settings,
inline_images=inline_images,
found_vector=found_vector,
found_text=found_text,
name_index=name_index,
)
def _get_dpi(ctm_shorthand, image_size) -> Resolution:
"""Given the transformation matrix and image size, find the image DPI.
PDFs do not include image resolution information within image data.
Instead, the PDF page content stream describes the location where the
image will be rasterized, and the effective resolution is the ratio of the
pixel size to raster target size.
Normally a scanned PDF has the paper size set appropriately but this is
not guaranteed. The most common case is a cropped image will change the
page size (/CropBox) without altering the page content stream. That means
it is not sufficient to assume that the image fills the page, even though
that is the most common case.
A PDF image may be scaled (always), cropped, translated, rotated in place
to an arbitrary angle (rarely) and skewed. Only equal area mappings can
be expressed, that is, it is not necessary to consider distortions where
the effective DPI varies with position.
To determine the image scale, transform an offset axis vector v0 (0, 0),
width-axis vector v0 (1, 0), height-axis vector vh (0, 1) with the matrix,
which gives the dimensions of the image in PDF units. From there we can
compare to actual image dimensions. PDF uses
row vector * matrix_transposed unlike the traditional
matrix * column vector.
The offset, width and height vectors can be combined in a matrix and
multiplied by the transform matrix. Then we want to calculated
magnitude(width_vector - offset_vector)
and
magnitude(height_vector - offset_vector)
When the above is worked out algebraically, the effect of translation
cancels out, and the vector magnitudes become functions of the nonzero
transformation matrix indices. The results of the derivation are used
in this code.
pdfimages -list does calculate the DPI in some way that is not completely
naive, but it does not get the DPI of rotated images right, so cannot be
used anymore to validate this. Photoshop works, or using Acrobat to
rotate the image back to normal.
It does not matter if the image is partially cropped, or even out of the
/MediaBox.
"""
a, b, c, d, _, _ = ctm_shorthand # pylint: disable=invalid-name
# Calculate the width and height of the image in PDF units
image_drawn = hypot(a, b), hypot(c, d)
def calc(drawn, pixels, inches_per_pt=72.0):
# The scale of the image is pixels per unit of default user space (1/72")
scale = pixels / drawn if drawn != 0 else inf
dpi = scale * inches_per_pt
return dpi
dpi_w, dpi_h = (calc(image_drawn[n], image_size[n]) for n in range(2))
return Resolution(dpi_w, dpi_h)
src/ocrmypdf/pdfinfo/_image.py
+378
View File
@@ -0,0 +1,378 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""PDF image analysis."""
from __future__ import annotations
import logging
from collections.abc import Iterator
from decimal import Decimal
from pikepdf import (
Dictionary,
Matrix,
Name,
Object,
Pdf,
PdfImage,
PdfInlineImage,
Stream,
UnsupportedImageTypeError,
)
from ocrmypdf.helpers import Resolution
from ocrmypdf.pdfinfo._contentstream import (
ContentsInfo,
TextMarker,
VectorMarker,
_get_dpi,
_interpret_contents,
_is_unit_square,
)
from ocrmypdf.pdfinfo._types import (
FRIENDLY_COLORSPACE,
FRIENDLY_COMP,
FRIENDLY_ENCODING,
UNIT_SQUARE,
Colorspace,
Encoding,
)
logger = logging.getLogger()
class ImageInfo:
"""Information about an image found in a PDF.
This gathers information from pikepdf and pdfminer.six, and is pickle-able
so that it can be passed to a worker process, unlike objects from those
libraries.
"""
DPI_PREC = Decimal('1.000')
_comp: int | None
_name: str
def __init__(
self,
*,
name='',
pdfimage: Object | None = None,
inline: PdfInlineImage | None = None,
shorthand=None,
):
"""Initialize an ImageInfo."""
self._name = str(name)
self._shorthand = shorthand
pim: PdfInlineImage | PdfImage
if inline is not None:
self._origin = 'inline'
pim = inline
elif pdfimage is not None and isinstance(pdfimage, Stream):
self._origin = 'xobject'
pim = PdfImage(pdfimage)
else:
raise ValueError("Either pdfimage or inline must be set")
self._width = pim.width
self._height = pim.height
if (smask := pim.obj.get(Name.SMask, None)) is not None:
# SMask is pretty much an alpha channel, but in PDF it's possible
# for channel to have different dimensions than the image
# itself. Some PDF writers use this to create a grayscale stencil
# mask. For our purposes, the effective size is the size of the
# larger component (image or smask).
if isinstance(smask, Stream | Dictionary):
self._width = max(smask.get(Name.Width, 0), self._width)
self._height = max(smask.get(Name.Height, 0), self._height)
if (mask := pim.obj.get(Name.Mask, None)) is not None:
# If the image has a /Mask entry, it has an explicit mask.
# /Mask can be a Stream or an Array. If it's a Stream,
# use its /Width and /Height if they are larger than the main
# image's.
if isinstance(mask, Stream | Dictionary):
self._width = max(mask.get(Name.Width, 0), self._width)
self._height = max(mask.get(Name.Height, 0), self._height)
# If /ImageMask is true, then this image is a stencil mask
# (Images that draw with this stencil mask will have a reference to
# it in their /Mask, but we don't actually need that information)
if pim.image_mask:
self._type = 'stencil'
else:
self._type = 'image'
self._bpc = int(pim.bits_per_component)
try:
self._enc = FRIENDLY_ENCODING.get(pim.filters[0])
except IndexError:
self._enc = None
try:
self._color = FRIENDLY_COLORSPACE.get(pim.colorspace or '')
except NotImplementedError:
self._color = None
if self._enc == Encoding.jpeg2000:
self._color = Colorspace.jpeg2000
self._comp = None
if self._color == Colorspace.icc and isinstance(pim, PdfImage):
self._comp = self._init_icc(pim)
else:
if isinstance(self._color, Colorspace):
self._comp = FRIENDLY_COMP.get(self._color)
# Bit of a hack... infer grayscale if component count is uncertain
# but encoding only supports monochrome.
if self._comp is None and self._enc in (Encoding.ccitt, Encoding.jbig2):
self._comp = FRIENDLY_COMP[Colorspace.gray]
def _init_icc(self, pim: PdfImage):
try:
icc = pim.icc
except UnsupportedImageTypeError as e:
logger.warning(
f"An image with a corrupt or unreadable ICC profile was found. "
f"Output PDF may not match the input PDF visually: {e}. {self}"
)
return None
# Check the ICC profile to determine actual colorspace
if icc is None or not hasattr(icc, 'profile'):
logger.warning(
f"An image with an ICC profile but no ICC profile data was found. "
f"The output PDF may not match the input PDF visually. {self}"
)
return None
try:
if icc.profile.xcolor_space == 'GRAY':
return 1
elif icc.profile.xcolor_space == 'CMYK':
return 4
else:
return 3
except AttributeError:
return None
@property
def name(self):
"""Name of the image as it appears in the PDF."""
return self._name
@property
def type_(self):
"""Type of image, either 'image' or 'stencil'."""
return self._type
@property
def width(self) -> int:
"""Width of the image in pixels."""
return self._width
@property
def height(self) -> int:
"""Height of the image in pixels."""
return self._height
@property
def bpc(self):
"""Bits per component."""
return self._bpc
@property
def color(self):
"""Colorspace of the image."""
return self._color if self._color is not None else '?'
@property
def comp(self):
"""Number of components/channels in the image."""
return self._comp if self._comp is not None else '?'
@property
def enc(self):
"""Encoding of the image."""
return self._enc if self._enc is not None else 'image'
@property
def renderable(self) -> bool:
"""Whether the image is renderable.
Some PDFs in the wild have invalid images that are not renderable,
due to unusual dimensions.
Stencil masks are not also not renderable, since they are not
drawn, but rather they control how rendering happens.
"""
return (
self.dpi.is_finite
and self.width >= 0
and self.height >= 0
and self.type_ != 'stencil'
)
@property
def dpi(self) -> Resolution:
"""Dots per inch of the image.
Calculated based on where and how the image is drawn in the PDF.
"""
return _get_dpi(self._shorthand, (self._width, self._height))
@property
def printed_area(self) -> float:
"""Physical area of the image in square inches."""
if not self.renderable:
return 0.0
return float((self.width / self.dpi.x) * (self.height / self.dpi.y))
def __repr__(self):
"""Return a string representation of the image."""
return (
f"<ImageInfo '{self.name}' {self.type_} {self.width}×{self.height} "
f"{self.color} {self.comp} {self.bpc} {self.enc} {self.dpi}>"
)
def _find_inline_images(contentsinfo: ContentsInfo) -> Iterator[ImageInfo]:
"""Find inline images in the contentstream."""
for n, inline in enumerate(contentsinfo.inline_images):
yield ImageInfo(
name=f'inline-{n:02d}', shorthand=inline.shorthand, inline=inline.iimage
)
def _image_xobjects(container) -> Iterator[tuple[Object, str]]:
"""Search for all XObject-based images in the container.
Usually the container is a page, but it could also be a Form XObject
that contains images. Filter out the Form XObjects which are dealt with
elsewhere.
Generate a sequence of tuples (image, xobj container), where container,
where xobj is the name of the object and image is the object itself,
since the object does not know its own name.
"""
if Name.Resources not in container:
return
resources = container[Name.Resources]
if Name.XObject not in resources:
return
for key, candidate in resources[Name.XObject].items():
if candidate is None or Name.Subtype not in candidate:
continue
if candidate[Name.Subtype] == Name.Image:
pdfimage = candidate
yield (pdfimage, key)
def _find_regular_images(
container: Object, contentsinfo: ContentsInfo
) -> Iterator[ImageInfo]:
"""Find images stored in the container's /Resources /XObject.
Usually the container is a page, but it could also be a Form XObject
that contains images.
Generates images with their DPI at time of drawing.
"""
for pdfimage, xobj in _image_xobjects(container):
if xobj not in contentsinfo.name_index:
continue
for draw in contentsinfo.name_index[xobj]:
if draw.stack_depth == 0 and _is_unit_square(draw.shorthand):
# At least one PDF in the wild (and test suite) draws an image
# when the graphics stack depth is 0, meaning that the image
# gets drawn into a square of 1x1 PDF units (or 1/72",
# or 0.35 mm). The equivalent DPI will be >100,000. Exclude
# these from our DPI calculation for the page.
continue
yield ImageInfo(name=draw.name, pdfimage=pdfimage, shorthand=draw.shorthand)
def _find_form_xobject_images(pdf: Pdf, container: Object, contentsinfo: ContentsInfo):
"""Find any images that are in Form XObjects in the container.
The container may be a page, or a parent Form XObject.
"""
if Name.Resources not in container:
return
resources = container[Name.Resources]
if Name.XObject not in resources:
return
xobjs = resources[Name.XObject].as_dict()
for xobj in xobjs:
candidate = xobjs[xobj]
if candidate is None or candidate.get(Name.Subtype) != Name.Form:
continue
form_xobject = candidate
for settings in contentsinfo.xobject_settings:
if settings.name != xobj:
continue
# Find images once for each time this Form XObject is drawn.
# This could be optimized to cache the multiple drawing events
# but in practice both Form XObjects and multiple drawing of the
# same object are both very rare.
ctm_shorthand = settings.shorthand
yield from _process_content_streams(
pdf=pdf, container=form_xobject, shorthand=ctm_shorthand
)
def _process_content_streams(
*, pdf: Pdf, container: Object, shorthand=None
) -> Iterator[VectorMarker | TextMarker | ImageInfo]:
"""Find all individual instances of images drawn in the container.
Usually the container is a page, but it may also be a Form XObject.
On a typical page images are stored inline or as regular images
in an XObject.
Form XObjects may include inline images, XObject images,
and recursively, other Form XObjects; and also vector graphic objects.
Every instance of an image being drawn somewhere is flattened and
treated as a unique image, since if the same image is drawn multiple times
on one page it may be drawn at differing resolutions, and our objective
is to find the resolution at which the page can be rastered without
downsampling.
"""
if container.get(Name.Type) == Name.Page and Name.Contents in container:
initial_shorthand = shorthand or UNIT_SQUARE
elif (
container.get(Name.Type) == Name.XObject
and container[Name.Subtype] == Name.Form
):
# Set the CTM to the state it was when the "Do" operator was
# encountered that is drawing this instance of the Form XObject
ctm = Matrix(shorthand) if shorthand else Matrix()
# A Form XObject may provide its own matrix to map form space into
# user space. Get this if one exists
form_shorthand = container.get(Name.Matrix, Matrix())
form_matrix = Matrix(form_shorthand)
# Concatenate form matrix with CTM to ensure CTM is correct for
# drawing this instance of the XObject
ctm = form_matrix @ ctm
initial_shorthand = ctm.shorthand
else:
return
contentsinfo = _interpret_contents(container, initial_shorthand)
if contentsinfo.found_vector:
yield VectorMarker()
if contentsinfo.found_text:
yield TextMarker()
yield from _find_inline_images(contentsinfo)
yield from _find_regular_images(container, contentsinfo)
yield from _find_form_xobject_images(pdf, container, contentsinfo)
src/ocrmypdf/pdfinfo/_types.py
+83
View File
@@ -0,0 +1,83 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""PDF type definitions and constants."""
from __future__ import annotations
from enum import Enum, auto
class Colorspace(Enum):
"""Description of common image colorspaces in a PDF."""
# pylint: disable=invalid-name
gray = auto()
rgb = auto()
cmyk = auto()
lab = auto()
icc = auto()
index = auto()
sep = auto()
devn = auto()
pattern = auto()
jpeg2000 = auto()
class Encoding(Enum):
"""Description of common image encodings in a PDF."""
# pylint: disable=invalid-name
ccitt = auto()
jpeg = auto()
jpeg2000 = auto()
jbig2 = auto()
asciihex = auto()
ascii85 = auto()
lzw = auto()
flate = auto()
runlength = auto()
FloatRect = tuple[float, float, float, float]
FRIENDLY_COLORSPACE: dict[str, Colorspace] = {
'/DeviceGray': Colorspace.gray,
'/CalGray': Colorspace.gray,
'/DeviceRGB': Colorspace.rgb,
'/CalRGB': Colorspace.rgb,
'/DeviceCMYK': Colorspace.cmyk,
'/Lab': Colorspace.lab,
'/ICCBased': Colorspace.icc,
'/Indexed': Colorspace.index,
'/Separation': Colorspace.sep,
'/DeviceN': Colorspace.devn,
'/Pattern': Colorspace.pattern,
'/G': Colorspace.gray, # Abbreviations permitted in inline images
'/RGB': Colorspace.rgb,
'/CMYK': Colorspace.cmyk,
'/I': Colorspace.index,
}
FRIENDLY_ENCODING: dict[str, Encoding] = {
'/CCITTFaxDecode': Encoding.ccitt,
'/DCTDecode': Encoding.jpeg,
'/JPXDecode': Encoding.jpeg2000,
'/JBIG2Decode': Encoding.jbig2,
'/CCF': Encoding.ccitt, # Abbreviations permitted in inline images
'/DCT': Encoding.jpeg,
'/AHx': Encoding.asciihex,
'/A85': Encoding.ascii85,
'/LZW': Encoding.lzw,
'/Fl': Encoding.flate,
'/RL': Encoding.runlength,
}
FRIENDLY_COMP: dict[Colorspace, int] = {
Colorspace.gray: 1,
Colorspace.rgb: 3,
Colorspace.cmyk: 4,
Colorspace.lab: 3,
Colorspace.index: 1,
}
UNIT_SQUARE = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
src/ocrmypdf/pdfinfo/_worker.py
+143
View File
@@ -0,0 +1,143 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""PDF page info worker process handling."""
from __future__ import annotations
import atexit
import logging
from collections.abc import Container, Sequence
from contextlib import contextmanager
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING
from pikepdf import Pdf
from ocrmypdf._concurrent import Executor
from ocrmypdf._progressbar import ProgressBar
from ocrmypdf.exceptions import InputFileError
from ocrmypdf.helpers import available_cpu_count, pikepdf_enable_mmap
if TYPE_CHECKING:
from ocrmypdf.pdfinfo.info import PageInfo
from ocrmypdf.pdfinfo.layout import PdfMinerState
logger = logging.getLogger()
worker_pdf = None # pylint: disable=invalid-name
def _pdf_pageinfo_sync_init(pdf: Pdf, infile: Path, pdfminer_loglevel):
global worker_pdf # pylint: disable=global-statement,invalid-name
pikepdf_enable_mmap()
logging.getLogger('pdfminer').setLevel(pdfminer_loglevel)
# If the pdf is not opened, open a copy for our worker process to use
if pdf is None:
worker_pdf = Pdf.open(infile)
def on_process_close():
worker_pdf.close()
# Close when this process exits
atexit.register(on_process_close)
@contextmanager
def _pdf_pageinfo_sync_pdf(thread_pdf: Pdf | None, infile: Path):
if thread_pdf is not None:
yield thread_pdf
elif worker_pdf is not None:
yield worker_pdf
else:
with Pdf.open(infile) as pdf:
yield pdf
def _pdf_pageinfo_sync(
pageno: int,
thread_pdf: Pdf | None,
infile: Path,
check_pages: Container[int],
detailed_analysis: bool,
miner_state: PdfMinerState | None,
) -> PageInfo:
# Import here to avoid circular import - info.py imports this module,
# but PageInfo is defined in info.py
from ocrmypdf.pdfinfo.info import PageInfo
with _pdf_pageinfo_sync_pdf(thread_pdf, infile) as pdf:
return PageInfo(
pdf, pageno, infile, check_pages, detailed_analysis, miner_state
)
def _pdf_pageinfo_concurrent(
pdf,
executor: Executor,
max_workers: int,
use_threads: bool,
infile,
progbar,
check_pages,
detailed_analysis: bool = False,
miner_state: PdfMinerState | None = None,
) -> Sequence[PageInfo | None]:
pages: list[PageInfo | None] = [None] * len(pdf.pages)
def update_pageinfo(page: PageInfo, pbar: ProgressBar):
if not page:
raise InputFileError("Could read a page in the PDF")
pages[page.pageno] = page
pbar.update()
if max_workers is None:
max_workers = available_cpu_count()
total = len(pdf.pages)
n_workers = min(1 + len(pages) // 4, max_workers)
if n_workers == 1:
# If we decided on only one worker, there is no point in using
# a separate process.
use_threads = True
if use_threads and n_workers > 1:
# If we are using threads, there is no point in using more than one
# worker thread - they will just fight over the GIL.
n_workers = 1
# If we use a thread, we can pass the already-open Pdf for them to use
# If we use processes, we pass a None which tells the init function to open its
# own
initial_pdf = pdf if use_threads else None
contexts = (
(n, initial_pdf, infile, check_pages, detailed_analysis, miner_state)
for n in range(total)
)
assert n_workers == 1 if use_threads else n_workers >= 1, "Not multithreadable"
logger.debug(
f"Gathering info with {n_workers} "
+ ('thread' if use_threads else 'process')
+ " workers"
)
executor(
use_threads=use_threads,
max_workers=n_workers,
progress_kwargs=dict(
total=total, desc="Scanning contents", unit='page', disable=not progbar
),
worker_initializer=partial(
_pdf_pageinfo_sync_init,
initial_pdf,
infile,
logging.getLogger('pdfminer').level,
),
task=_pdf_pageinfo_sync,
task_arguments=contexts,
task_finished=update_pageinfo,
)
return pages
src/ocrmypdf/pdfinfo/info.py
+10 -766
View File
@@ -6,41 +6,25 @@
from __future__ import annotations
import atexit
import logging
import re
import statistics
from collections import defaultdict
from collections.abc import Callable, Container, Iterable, Iterator, Mapping, Sequence
from contextlib import contextmanager, nullcontext
from collections.abc import Callable, Container, Iterable, Iterator
from contextlib import nullcontext
from decimal import Decimal
from enum import Enum, auto
from functools import partial
from math import hypot, inf, isclose
from os import PathLike
from pathlib import Path
from typing import NamedTuple
from warnings import warn
from pdfminer.layout import LTPage, LTTextBox
from pikepdf import (
Dictionary,
Matrix,
Name,
Object,
Page,
Pdf,
PdfImage,
PdfInlineImage,
Stream,
UnsupportedImageTypeError,
parse_content_stream,
)
from pikepdf import Name, Page, Pdf
from ocrmypdf._concurrent import Executor, SerialExecutor
from ocrmypdf._progressbar import ProgressBar
from ocrmypdf.exceptions import EncryptedPdfError, InputFileError
from ocrmypdf.helpers import Resolution, available_cpu_count, pikepdf_enable_mmap
from ocrmypdf.exceptions import EncryptedPdfError
from ocrmypdf.helpers import Resolution
from ocrmypdf.pdfinfo._contentstream import TextboxInfo, TextMarker, VectorMarker
from ocrmypdf.pdfinfo._image import ImageInfo, _process_content_streams
from ocrmypdf.pdfinfo._types import FloatRect
from ocrmypdf.pdfinfo._worker import _pdf_pageinfo_concurrent
from ocrmypdf.pdfinfo.layout import (
LTStateAwareChar,
PdfMinerState,
@@ -50,632 +34,6 @@ from ocrmypdf.pdfinfo.layout import (
logger = logging.getLogger()
class Colorspace(Enum):
"""Description of common image colorspaces in a PDF."""
# pylint: disable=invalid-name
gray = auto()
rgb = auto()
cmyk = auto()
lab = auto()
icc = auto()
index = auto()
sep = auto()
devn = auto()
pattern = auto()
jpeg2000 = auto()
class Encoding(Enum):
"""Description of common image encodings in a PDF."""
# pylint: disable=invalid-name
ccitt = auto()
jpeg = auto()
jpeg2000 = auto()
jbig2 = auto()
asciihex = auto()
ascii85 = auto()
lzw = auto()
flate = auto()
runlength = auto()
FloatRect = tuple[float, float, float, float]
FRIENDLY_COLORSPACE: dict[str, Colorspace] = {
'/DeviceGray': Colorspace.gray,
'/CalGray': Colorspace.gray,
'/DeviceRGB': Colorspace.rgb,
'/CalRGB': Colorspace.rgb,
'/DeviceCMYK': Colorspace.cmyk,
'/Lab': Colorspace.lab,
'/ICCBased': Colorspace.icc,
'/Indexed': Colorspace.index,
'/Separation': Colorspace.sep,
'/DeviceN': Colorspace.devn,
'/Pattern': Colorspace.pattern,
'/G': Colorspace.gray, # Abbreviations permitted in inline images
'/RGB': Colorspace.rgb,
'/CMYK': Colorspace.cmyk,
'/I': Colorspace.index,
}
FRIENDLY_ENCODING: dict[str, Encoding] = {
'/CCITTFaxDecode': Encoding.ccitt,
'/DCTDecode': Encoding.jpeg,
'/JPXDecode': Encoding.jpeg2000,
'/JBIG2Decode': Encoding.jbig2,
'/CCF': Encoding.ccitt, # Abbreviations permitted in inline images
'/DCT': Encoding.jpeg,
'/AHx': Encoding.asciihex,
'/A85': Encoding.ascii85,
'/LZW': Encoding.lzw,
'/Fl': Encoding.flate,
'/RL': Encoding.runlength,
}
FRIENDLY_COMP: dict[Colorspace, int] = {
Colorspace.gray: 1,
Colorspace.rgb: 3,
Colorspace.cmyk: 4,
Colorspace.lab: 3,
Colorspace.index: 1,
}
UNIT_SQUARE = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
def _is_unit_square(shorthand):
values = map(float, shorthand)
pairwise = zip(values, UNIT_SQUARE)
return all(isclose(a, b, rel_tol=1e-3) for a, b in pairwise)
class XobjectSettings(NamedTuple):
"""Info about an XObject found in a PDF."""
name: str
shorthand: tuple[float, float, float, float, float, float]
stack_depth: int
class InlineSettings(NamedTuple):
"""Info about an inline image found in a PDF."""
iimage: PdfInlineImage
shorthand: tuple[float, float, float, float, float, float]
stack_depth: int
class ContentsInfo(NamedTuple):
"""Info about various objects found in a PDF."""
xobject_settings: list[XobjectSettings]
inline_images: list[InlineSettings]
found_vector: bool
found_text: bool
name_index: Mapping[str, list[XobjectSettings]]
class TextboxInfo(NamedTuple):
"""Info about a text box found in a PDF."""
bbox: tuple[float, float, float, float]
is_visible: bool
is_corrupt: bool
class VectorMarker:
"""Sentinel indicating vector drawing operations were found on a page."""
class TextMarker:
"""Sentinel indicating text drawing operations were found on a page."""
def _normalize_stack(graphobjs):
"""Convert runs of qQ's in the stack into single graphobjs."""
for operands, operator in graphobjs:
operator = str(operator)
if re.match(r'Q*q+$', operator): # Zero or more Q, one or more q
for char in operator: # Split into individual
yield ([], char) # Yield individual
else:
yield (operands, operator)
def _interpret_contents(contentstream: Object, initial_shorthand=UNIT_SQUARE):
"""Interpret the PDF content stream.
The stack represents the state of the PDF graphics stack. We are only
interested in the current transformation matrix (CTM) so we only track
this object; a full implementation would need to track many other items.
The CTM is initialized to the mapping from user space to device space.
PDF units are 1/72". In a PDF viewer or printer this matrix is initialized
to the transformation to device space. For example if set to
(1/72, 0, 0, 1/72, 0, 0) then all units would be calculated in inches.
Images are always considered to be (0, 0) -> (1, 1). Before drawing an
image there should be a 'cm' that sets up an image coordinate system
where drawing from (0, 0) -> (1, 1) will draw on the desired area of the
page.
PDF units suit our needs so we initialize ctm to the identity matrix.
According to the PDF specification, the maximum stack depth is 32. Other
viewers tolerate some amount beyond this. We issue a warning if the
stack depth exceeds the spec limit and set a hard limit beyond this to
bound our memory requirements. If the stack underflows behavior is
undefined in the spec, but we just pretend nothing happened and leave the
CTM unchanged.
"""
stack = []
ctm = Matrix(initial_shorthand)
xobject_settings: list[XobjectSettings] = []
inline_images: list[InlineSettings] = []
name_index = defaultdict(lambda: [])
found_vector = False
found_text = False
vector_ops = set('S s f F f* B B* b b*'.split())
text_showing_ops = set("""TJ Tj " '""".split())
image_ops = set('BI ID EI q Q Do cm'.split())
operator_whitelist = ' '.join(vector_ops | text_showing_ops | image_ops)
for n, graphobj in enumerate(
_normalize_stack(parse_content_stream(contentstream, operator_whitelist))
):
operands, operator = graphobj
if operator == 'q':
stack.append(ctm)
if len(stack) > 32: # See docstring
if len(stack) > 128:
raise RuntimeError(
f"PDF graphics stack overflowed hard limit at operator {n}"
)
warn("PDF graphics stack overflowed spec limit")
elif operator == 'Q':
try:
ctm = stack.pop()
except IndexError:
# Keeping the ctm the same seems to be the only sensible thing
# to do. Just pretend nothing happened, keep calm and carry on.
warn("PDF graphics stack underflowed - PDF may be malformed")
elif operator == 'cm':
try:
ctm = Matrix(operands) @ ctm
except ValueError:
raise InputFileError(
"PDF content stream is corrupt - this PDF is malformed. "
"Use a PDF editor that is capable of visually inspecting the PDF."
)
elif operator == 'Do':
image_name = operands[0]
settings = XobjectSettings(
name=image_name, shorthand=ctm.shorthand, stack_depth=len(stack)
)
xobject_settings.append(settings)
name_index[str(image_name)].append(settings)
elif operator == 'INLINE IMAGE': # BI/ID/EI are grouped into this
iimage = operands[0]
inline = InlineSettings(
iimage=iimage, shorthand=ctm.shorthand, stack_depth=len(stack)
)
inline_images.append(inline)
elif operator in vector_ops:
found_vector = True
elif operator in text_showing_ops:
found_text = True
return ContentsInfo(
xobject_settings=xobject_settings,
inline_images=inline_images,
found_vector=found_vector,
found_text=found_text,
name_index=name_index,
)
def _get_dpi(ctm_shorthand, image_size) -> Resolution:
"""Given the transformation matrix and image size, find the image DPI.
PDFs do not include image resolution information within image data.
Instead, the PDF page content stream describes the location where the
image will be rasterized, and the effective resolution is the ratio of the
pixel size to raster target size.
Normally a scanned PDF has the paper size set appropriately but this is
not guaranteed. The most common case is a cropped image will change the
page size (/CropBox) without altering the page content stream. That means
it is not sufficient to assume that the image fills the page, even though
that is the most common case.
A PDF image may be scaled (always), cropped, translated, rotated in place
to an arbitrary angle (rarely) and skewed. Only equal area mappings can
be expressed, that is, it is not necessary to consider distortions where
the effective DPI varies with position.
To determine the image scale, transform an offset axis vector v0 (0, 0),
width-axis vector v0 (1, 0), height-axis vector vh (0, 1) with the matrix,
which gives the dimensions of the image in PDF units. From there we can
compare to actual image dimensions. PDF uses
row vector * matrix_transposed unlike the traditional
matrix * column vector.
The offset, width and height vectors can be combined in a matrix and
multiplied by the transform matrix. Then we want to calculated
magnitude(width_vector - offset_vector)
and
magnitude(height_vector - offset_vector)
When the above is worked out algebraically, the effect of translation
cancels out, and the vector magnitudes become functions of the nonzero
transformation matrix indices. The results of the derivation are used
in this code.
pdfimages -list does calculate the DPI in some way that is not completely
naive, but it does not get the DPI of rotated images right, so cannot be
used anymore to validate this. Photoshop works, or using Acrobat to
rotate the image back to normal.
It does not matter if the image is partially cropped, or even out of the
/MediaBox.
"""
a, b, c, d, _, _ = ctm_shorthand # pylint: disable=invalid-name
# Calculate the width and height of the image in PDF units
image_drawn = hypot(a, b), hypot(c, d)
def calc(drawn, pixels, inches_per_pt=72.0):
# The scale of the image is pixels per unit of default user space (1/72")
scale = pixels / drawn if drawn != 0 else inf
dpi = scale * inches_per_pt
return dpi
dpi_w, dpi_h = (calc(image_drawn[n], image_size[n]) for n in range(2))
return Resolution(dpi_w, dpi_h)
class ImageInfo:
"""Information about an image found in a PDF.
This gathers information from pikepdf and pdfminer.six, and is pickle-able
so that it can be passed to a worker process, unlike objects from those
libraries.
"""
DPI_PREC = Decimal('1.000')
_comp: int | None
_name: str
def __init__(
self,
*,
name='',
pdfimage: Object | None = None,
inline: PdfInlineImage | None = None,
shorthand=None,
):
"""Initialize an ImageInfo."""
self._name = str(name)
self._shorthand = shorthand
pim: PdfInlineImage | PdfImage
if inline is not None:
self._origin = 'inline'
pim = inline
elif pdfimage is not None and isinstance(pdfimage, Stream):
self._origin = 'xobject'
pim = PdfImage(pdfimage)
else:
raise ValueError("Either pdfimage or inline must be set")
self._width = pim.width
self._height = pim.height
if (smask := pim.obj.get(Name.SMask, None)) is not None:
# SMask is pretty much an alpha channel, but in PDF it's possible
# for channel to have different dimensions than the image
# itself. Some PDF writers use this to create a grayscale stencil
# mask. For our purposes, the effective size is the size of the
# larger component (image or smask).
if isinstance(smask, Stream | Dictionary):
self._width = max(smask.get(Name.Width, 0), self._width)
self._height = max(smask.get(Name.Height, 0), self._height)
if (mask := pim.obj.get(Name.Mask, None)) is not None:
# If the image has a /Mask entry, it has an explicit mask.
# /Mask can be a Stream or an Array. If it's a Stream,
# use its /Width and /Height if they are larger than the main
# image's.
if isinstance(mask, Stream | Dictionary):
self._width = max(mask.get(Name.Width, 0), self._width)
self._height = max(mask.get(Name.Height, 0), self._height)
# If /ImageMask is true, then this image is a stencil mask
# (Images that draw with this stencil mask will have a reference to
# it in their /Mask, but we don't actually need that information)
if pim.image_mask:
self._type = 'stencil'
else:
self._type = 'image'
self._bpc = int(pim.bits_per_component)
try:
self._enc = FRIENDLY_ENCODING.get(pim.filters[0])
except IndexError:
self._enc = None
try:
self._color = FRIENDLY_COLORSPACE.get(pim.colorspace or '')
except NotImplementedError:
self._color = None
if self._enc == Encoding.jpeg2000:
self._color = Colorspace.jpeg2000
self._comp = None
if self._color == Colorspace.icc and isinstance(pim, PdfImage):
self._comp = self._init_icc(pim)
else:
if isinstance(self._color, Colorspace):
self._comp = FRIENDLY_COMP.get(self._color)
# Bit of a hack... infer grayscale if component count is uncertain
# but encoding only supports monochrome.
if self._comp is None and self._enc in (Encoding.ccitt, Encoding.jbig2):
self._comp = FRIENDLY_COMP[Colorspace.gray]
def _init_icc(self, pim: PdfImage):
try:
icc = pim.icc
except UnsupportedImageTypeError as e:
logger.warning(
f"An image with a corrupt or unreadable ICC profile was found. "
f"Output PDF may not match the input PDF visually: {e}. {self}"
)
return None
# Check the ICC profile to determine actual colorspace
if icc is None or not hasattr(icc, 'profile'):
logger.warning(
f"An image with an ICC profile but no ICC profile data was found. "
f"The output PDF may not match the input PDF visually. {self}"
)
return None
try:
if icc.profile.xcolor_space == 'GRAY':
return 1
elif icc.profile.xcolor_space == 'CMYK':
return 4
else:
return 3
except AttributeError:
return None
@property
def name(self):
"""Name of the image as it appears in the PDF."""
return self._name
@property
def type_(self):
"""Type of image, either 'image' or 'stencil'."""
return self._type
@property
def width(self) -> int:
"""Width of the image in pixels."""
return self._width
@property
def height(self) -> int:
"""Height of the image in pixels."""
return self._height
@property
def bpc(self):
"""Bits per component."""
return self._bpc
@property
def color(self):
"""Colorspace of the image."""
return self._color if self._color is not None else '?'
@property
def comp(self):
"""Number of components/channels in the image."""
return self._comp if self._comp is not None else '?'
@property
def enc(self):
"""Encoding of the image."""
return self._enc if self._enc is not None else 'image'
@property
def renderable(self) -> bool:
"""Whether the image is renderable.
Some PDFs in the wild have invalid images that are not renderable,
due to unusual dimensions.
Stencil masks are not also not renderable, since they are not
drawn, but rather they control how rendering happens.
"""
return (
self.dpi.is_finite
and self.width >= 0
and self.height >= 0
and self.type_ != 'stencil'
)
@property
def dpi(self) -> Resolution:
"""Dots per inch of the image.
Calculated based on where and how the image is drawn in the PDF.
"""
return _get_dpi(self._shorthand, (self._width, self._height))
@property
def printed_area(self) -> float:
"""Physical area of the image in square inches."""
if not self.renderable:
return 0.0
return float((self.width / self.dpi.x) * (self.height / self.dpi.y))
def __repr__(self):
"""Return a string representation of the image."""
return (
f"<ImageInfo '{self.name}' {self.type_} {self.width}×{self.height} "
f"{self.color} {self.comp} {self.bpc} {self.enc} {self.dpi}>"
)
def _find_inline_images(contentsinfo: ContentsInfo) -> Iterator[ImageInfo]:
"""Find inline images in the contentstream."""
for n, inline in enumerate(contentsinfo.inline_images):
yield ImageInfo(
name=f'inline-{n:02d}', shorthand=inline.shorthand, inline=inline.iimage
)
def _image_xobjects(container) -> Iterator[tuple[Object, str]]:
"""Search for all XObject-based images in the container.
Usually the container is a page, but it could also be a Form XObject
that contains images. Filter out the Form XObjects which are dealt with
elsewhere.
Generate a sequence of tuples (image, xobj container), where container,
where xobj is the name of the object and image is the object itself,
since the object does not know its own name.
"""
if Name.Resources not in container:
return
resources = container[Name.Resources]
if Name.XObject not in resources:
return
for key, candidate in resources[Name.XObject].items():
if candidate is None or Name.Subtype not in candidate:
continue
if candidate[Name.Subtype] == Name.Image:
pdfimage = candidate
yield (pdfimage, key)
def _find_regular_images(
container: Object, contentsinfo: ContentsInfo
) -> Iterator[ImageInfo]:
"""Find images stored in the container's /Resources /XObject.
Usually the container is a page, but it could also be a Form XObject
that contains images.
Generates images with their DPI at time of drawing.
"""
for pdfimage, xobj in _image_xobjects(container):
if xobj not in contentsinfo.name_index:
continue
for draw in contentsinfo.name_index[xobj]:
if draw.stack_depth == 0 and _is_unit_square(draw.shorthand):
# At least one PDF in the wild (and test suite) draws an image
# when the graphics stack depth is 0, meaning that the image
# gets drawn into a square of 1x1 PDF units (or 1/72",
# or 0.35 mm). The equivalent DPI will be >100,000. Exclude
# these from our DPI calculation for the page.
continue
yield ImageInfo(name=draw.name, pdfimage=pdfimage, shorthand=draw.shorthand)
def _find_form_xobject_images(pdf: Pdf, container: Object, contentsinfo: ContentsInfo):
"""Find any images that are in Form XObjects in the container.
The container may be a page, or a parent Form XObject.
"""
if Name.Resources not in container:
return
resources = container[Name.Resources]
if Name.XObject not in resources:
return
xobjs = resources[Name.XObject].as_dict()
for xobj in xobjs:
candidate = xobjs[xobj]
if candidate is None or candidate.get(Name.Subtype) != Name.Form:
continue
form_xobject = candidate
for settings in contentsinfo.xobject_settings:
if settings.name != xobj:
continue
# Find images once for each time this Form XObject is drawn.
# This could be optimized to cache the multiple drawing events
# but in practice both Form XObjects and multiple drawing of the
# same object are both very rare.
ctm_shorthand = settings.shorthand
yield from _process_content_streams(
pdf=pdf, container=form_xobject, shorthand=ctm_shorthand
)
def _process_content_streams(
*, pdf: Pdf, container: Object, shorthand=None
) -> Iterator[VectorMarker | TextMarker | ImageInfo]:
"""Find all individual instances of images drawn in the container.
Usually the container is a page, but it may also be a Form XObject.
On a typical page images are stored inline or as regular images
in an XObject.
Form XObjects may include inline images, XObject images,
and recursively, other Form XObjects; and also vector graphic objects.
Every instance of an image being drawn somewhere is flattened and
treated as a unique image, since if the same image is drawn multiple times
on one page it may be drawn at differing resolutions, and our objective
is to find the resolution at which the page can be rastered without
downsampling.
"""
if container.get(Name.Type) == Name.Page and Name.Contents in container:
initial_shorthand = shorthand or UNIT_SQUARE
elif (
container.get(Name.Type) == Name.XObject
and container[Name.Subtype] == Name.Form
):
# Set the CTM to the state it was when the "Do" operator was
# encountered that is drawing this instance of the Form XObject
ctm = Matrix(shorthand) if shorthand else Matrix()
# A Form XObject may provide its own matrix to map form space into
# user space. Get this if one exists
form_shorthand = container.get(Name.Matrix, Matrix())
form_matrix = Matrix(form_shorthand)
# Concatenate form matrix with CTM to ensure CTM is correct for
# drawing this instance of the XObject
ctm = form_matrix @ ctm
initial_shorthand = ctm.shorthand
else:
return
contentsinfo = _interpret_contents(container, initial_shorthand)
if contentsinfo.found_vector:
yield VectorMarker()
if contentsinfo.found_text:
yield TextMarker()
yield from _find_inline_images(contentsinfo)
yield from _find_regular_images(container, contentsinfo)
yield from _find_form_xobject_images(pdf, container, contentsinfo)
def _page_has_text(text_blocks: Iterable[FloatRect], page_width, page_height) -> bool:
"""Smarter text detection that ignores text in margins."""
pw, ph = float(page_width), float(page_height) # pylint: disable=invalid-name
@@ -722,120 +80,6 @@ def simplify_textboxes(
yield TextboxInfo(box.bbox, visible, corrupt)
worker_pdf = None # pylint: disable=invalid-name
def _pdf_pageinfo_sync_init(pdf: Pdf, infile: Path, pdfminer_loglevel):
global worker_pdf # pylint: disable=global-statement,invalid-name
pikepdf_enable_mmap()
logging.getLogger('pdfminer').setLevel(pdfminer_loglevel)
# If the pdf is not opened, open a copy for our worker process to use
if pdf is None:
worker_pdf = Pdf.open(infile)
def on_process_close():
worker_pdf.close()
# Close when this process exits
atexit.register(on_process_close)
@contextmanager
def _pdf_pageinfo_sync_pdf(thread_pdf: Pdf | None, infile: Path):
if thread_pdf is not None:
yield thread_pdf
elif worker_pdf is not None:
yield worker_pdf
else:
with Pdf.open(infile) as pdf:
yield pdf
def _pdf_pageinfo_sync(
pageno: int,
thread_pdf: Pdf | None,
infile: Path,
check_pages: Container[int],
detailed_analysis: bool,
miner_state: PdfMinerState | None,
) -> PageInfo:
with _pdf_pageinfo_sync_pdf(thread_pdf, infile) as pdf:
return PageInfo(
pdf, pageno, infile, check_pages, detailed_analysis, miner_state
)
def _pdf_pageinfo_concurrent(
pdf,
executor: Executor,
max_workers: int,
use_threads: bool,
infile,
progbar,
check_pages,
detailed_analysis: bool = False,
miner_state: PdfMinerState | None = None,
) -> Sequence[PageInfo | None]:
pages: list[PageInfo | None] = [None] * len(pdf.pages)
def update_pageinfo(page: PageInfo, pbar: ProgressBar):
if not page:
raise InputFileError("Could read a page in the PDF")
pages[page.pageno] = page
pbar.update()
if max_workers is None:
max_workers = available_cpu_count()
total = len(pdf.pages)
n_workers = min(1 + len(pages) // 4, max_workers)
if n_workers == 1:
# If we decided on only one worker, there is no point in using
# a separate process.
use_threads = True
if use_threads and n_workers > 1:
# If we are using threads, there is no point in using more than one
# worker thread - they will just fight over the GIL.
n_workers = 1
# If we use a thread, we can pass the already-open Pdf for them to use
# If we use processes, we pass a None which tells the init function to open its
# own
initial_pdf = pdf if use_threads else None
contexts = (
(n, initial_pdf, infile, check_pages, detailed_analysis, miner_state)
for n in range(total)
)
assert n_workers == 1 if use_threads else n_workers >= 1, "Not multithreadable"
logger.debug(
f"Gathering info with {n_workers} "
+ ('thread' if use_threads else 'process')
+ " workers"
)
executor(
use_threads=use_threads,
max_workers=n_workers,
progress_kwargs=dict(
total=total, desc="Scanning contents", unit='page', disable=not progbar
),
worker_initializer=partial(
_pdf_pageinfo_sync_init,
initial_pdf,
infile,
logging.getLogger('pdfminer').level,
),
task=_pdf_pageinfo_sync,
task_arguments=contexts,
task_finished=update_pageinfo,
)
return pages
class PageResolutionProfile(NamedTuple):
"""Information about the resolutions of a page."""
@@ -1208,7 +452,7 @@ class PdfInfo:
)
@property
def pages(self) -> Sequence[PageInfo | None]:
def pages(self) -> list[PageInfo | None]:
"""Return list of PageInfo objects, one per page in the PDF."""
return self._pages