Dao PDF или Автоматизация печати в конструкторском бюро (часть 2)

Процесс печати PDF-файла редко сводится к открытию файла и нажатию кнопки «Печать». Особенно если речь идёт о конструкторском бюро, выпускающем сотни комплектов разнокалиберных чертежей. Автоматизация этого процесса решает множество сопутствующих и часто неочевидных задач. И вполне может оказаться, что в действительности всё не так, как на самом деле.

Содержание:

• Начало

• Зазеркальный дом или Постановка задачи

• Сад, где цветы говорили или Требования и ограничения

• Траляля и Труляля или MVP

• Шалтай-Болтай или Goodbye MVP

• Лев и Единорог или Разделяй и Властвуй

• Королева Алиса или Deploy

• Чей же это был сон или Жизнь в production

• Вместо заключения

Начало

Меня зовут Макс и периодически я занимаюсь автоматизацией процессов для конструкторского бюро как сайд-проектом. В предыдущей части я рассказывал о кейсе автоматизации печати чертежей из AutoCAD. Мы прошли по этапам проектирования и разработки, разобрали нетривиальные подводные камни, с которыми я столкнулся в процессе реализации. В этой статье я хочу рассказать об эволюции системы автоматизированной печати PDF от простого SPA на Django до комплекса, состоящего из WEB-приложения, брокера сообщений и отдельного сервиса печати.

Эта статья - не руководство по использованию отдельных библиотек хотя гонка на них тоже будет. Решения, описанные в статье не идеальны, что-то я бы сейчас возможно сделал иначе. В общем, перед вами история костылей и велосипедов, которые появлялись по мере развития проекта. Мы пройдем путь от постановки задачи до эксплуатации, оптимизации и переработки архитектуры. Местами даже увидим код (но это не точно)

Зазеркальный дом или Постановка задачи

Проект появился как инструмент автоматизации процесса печати инженерной документации в КБ. Инженерные документы довольно сильно отличаются от обычных офисных:

• один файл может содержать от десятка до более чем тысячи страниц

• внутри одного файла одновременно встречаются листы различных форматов (A4, A3, A2, A1 и составные)

• часть листов должна печататься в цвете, часть - в монохромном режиме

Итак. Мы будем печатать многостраничные PDF файлы. Файлы всегда находятся на выделенном файловом сервере. Каждый файл содержит книгу из неопределенного числа страниц, каждая из которых имеет случайный (но из стандартного списка) размер. Каждая страница может быть как цветной, так и черно-белой. КБ использует несколько плоттеров и принтеров для печати определённых форматов

Процесс AS IS описывался следующим алгоритмом:

1. Для каждого файла PDF:

2. Для каждого листа:

3. Определить принтер и параметры печати и добавить в группу

4. Вернуться к 2

5. Для каждой группы страниц с одинаковыми параметрами:

6. Задать параметры печати и количество экземпляров

7. Отправить группу на печать

8. Вернуться к 5

9. Вернуться к 1

Система автоматизированной печати должна позволять выбирать и пакетно обрабатывать файлы PDF, выводить на печать все страницы каждого файла в автоматическом режиме. Система должна самостоятельно определить характеристики каждой страницы документа и принять решение, на каком устройстве её необходимо печатать. Фактически требовалось построить pipeline обработки документа. На вход система получает один PDF-файл. На выходе отдает полностью распечатанный комплект, автоматически распределённый между несколькими принтерами.

Основной задачей проекта стала не сама печать PDF-документов, а автоматическое принятие решения о том, как именно должен быть напечатан каждый лист документа. Это определило pipeline и архитектуру.

Сад, где цветы говорили или Требования и ограничения

Любая архитектура является отражением требований и ограничений предметной области. В моем случае они оказались достаточно кхм… специфичными:

Работа внутри локальной сети

Система изначально проектировалась как внутренний корпоративный сервис. Это облегчало задачу: отсутствовали требования к работе через Интернет, все пользователи находились внутри закрытой локальной сети предприятия, все документы уже размещались на корпоративном файловом сервере. Вместо загрузки больших PDF-файлов через веб-интерфейс значительно эффективнее передавать системе только путь к документу. Такой подход позволял избежать копирования файлов между сервисами.

Максимально простой пользовательский интерфейс

Одним из главных требований Будем честными: единственным требованием от бизнеса стала минимизация количества действий пользователя. Пользователь не должен разбираться в особенностях работы плоттеров, форматах бумаги или настройках драйверов, идеальный процесс   TO BE должен выглядеть так:

1. Выбрать файлы

2. Нажать кнопку

3. Забрать распечатанные комплекты документов

На практике это означает что пользователь не должен думать о том как выбрать:

• формат бумаги

• принтер

• режим печати (цветной/монохромный)

• масштаб

• параметры драйвера

Все решения должна принимать система. Подобный подход значительно снижает вероятность ошибок и практически исключает влияние человеческого фактора.

Автоматический выбор устройств

Нужна возможность работы одновременно с несколькими принтерами:

• плоттер A1

• лазерный принтер A3

• несколько офисных принтеров A4

