s3printlog/s3printlog/report.py

import io
import PIL

from collections import namedtuple
from itertools import zip_longest
from reportlab.pdfgen.canvas import Canvas
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.lib.units import inch, cm
from reportlab.platypus import (
    Flowable,
    Image,
    KeepTogether,
    PageBreak,
    Paragraph,
    Spacer,
    SimpleDocTemplate,
    Table,
)

ImageBuffer = namedtuple("ImageBuffer", ["buffer", "width", "height"])
FailedDropImage = namedtuple("FailedDropImage", ["path", "well"])

styles = getSampleStyleSheet()
style_n = styles["Normal"]
style_h1 = styles["Heading1"]
style_h2 = styles["Heading2"]

TABLE_STYLE = [
    ("TOPPADDING", (0, 0), (-1, -1), 0),
    ("RIGHTPADDING", (0, 0), (-1, -1), 7),
    ("BOTTOMPADDING", (0, 0), (-1, -1), 0),
    ("LEFTPADDING", (0, 0), (-1, -1), 0),
    ("FONTSIZE", (0, 0), (-1, -1), 8),
]


class DropPictures(Flowable):
    """A row of drop pictures flowable."""

    def __init__(self, pictures, xoffset=0):
        self.pictures = filter(None, pictures)
        self.xoffset = xoffset
        self.size = 3.75 * cm
        self.offsets = [0 * cm, 6 * cm, 12 * cm]

    def wrap(self, *args):
        return (self.xoffset, self.size + 1 * cm)

    def draw(self):
        canvas = self.canv
        for offset, picture in zip(self.offsets, self.pictures):
            canvas.drawImage(picture.path, offset, 0, width=5 * cm, height=self.size)
            canvas.drawString(offset + 0.5 * cm, 3.0 * cm, picture.well)



def print_info_flowable(data):
    version = data.print.software_version
    content = [
        ("Printer:", data.print.printer),
        ("Software version:", f"{version.major}.{version.minor}.{version.patch}"),
        (
            "Humidity Setting:",
            f"{data.print.humidity_setting} (humidifier might be turned off)",
        ),
        ("Run Method:", data.print.run_method),
        ("Source Plate:", data.print.source_plate),
        ("Print Solutions:", f"{data.print.print_solutions} solutions"),
        ("Target Substrate:", data.print.target_substrate),
        ("Number of Targets:", f"{data.print.target_count} targets printed"),
    ]
    if data.print.pattern_file:
        content.append(("Pattern File:", data.print.pattern_file))
    nozzles = sorted(data.statistics.nozzles)
    content.append(("Number of Nozzles:", len(nozzles)))
    for nozzle in nozzles:
        content.append(
            (
                f"Settings Nozzle #{nozzle.number}:",
                f"{nozzle.voltage}V, {nozzle.pulse}µs",
            )
        )
    content.append(("Failed Drop Checks, Pre Run:", data.statistics.failed_pre_run))
    content.append(("Failed Drop Checks, Post Run:", data.statistics.failed_post_run))

    return Table(content, style=TABLE_STYLE, hAlign="LEFT")


def trim_image(image_path):
    original = PIL.Image.open(image_path)
    background = PIL.Image.new(original.mode, original.size, original.getpixel((0, 0)))
    diff = PIL.ImageChops.difference(original, background)
    diff = PIL.ImageChops.add(diff, diff, 2.0, -100)
    left, upper, right, lower = diff.getbbox()
    bbox = (left - 10, upper - 10, right + 12, lower + 10)
    cropped = original.crop(bbox) if bbox else original
    buffer = io.BytesIO()
    cropped.save(buffer, format="png")
    return ImageBuffer(buffer, cropped.width, cropped.height)


def scaled_image_flowable(image_path, width=17 * cm):
    image_buffer = trim_image(image_path)
    height = (width / image_buffer.width) * image_buffer.height
    return Image(image_buffer.buffer, width=width, height=height)


def graph_flowable(title, file_path):
    section = [
        Paragraph(title, style_h2),
        Spacer(width=17 * cm, height=0.5 * cm),
        scaled_image_flowable(file_path),
    ]
    return KeepTogether(section)

def get_failed_drop_images(failed_checks):
    return [
        FailedDropImage(item.path.with_suffix(".jpg"), item.well)
        for item in failed_checks.itertuples()
    ]

def failed_drops_flowable(nozzle, measurement):
    if measurement == "Pre Run":
        failed_checks = nozzle.drops_failed.pre_run
    elif measurement == "Post Run":
        failed_checks = nozzle.drops_failed.post_run
    else:
        raise ValueError(f"Unknown mesurement: {measurement}")
    failed_images = get_failed_drop_images(failed_checks)

    if len(failed_images) == 0:
        # no images to display here, we return early
        return []

    section = [
            PageBreak(), 
            Paragraph(f"Failed Drop Images: Nozzle #{nozzle.number}, {measurement}", style_h2)
            ]

    # group three images together
    failed_iterator = iter(failed_images)
    failed_groups = zip_longest(failed_iterator, failed_iterator, failed_iterator)

    for group in failed_groups:
        section.append(DropPictures(group))

    return section


def generate_report(data, graphs):

    story = []

    start = data.print.environment.index.min()
    start_str = start.strftime("%Y-%m-%d %H:%m")
    end = start = data.print.environment.index.max()
    end_str = end.strftime("%Y-%m-%d %H:%m")
    headline = Paragraph(f"Print {start_str} - {end_str}", style_h1)
    story.append(headline)
    story.append(Spacer(width=17 * cm, height=0.5 * cm))

    story.append(print_info_flowable(data))
    story.append(Spacer(width=17 * cm, height=0.5 * cm))

    story.append(graph_flowable("Environment Graphs", graphs.environment))

    for nozzle in sorted(data.statistics.nozzles):
        story.append(PageBreak())
        
        path = graphs.drops[nozzle.number]
        story.append(
            graph_flowable(f"Drop Check Graphs, Nozzle #{nozzle.number}", path)
        )

        story.append(Spacer(width=17 * cm, height=0.5 * cm))
        
        if len(nozzle.drops_failed.pre_run) == 0:
            failed_wells_pre_run = "-"
        else:
            failed_wells_pre_run = ", ".join(nozzle.drops_failed.pre_run["well"])
        if len(nozzle.drops_failed.post_run) == 0:
            failed_wells_post_run = "-"
        else:
            failed_wells_post_run = ", ".join(nozzle.drops_failed.post_run["well"])
        content = [
            ("Failed Pre Run Checks:", failed_wells_pre_run),
            ("Failed Post Run Checks:", failed_wells_post_run),
        ]
        story.append(Table(content, style=TABLE_STYLE, hAlign="LEFT"))
        story.extend(failed_drops_flowable(nozzle, "Pre Run"))
        story.extend(failed_drops_flowable(nozzle, "Post Run"))

    pdf_path = data.files.folder / f"{data.files.folder.name}_report.pdf"
    doc = SimpleDocTemplate(
        str(pdf_path),
        pagesize=A4,
        leftMargin=2 * cm,
        rightMargin=2 * cm,
        topMargin=2 * cm,
        bottomMargin=2 * cm,
    )
    doc.build(story)

    return pdf_path