В идеальном мире можно просто написать скрипт и отправлять каждый документ на большой плоттер и потом открыть в КБ филиал кружка кройки и шитья на несколько дней. Идеальный план, надежный как швейцарские часы… Бизнес не поймет.

Система должна самостоятельно определять характеристики каждой страницы и принять решение, куда именно следует отправить конкретный лист. Вокруг алгоритма выбора мы и будем строить процесс обработки документов.

Хранение истории

Так исторически сложилось, что система никогда не рассматривалась как highload сервис. А вот отслеживание задач и их результатов (и полных и постраничных) было крайне необходимо. Хранение истории и отслеживанию состояния каждого задания появилось сразу же. Это потом ограничения приведут к появлению отдельного сервиса обработки и использованию брокера сообщений. Но вначале было слово Django-приложение с историей заданий в базе, которое должно было показать жизнеспособность автоматизации.

Windows only

Вся инфраструктура предприятия была построена на Windows. Плоттеры, лазерные принтеры, существующие средства администрирования ориентировались исключительно на эту платформу. Даже не знаю, кто скажет почему так… AUTODESK?

Траляля и Труляля или MVP

Поиск в интернете дал достаточно широкий разброс решений: от печати файлов напрямую на дефолтном системном принтере (может это и полезно кому-то но явно не наш кейс) до ответа о принципиальной возможности решения задачи на Python с использованием WinAPI. Сразу скажу что выбрал этот вариант как наиболее гибкий. Google даже выдал документацию для PyWin32, любезно предоставленную Tim Golden. Приступим к реализации , или мы хотим жить вечно?

Первая версия создавалась максимально простой и была обычной SPA на Django. Авторизация пользователей не предусмотрена (за исключением Администратора), для истории заданий достаточно хранить IP и Hostname клиента с которого отправили задание. Поскольку система изначально предназначалась для корпоративной сети под Windows, выбор веб-сервера тоже оказался достаточно очевидным. Использование IIS позволяло использовать существующую инфраструктуру предприятия. Фактически приложение разворачивалось так же, как это делают взрослые большинство внутренних корпоративных веб-сервисов.

Практически сразу появилось еще одно ограничение. Документы находятся на файловом сервере и вместо загрузки файлов пользователь просто указывает путь к документу. Подробно рассматривать механизм получения пути к файлу в статье я не буду. Остановлюсь лишь на том, что для формирования его backend-у необходимо иметь доступ на чтение к файловой системе. Впоследствии пришлось вынести эту задачу отдельно, так как это потенциальная угроза безопасности.

Интерфейс первой версии ладно, следующих тоже был минималистским. После выбора документа пользователь фактически использовал одну большую красную кнопку Отправить. Вся логика обработки скрывалась за ней. Все решения принимала система. Этот принцип сохранился и впоследствии.

После создания задания происходило сохранение записи в базе данных. Затем Django генерировал сигнал post_save. Обработчик сигнала запускал отдельный процесс, который выполнял всю основную работу. Front-end периодически спрашивал сервер за статус задачи. Easy Для MVP этого оказалось более чем достаточно.

Рассмотрим процесс обработки файла подробнее. Для обработки документа будем использовать библиотеку PyPDF2.

Извлечение

После открытия файла получим его страницы.

from PyPDF2 import PdfFileReader


def get_pages(item):
    result = []
    input_data = PdfFileReader(open(item, 'rb'), strict=False)
    count = input_data.getNumPages()
    for i in range(count):
        page = input_data.getPage(i)
        result.append(page)
    return result

Теперь вместо одного большого файла мы работаем с последовательностью независимых страниц. Это позволяет реализовать их автоматическое распределение между принтерами. Для каждой страницы документа получаем объект RawPdfItem, содержащий необходимую информацию и содержимое страницы в виде байтов.

import io
from PyPDF2 import PdfFileWriter


class RawPdfItem:
	"""RAW Page Data"""
    _delim = 72.0 / 25.4
    path = ""
    page = 0
    paper_size = None
    data = None
    
    def __init__(self, name, page, page_num=0):
        self.name = name
        self.page = page_num
        page_rect = page.mediaBox.upperRight
        self.paper_size = {
            "w": float(page_rect[0]) / self._delim,
            "h": float(page_rect[1]) / self._delim
        }
        wrt = PdfFileWriter()
        wrt.addPage(page)
        bts = io.BytesIO()
        wrt.write(bts)
        self.data = bts.getvalue()
        bts.close()

Растеризация

Далее получим растровое изображения страницы и определим необходимость цветной печати. Для этого воспользуемся ImageMagick в компании с Ghostscript для получения растра и Pillow для окончательного преобразования растра в формат, понятный принтеру.

from wand.image import Image as wImage
from wand.color import Color
from PIL import Image, ImageStat

from main_app.config import COLOR_MIN


def get_image(self, data, fmt="png", res=300, rem=True):
    with wImage(blob=data, resolution=res) as orig:
        with orig.convert(fmt) as img:
            img.background_color = Color("white")
            if rem:
                img.alpha_channel = "remove"
            else:
                img.alpha_channel = "background"
            img_bin = img.make_blob(format=fmt)
            buff = numpy.asarray(bytearray(img_bin), dtype="uint8")
            data = io.BytesIO(buff)
            res_img = Image.open(data)
            if res_img.mode != "RGB":
                res_img = res_img.convert("RGB")
            return create_img(res_img)


def create_img(src):
    bg_color = (255, 255, 255)
    res = Image.new("RGB", src.size, bg_color)
    res.paste(src)
    return res

Здесь хорошо видно двойное преобразование. Таков путь Все просто: принтер ожидает PIL.Image. Да, у нас уже есть страница в виде байтов, но чтобы получить изображение ее нужно отрендерить. Pillow в такое не умеет, зато умеет ImageMagick. Введение метода create_img() с принудительной установкой белого фона потребовалось для устранения эффекта заливки прозрачных областей при печати некоторых файлов. Да, принудительное удаление или замена альфа-канала при первом преобразовании помогало не во всех случаях (Слава Stackoverflow!). В качестве сайд эфекта мы получили возможность контролировать ход процесса для тестирования через сохранение промежуточных PNG файлов.

Печать

Здесь уже все просто (НЕТ). За работу с принтерами отвечает класс Device. За печать отвечает метод print(), принимающий объект содержащий PIL.Image и необходимую информацию. Принтеру мы задаем физические размеры листа на котором печатать, отмасштабированную область печати с требуемым сдвигом и говорим “Напечатай это изображение вот так”. Детали печати растра через WinAPI легко гуглятся поэтому здесь я их приводить не буду.

from dataclasses import dataclass
from PIL import Image


@dataclass
class Layout:
    """Layout data"""
    image: Image
    print_area: list[int]
    draw_rect: tuple[int, int, int, int]
    scaled_size: tuple[int, int]

import win32ui
import win32gui
import win32print

from app_log import get_logger
from exceptions import PrintProcessingError
from print_service.PrintedItem import PrintedItem


logger = get_logger(LOG_FILE, __name__)


class Device:
	"""Printing device"""
    key: str
    name: str
    papers: list[dict]
    access_mode: dict = {"DesiredAccess": win32print.PRINTER_ACCESS_USE}

    def __init__(self, key, name):
        self.key = key
        self.name = name
        self._set_device_papers()


    def print(self, item: PrintedItem) -> dict:
        """Print item"""
        result = {'Success': False, 'msg': 'Error: Printing failed'}

        handle = None
	hdc = None
	dc = None

        try:
            # определяем параметры листа
            paper_name, paper_value = self._get_paper_params(item)

            handle = win32print.OpenPrinter(self.name, self.access_mode)

            # получаем валидированный объект DEVMODE 
            paper_name, devmode_obj = self._prepare_devmode(handle, item, paper_name, paper_value)

            # Используем валидированный объект DEVMODE для создания контекста устройства
            hdc = win32gui.CreateDC('WINSPOOL', self.name, devmode_obj)
            dc = win32ui.CreateDCFromHandle(hdc)

            # получаем объект Layout
            layout = self._get_print_layout(dc, item, paper_name)

            # отправляем Layout на печать
            res, msg = self._send_to_printer(dc, layout)

            if not res:
                raise PrintProcessingError(msg)
            result = {'Success': res, 'msg': msg}
            logger.debug(f"Printing finished")
        except Exception as exc:
            logger.exception(exc)
        finally:
        	# Освобождаем ресурсы
            if dc:
                try:
                    dc.DeleteDC()
                except Exception as exc:
                    logger.debug(f'Unable to release printer DC wrapper: {str(exc)}')
            if hdc:
                try:
                    win32gui.DeleteDC(hdc)
                except Exception as exc:
                    logger.debug(f'Unable to release printer DC handle: {str(exc)}')
            if handle:
                win32print.ClosePrinter(handle)

        return result

Для получения требуемого размера листа мы сопоставим физические размеры страницы с форматами бумаги, поддерживаемыми конкретным принтером. Для стандартных форматов используются значения DMPAPER_A4 и DMPAPER_A3, для остальных размеров выполняется поиск среди форматов, возвращённых драйвером устройства с учетом допуска: abs(paper_size['h']*10 - data['x']) < TOLERANCE*10. Таким образом учитываются небольшие отклонения от номинальных размеров бумаги.

pipeline обработки выглядит следующим образом: PDF -> PyPDF2 -> Разбивка на страницы -> ImageMagick -> PNG -> Pillow -> PIL.Image -> Печать

Итоги

Итак задача решена. Пользователь получил максимально простой интерфейс. WEB-приложение самостоятельно разбивает документы на страницы и печатает их. Подробная история заданий сохраняется в базе данных и доступна по требованию как для просмотра так и для выгрузки. Администратору (после входа в систему) также доступна для выгрузки история заданий всех пользователей. Обработка файлов выполняется в отдельном процессе, не блокируя работу веб-интерфейса. MVP презентован и запущен в тестовом режиме, инструкция для пользователей написана и выложена на файловый сервер, обучение проведено. Пользователь счастлив. Возможно потому, что альтернатива - печатать вручную, но это не точно

Шалтай-Болтай или Goodbye DPI MVP

Быстро сделано, время делать хорошо. А недостатков у MVP… (много). Хотя обработка задания и выполняется в отдельном процессе, её жизненным циклом по-прежнему управляет Django. Любая правка механизма печати требует нового деплоя всего веб-приложения. Напомню что Django-приложение развернуто на IIS, кто знает - тот знает. Да и выбранный путь преобразования в растр оказался слегка неоптимальным.

Инженерные комплекты документов включают десятки (иногда сотни) страниц различных форматов. Некоторые чертежи очень детализированы и при этом очень большие (форматы А2х3 - А2х5 в разделе генплана для КБ - обычное дело). Каждая страница проходит через несколько преобразований с созданием временных файлов в памяти. Если документ содержит 100–150 страниц, решение с ImageMagic начинает с аппетитом кушать ресурсы и время. Так, конвертация в PNG с разрешением 300 DPI одной страницы с насыщенным чертежом формата A2*3 займет около 3 минут если хватит памяти на машине, если нет - лист зафейлится и перепечатывать его придется потом вручную. Представили лицо пользователя, да?

Подсистема печати фактически представляет собой самостоятельное приложение со своим жизненным циклом, собственной логикой обработки документов и собственными требованиями. Идея разнести веб-интерфейс и сервис обработки файлов по разным сервисам просто витает в воздухе.

Подвезли и технических ограничений. Так как новые ПК приобретались только для 3D-моделирования и апгрейд железа предусматривался в первую очередь для инженерных машин, для развертывания системы в КБ могли быть выделены только достаточно старые ПК со скромными ресурсами.

Распил и откат

Так появляется самостоятельный сервис обработки документов, взаимодействующий с WEB-приложением через шину данных. Монолитное WEB-приложение превращаются в независимые сервисы, каждый из которых решает собственную задачу. WEB-сервис занимается регистрацией задания, сохранением информации в базе данных. Ну и публикует сообщения в шину. Вся дальнейшая обработка выполняется в отдельном Воркере. В качестве шины данных выбран RabbitMQ. WEB-приложение остается жить на IIS, RABBITMQ также разворачивается в Windows. Разделение сервисов позволяет и заниматься ими отдельно и использовать для них разные машины впоследствии.

Рассмотрим Воркер более подробно. При получении задания Воркер извлекает из него имена файлов и отправляет каждый файл на печать постранично. Каждую страницу преобразует в изображение определенного формата с достаточным для качественной печати разрешением (помним, что для MVP я остановился на PNG и 300 DPI). После выполнения задания - отправляет WEB-приложению отчет с результатом. Также было бы неплохо возложить на него задачу проверки доступности принтеров по запросу. Для чего запустим легкий асинхронный сервер на AIOHTTP в отдельном процессе.

Схематично Воркер выглядит так:

        ┌─────────────────┐    ┌────────────┐    ┌───────────┐
        │    PrintSrv     │    │  ReportSrv │    │   RmqSrv  │ <-> RabbitMQ
        ├────────┬────────┤    └──────┬─────┘    └───────────┘
        │        │        │           │ 
    PdfSrv DeviceSrv -> Device   WEB-server
                          │
          			    WinAPI

Разделение позволило локализовать ответственность каждого компонента:

• PrintSrv организует весь процесс печати

• PdfSrv отвечает за обработку PDF

• DeviceSrv управляет доступными устройствами

• Device управляет печатью

• ReportSrv отправляет отчеты по запросу

• RmqSrv отвечает за работу с шиной данных

Требовательный к ресурсам ImageMagick заменен на Poppler с Python-оберткой pdf2image.

from os.path import isfile

import win32con
from pdf2image import convert_from_bytes
from PIL import Image, ImageStat

from device_service.DeviceSrv import DeviceSrv
from settings import PRINTERS, COLOR_MIN, TOLERANCE, THREAD_COUNT, POPPLER_PATH


Image.MAX_IMAGE_PIXELS = None  # for large images


def get_image(data, fmt='png', res=300, rem=True):
    USE_CROPBOX = False
    STRICT = False

    pil_image = convert_from_bytes(data,
                                   dpi=res,
                                   fmt=fmt,
                                   thread_count=THREAD_COUNT,
                                   use_cropbox=USE_CROPBOX,
                                   strict=STRICT,
                                   poppler_path=POPPLER_PATH)[0]

    if rem:
        if pil_image.mode in ('RGBA', 'LA') or \
                (pil_image.mode == 'P' and 'transparency' in pil_image.info):
            alpha = pil_image.convert('RGBA').split()[-1]
            bg = Image.new("RGBA", pil_image.size, (255, 255, 255, 255))
            bg.paste(pil_image, mask=alpha)
            pil_image = bg

    if pil_image.mode != 'RGB':
        pil_image = pil_image.convert('RGB')

    return create_img(pil_image)

Переход на pdf2image упрощает pipeline, добавляет стабильности и снижает потребление ресурсов. В таком виде система живет достаточно длительное (более года) время, пока не возникает необходимость переезжать на другой ПК. Для машин которые доступны к использованию аппетиты воркера черезмерны. Плюс развертывание также требует установки дополнительных пакетов.

Выборы, Выборы… (с)

Следующий кандидат MuPDF. У библиотеки довольно сложная система лицензирования, и нужно быть абсолютно уверенным что переход на нее будет оправдан. Чтобы сравнить производительность и потребление ресурсов напишем тестовый скрипт с конвертацией в растр одной и той же книги чертежей и сохранением получившихся PNG на диск. Будем замерять потребление памяти для каждой страницы и найдем максимальное. Для замеров воспользуемся tracemalloc, встроенный в стандартную библиотеку.

Делаем тестовый прогон для pdf2image и получаем

WHAT? А что в Task Manager?

Да. Все верно. Так как pdf2image запускает дочерний процесс tracemalloc просто его не видит. Чтош. Оставим его для общего развития но для реальных замеров будем использовать psutil. Измерять будем потребление памяти и процессорное время из которого получим среднюю нагрузку на процессор. Естественно замерять будем для всего дерева процессов рекурсивно.

import os
import time
import threading
from abc import ABC, abstractmethod
from collections import namedtuple
from dataclasses import dataclass

import psutil


@dataclass
class CPUStats:
    user: float = 0.0
    system: float = 0.0


class BaseMonitor(ABC):
    interval: float
    thread: threading.Thread
    stop_flag: threading.Event
    pid: int
    max_val: float
    samples: list
    results: dict
    started: bool = False

    def __init__(self, interval: float = 0.05):
        self.interval = interval
        self.stop_flag = threading.Event()
        self.pid = os.getpid()
        self.max_val = 0.0
        self.samples = []
        self.results = dict()

    def start(self) -> None:
        self.started = True
        self.thread = threading.Thread(target=self._monitor, daemon=True)
        self.thread.start()

    def stop(self) -> None:
    	if not self.started:
            raise RuntimeError("Monitoring not started")
        self.started = False
        self.stop_flag.set()
        self.thread.join()

    @abstractmethod
    def _monitor(self) -> None:
        pass


class MemoryMonitor(BaseMonitor):

    def _monitor(self) -> None:
        main_process = psutil.Process(self.pid)

        while not self.stop_flag.is_set():
            res = main_process.memory_info().rss
            for child in main_process.children(recursive=True):
                try:
                    res += child.memory_info().rss
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    pass

            self.samples.append(res)
            self.max_val = max(self.max_val, res)

            time.sleep(self.interval)

    def stop(self) -> None:
    	super().stop()
    	if len_smpl := len(self.samples):
	        self.results = {
	            "avg": sum(self.samples) / len_smpl,
	            "max": self.max_val
	        }

class CPUTimeMonitor(BaseMonitor):
	_data: dict[int, CPUStats] = dict()
	_start_time: float = 0.0

    def __init__(self, interval: float = 0.05):
    	super().__init__(interval=interval)
        self._process = psutil.Process(self.pid)
        self._process_cpu_before: namedtuple()

    def start(self):
    	self._data = {}
    	self.results = {}
        self._process_cpu_before = self._process.cpu_times()
        self._start_time = time.time()
        super().start()

    def stop(self) -> None:
        super().stop()

        stop_time = time.time()
        process_cpu_after = self._process.cpu_times()
        process_cpu_user = process_cpu_after.user - self._process_cpu_before.user
        process_cpu_sys = process_cpu_after.system - self._process_cpu_before.system

        children_cpu = CPUStats()
        for pid, stats in self._data.items():
            children_cpu.user += stats.user
            children_cpu.system += stats.system
        total_user = process_cpu_user + children_cpu.user
        total_system = process_cpu_sys + children_cpu.system

        self.results = {
        	"avg": 100 * (total_user + total_system) / (stop_time - self._start_time),
            "user": process_cpu_user + children_cpu.user,
            "sys": process_cpu_sys + children_cpu.system
        }

    def _monitor(self):
        while not self.stop_flag.is_set():
            self._collect_processes()
            time.sleep(self.interval)

    def _collect_processes(self):
        try:
            processes = [
                self._process,
                *self._process.children(recursive=True),
            ]
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            return

        for proc in processes:
            try:
                if proc.pid == self.pid:
                    continue
                cpu = proc.cpu_times()
                self._data[proc.pid] = CPUStats(user=cpu.user, system=cpu.system)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass

from time import time
import tracemalloc

from monitor import MemoryMonitor, CPUTimeMonitor
from pdf_service.PdfSrv import PdfSrv
from print_service.PrintedItem import PrintedItem


def main():
    src = r"Здесь захардкожен путь к PDF файлу З)"
    result = []

    mem_monitor = MemoryMonitor()
    cpu_time_monitor = CPUTimeMonitor()
    mem_monitor.start()
    cpu_time_monitor.start()
    tracemalloc.start()

    start_pdf = time()
    items = PdfSrv.get_raw_page_items(src)
    end_pdf = time()
    for i, j in enumerate(items):
        tracemalloc.reset_peak()
        start = time()
        item = PrintedItem(j)
        result.append((i, time() - start, tracemalloc.get_traced_memory()))
        item.save(f'tmp\\result-{i}.png')
        del item
        del j
    end = time()

    tracemalloc.stop()
    mem_monitor.stop()
    cpu_time_monitor.stop()

    print("Conversion details:\n")
    for j in result:
        print(f"Time [Page {j[0]}]: {j[1]}")
        print(f"Memory usage [Page {j[0]}]: {j[2][0] / (1024 * 1024):.2f} MB")
        print(f"Peak memory usage [Page {j[0]}]: {j[2][1] / (1024 * 1024):.2f} MB")

    target = max(result, key=lambda x: x[1])

    print(f"\nPDF info:\nName: {src}\nPages: {len(result)}")
    print("\nConversion time: ", end - end_pdf)

    # для общего развития выведем и результаты tracemalloc для самой тяжелой страницы
    print(f"Max memory usage [Page {target[0]}]: {target[2][0] / (1024 * 1024):.2f} MB")
    print(f"Max Peak memory usage [Page {target[0]}]: {target[2][1] / (1024 * 1024):.2f} MB")
    print(f"Max conversion time [Page {target[0]}]: {target[1]}")

    print("\nTotal Time: ", end - start_pdf)
    print(f"Max total memory usage: {mem_monitor.results['max'] / (1024 * 1024):.2f} MB")
    print(f"Average total memory usage: {mem_monitor.results['avg'] / (1024 * 1024):.2f} MB")
    print(f"Average total CPU usage: {cpu_time_monitor.results['avg']} %")
    print(f"CPU time User: ", cpu_time_monitor.results['user'])
    print(f"CPU time System: ", cpu_time_monitor.results['sys'])

Для теста возьмем один файл о 43 страницах. Конвертацию будем выполнять в PNG с DPI 300. Параметры железа: CPU: Intel® Xeon® E3-1270 V2 @ 3.50GHzRAM: 16Гб. На проде железо будет немного слабее. Таков Путь. Результаты замеров выглядят неоднозначно:

Результаты в табличной форме

Утилизация процессора более 100% говорит в том числе и о использовании более чем одного ядра. При этом утилизация CPU и общее потребление памяти как среднее так и пиковое остается сравнимым.

Для самой тяжелой страницы (она своя для каждой из библиотек но всегда это один из насыщенных чертежей генплана и номер страницы постоянен для любого замера с одной и той же библиотекой) результат тоже вполне показателен

• по данным tracemalloc без учета внешнего процесса

Если принять во внимание потребление памяти внешним процессом для pdf2image то Внезапно оказывается что MuPDF выигрывает по скорости, суммарному CPU времени и памяти (мы же помним скриншот из task manager?) на самой тяжёлой странице и проигрывает только по общему потреблению памяти.

Итак. При вдвое больших средней загрузке CPU и средним потреблением памяти, он требует в 3.3 раза меньше процессорного времени на обработку одного и того же файла и выполняет задачу в 6 (и в 60 для самой тяжелой страницы) раз быстрее.

Использование библиотеки сократит стоимость и продолжительность конвертации, что крайне актуально для относительно слабого железа. Также преимуществом библиотеки будет то, что она заменяет и PyPDF2 и pdf2image и итоге pipeline обработки выглядит так: PDF -> MuPDF -> Страница -> Pixmap -> Pillow -> PIL.Image -> Печать

Код представления документа и страницы стал значительно проще:

import fitz


class PdfItem:
	"""PDF Data"""
    _data: fitz.Document

    def __init__(self, path: str):
        self._data = fitz.open(path)

    def get_base_name(self):
        start = self._data.name.rfind('\\') + 1
        end = self._data.name.rfind('.')
        return self._data.name[start: end]

    def get_base_dir(self):
        end = self._data.name.rfind('\\')
        return self._data.name[: end]

    def get_pages(self):
        return self._data.pages()

    def close(self):
        self._data.close()


class RawPdfItem:
	"""RAW Page Data"""
    # The data has to be stated in the unit point.
    # One point is defined as 1/72 of an inch (25.4 mm)
    _delim = 72.0
    _PT_TO_MM = 25.4 / 72.0
    name: str
    page: int
    paper_size: dict
    image_size: tuple
    data: bytes
    fmt: str
    dpi: int
    mode: str

    def __init__(self, name: str, page_num=0, page=fitz.Page, fmt: str = "png", dpi: int = 300):
        self.name = name
        self.page = page_num
        self.fmt = fmt
        self.dpi = dpi
        self.paper_size = {
            'w': page.mediabox.width * self._PT_TO_MM,
            'h': page.mediabox.height * self._PT_TO_MM
        }
        zoom = dpi / self._delim
        pix = page.get_pixmap(matrix=fitz.Matrix(zoom, zoom))
        self.image_size = (pix.width, pix.height)
        self.mode = "RGBA" if pix.alpha else "RGB"
        self.data = pix.samples

Работа с файлом выглядит таким образом:

    def get_raw_page_items(path):
        pdf_obj = PdfItem(path)
        for i, page in enumerate(pdf_obj.get_pages()):
            yield RawPdfItem(path, i, page=page)
        pdf_obj.close()

Здесь тоже можно заметить оптимизацию - ленивую передачу страниц в pipeline. Это позволяет начать обработку документа до полного формирования набора страниц и уменьшает потребление памяти.

Лев и Единорог или Разделяй и Властвуй

Упрощённо workflow выглядит так:

Получение задания -> Открытие PDF -> Извлечение страницы -> Получение размеров -> Получение растрового изображения -> Определение цветности -> Выбор принтера -> Печать страницы -> Следующая страница

В каждый момент времени мы работаем с одной страницей. Это позволяет автоматически распределять различные листы одного документа между несколькими устройствами.

Извлечение страницы

Открытие документа занимает буквально несколько строк. В отличие от предыдущих версий здесь нет промежуточных результаты и запуска внешних процессов. Каждая страница сразу представляется объектом fitz.Page. Одновременно с созданием объекта страницы получаем её физические размеры. Это реализовано внутри класса RawPdfItem.

self.paper_size = {
    'w': page.mediabox.width * self._PT_TO_MM,
    'h': page.mediabox.height * self._PT_TO_MM
}

MuPDF хранит размеры страницы в типографских пунктах, переводим их в миллиметры и уже работаем с привычными размерами бумаги.

Получение растрового представления

Далее извлекаем изображение страницы. Используется метод get_pixmap(). Схраняем непосредственно массив пикселей.

zoom = dpi / 72

pix = page.get_pixmap(
    matrix=fitz.Matrix(zoom, zoom)
)

self.data = pix.samples

self.image_size = (
    pix.width,
    pix.height
)

Анализ цветности страницы

Анализ страницы можно выделить два этапа: извлечение признаков и принятие решения. На первом этапе из PDF и растрового представления извлекаются геометрические и цветовые характеристики страницы. На втором эти характеристики используются как входные данные для выбора устройства

Определим класс PrintedItem. Он отвечает за анализ страницы, хранит ее параметры и представление в PIL.Image.

from PIL import Image, ImageStat
from pdf_service.RawPdfItem import RawPdfItem
from settings import COLOR_MIN, PRINTERS, TOLERANCE, LOG_FILE

Image.MAX_IMAGE_PIXELS = None


class PrintedItem:
    name: str
    page: int
    paper_size: dict
    _img: Image
    _color: bool = False
    res: int

    def __init__(self, item: RawPdfItem):
        self.name = item.name
        self.res = item.dpi
        self.page = item.page
        self.paper_size = {
            'w': self.round_value(item.paper_size['w']),
            'h': self.round_value(item.paper_size['h'])
        }
        self.set_image(item.data, item.mode, item.image_size)
        self.set_color_mode()

Магия определения цветности в методе set_color_mode() достаточно проста: вычислим дисперсию значений каждого цветового канала. Если различие между дисперсиями превышает установленный порог, страница считается цветной.

def set_color_mode(self):
    dispersion = ImageStat.Stat(self._img).var
    delta = abs(max(dispersion) - min(dispersion))
    self._color = True if delta > COLOR_MIN else False

Простая эвристика, достаточная для инженерной документации. Большинство чертежей представляет собой чёрно-белую Нет графику. При наличии цветных элементов различие между распределениями RGB-каналов становится достаточно заметным. Порог COLOR_MIN подобран эмпирически на наборе реальных документов.

Автоматическое определение принтера

После определения размеров листа и цветности выберем подходящее устройство печати. Это реализовано как вычисляемое свойство device_name() нашего класса. Для назначения принтера нам нужно знать размер страницы и ее цветность. Сравним размеры страницы с эталонными с учетом допуска: abs(width - 297) < TOLERANCE. Таким образом учтем отклонения размеров PDF от стандартов ISO. В упрощённом виде набор правил для назначения принтеров выглядят следующим образом:

| Формат | Цвет | Устройство       |
|--------|------|------------------|
| A4     | Нет  | printer_a3       |
| A4     | Да   | printer_a4_color |
| A3     | Нет  | printer_a3       |
| A3     | Да   | printer_a3_color |
| A2+    | Нет  | printer_a2       |
| A2+    | Да   | printer_a2_color |

Правила проще всего хранить в конфиге по-факту, когда через несколько лет заменили сломавшийся принтер, в конфиге изменилась одна строчка. Если пользователю нужно знать какие принтеры в работе - эту информацию ему предоставит по запросу WEB-приложение.

Поиск доступных устройств

Воркер не использует заранее заданные имена устройств Windows. При запуске происходит опрос устройств печати. Для этого используется WinAPI.

printers = win32print.EnumPrinters(
    win32print.PRINTER_ENUM_NAME,
    server,
    LEVEL
)

После получения списка создаем объекты устройств. В конструктор передаем логическое и реальное имена принтера из свойств pShareName и pPrinterName.

device = Device(
    human_name,
    name
)

Печать

Печатать будем в цикле в модуле PrintSrv. Метод get_raw_page_items() мы рассматрели выше.

for raw_item in PdfSrv.get_raw_page_items(file):

    item = PrintedItem(raw_item)

    device = DeviceSrv.get_device(
        item.device_name
    )

    device.print_item(item)

Соберем результаты всех предыдущих действий:

1. PdfSrv отдает страницу

2. PrintedItem определяет ее свойства и содержание готовое к печати

3. DeviceSrv проверяет доступность нужного принтера и отдает его

4. Device выполняет печать

5. PROFIT!

После отправки страницы на печать не будет лишним подождать пока освободится очередь устройства.

while len(device.get_device_tasklist()) > 0:
    time.sleep(TIME_OUT)

Ожидание исключит переполнение очереди печати и позволит стабильно обрабатывать большие комплекты документации к слову, печать на лазерных принтерах A3 выполняется настолько быстро что на больших файлах они перегреваются.

Почему все ТАК СЛОЖНО?

Инженерные PDF неоднородны. Один комплект может содержать одновременно множество страниц различных размеров и цветности. Если анализировать документ целиком, распределение по устройствам становится достаточно ресурсоемким (держать в памяти весь документ в готовом к печати виде - то еще удовольствие). Смещение фокуса внимания на отдельную страницу упростило реализацию и позволило мне реализовать «большую красную кнопку» из требований бизнеса.

Пара слов об отчетах.

Отчет о выполнении задания имеет следующую структуру:

{
    "id": 100500,
    "ok": false,
    "msg": "ERROR_MESSAGE",
    "detail": [
        {
            "name": "FILE_NAME",
            "ok": false,
            "pages": [
                {
                    "ok": true,
                    "msg": ""
                },
                {
                    "ok": false,
                    "msg": "ERROR_MESSAGE"
                }
            ]
        }
    ]
}

и в таком виде отправляется WEB-приложению, которое уже обновляет результат в базе и отчитывается пользователю о выполнении. Задание считаем корректно выполненным, если все страницы каждого файла напечатались без ошибок.

А что если один из принтеров недоступен?

В этом случае воркер не станет выполнять задание и сразу пометит его как Failed c соответствующей записью в поле “msg”. Без администратора тут никак.

Королева Алиса или Deploy

Так как я уже давно не работал в КБ, одним из новых требований стало максимально простое развертывание и администрирование в существующей инфраструктуре предприятия. При этом разные компоненты предъявляют совершенно различные требования к окружению. Деплой Django-приложения на IIS не выглядит простым да и RabbitMQ деплоить под Windows тот еще квест. Подсистема печати же настолько тесно связана с Windows, драйверами принтеров и WinAPI что просто не работает в виртуальном окружении. В итоге я остановился на гибридной схеме:

             Docker Host
    ┌────────────────────────────┐
    │ WEB Interface │  RabbitMQ  │
    └────────────────────────────┘
                  │
            Local Network
                  │
         Windows Print Server
         ┌──────────────────┐
         │     WORKER       │
         └──────────────────┘

Чей же это был сон или Есть ли Жизнь в production

Так исторически сложилось, что на проекте я совмещал роли архитектора и единственного разработчика, отвечая за весь жизненный цикл системы: от анализа задачи и проектирования архитектуры до реализации, внедрения и последующей поддержки production-системы. Принимал ключевые технические решения: выбирал технологии, определял границы компонентов, анализировал узкие места. Которые к слову иногда были неожиданными. Так, в процессе эксплуатации мне пришлось снижать скорость обработки документов вводя принудительный таймаут после определенного количества страниц для лазерных принтеров. Просто не вывозили печать документа в 1к+ страниц и перегревались ¯\_(ツ)_/¯

Так как я не был в штате, на новый уровень вышли требования к документированию. Для пользователей была разработана инструкция со сценариями работы от выбора файлов и распечатки комплектов до получения выгрузки истории заданий. Для системных администраторов - инструкция по устранению неполадок, перезапуску и развертыванию всех компонентов. Для дальнейшей поддержки - описание архитектуры и основных модулей. Весь пакет документов был размещен на файловом сервере КБ.

В итоге разработанная мной система не претендуя на идеальные архитектуру и код позволила сократить время выпуска чертежей более чем на 75%. Пользовательский сценарий свёлся к указанию PDF и нажатию одной кнопки. Одним из ключевых решений стал алгоритм автоматического выбора принтера и настроек печати для каждой страницы документа. Он позволил исключить ручной выбор устройств, задание параметров печати и на порядок снизить количество ошибок.

Вместо заключения

Хороший дизайн системы это не про идеальную архитектуру. Это про подходящие именно к твоей ситуации компромиссы и решения. Про путь, который ты должен пройти.

Когда проект только начинался, задача была простой: ускорить печать PDF-документов. А получилась система, которая самостоятельно анализирует каждую страницу, определяет её характеристики, выбирает подходящее устройство, распределяет задания между принтерами и хранит историю выполнения.

Первая версия была ориентирована на быстрый результат не считаясь с затратами. По мере развития возникла необходимость пересмотреть и архитектуру и технические решения. Переход на MuPDF позволил сократить потребление ресурсов и упростить pipeline. Выделение подсистемы печати устранило связанность между интерфейсом и механизмом печати, дало возможность развивать обе части независимо. Использование RabbitMQ обеспечило асинхронную передачу и буферизацию заданий.

Что интересно… С технической точки зрения задача оказалась не столько задачей печати, сколько задачей построения вычислительного pipeline с неоднородными входными данными.