Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ dependencies = [
"aiofiles>=24.1.0",
"httpx>=0.28.0",
"segno>=1.6.0",
"pystrich>=0.19",
"reportlab>=4.0.0",
"packaging>=24.0",
"mcp>=1.28.1", # CVE-2026-52869/52870/59950 (not reachable on stdio, cleared anyway)
"argon2-cffi>=25.1.0",
Expand Down
28 changes: 28 additions & 0 deletions src/opal/api/routes/inventory.py
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,34 @@ def get_inventory_qrcode(
return Response(content=buf.getvalue(), media_type="image/svg+xml")


@router.get("/{inventory_id}/datamatrix")
def get_inventory_datamatrix(
db: DbSession,
inventory_id: int,
) -> Response:
"""Generate a Data Matrix SVG of the OPAL number.

A short identifier needs far fewer modules per side as Data Matrix
than as a QR code (~16x16 vs. QR's 21x21+), which matters on a
narrow continuous-tape label printer (e.g. the DYMO LabelManager
280's 12mm/180dpi print strip): more px/module survives the print
resolution instead of blurring into an unscannable square.
"""
from pystrich.datamatrix import DataMatrixEncoder

record = (
db.query(InventoryRecord)
.join(Part)
.filter(InventoryRecord.id == inventory_id, Part.deleted_at.is_(None))
.first()
)
if not record:
raise HTTPException(status_code=404, detail=f"Inventory record {inventory_id} not found")

encoder = DataMatrixEncoder(record.opal_number)
return Response(content=encoder.get_svg(), media_type="image/svg+xml")


@router.get("/locations")
def list_locations(db: DbSession) -> list[LocationSummary]:
"""List all inventory locations with summary."""
Expand Down
23 changes: 23 additions & 0 deletions src/opal/api/routes/parts.py
Original file line number Diff line number Diff line change
Expand Up @@ -504,6 +504,29 @@ def get_part_qrcode(
return Response(content=buf.getvalue(), media_type="image/svg+xml")


@router.get("/{part_id}/datamatrix")
def get_part_datamatrix(
db: DbSession,
part_id: int,
) -> Response:
"""Generate a Data Matrix SVG of the part number.

A short identifier needs far fewer modules per side as Data Matrix
than as a QR code (~16x16 vs. QR's 21x21+), which matters on a
narrow continuous-tape label printer (e.g. the DYMO LabelManager
280's 12mm/180dpi print strip): more px/module survives the print
resolution instead of blurring into an unscannable square.
"""
from pystrich.datamatrix import DataMatrixEncoder

part = db.query(Part).filter(Part.id == part_id, Part.deleted_at.is_(None)).first()
if not part:
raise HTTPException(status_code=404, detail=f"Part {part_id} not found")

encoder = DataMatrixEncoder(part.internal_pn)
return Response(content=encoder.get_svg(), media_type="image/svg+xml")


@router.get("/{part_id}", response_model=PartResponse)
def get_part(
db: DbSession,
Expand Down
102 changes: 102 additions & 0 deletions src/opal/core/dymo_label.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
"""Native PDF rendering for direct-print DYMO LabelManager 280 labels.

A separate render path from web/templates/label_dymo.html (the browser
print-dialog fallback) — reportlab keeps this dependency-light instead of
bundling a headless browser, at the cost of maintaining the layout twice.
Geometry (page presets, rotation direction, cross-tape budget) is carried
over from the HTML/CSS version, which was tuned against physical hardware;
keep the two in sync by hand if either changes.
"""

from io import BytesIO

from pystrich.datamatrix import DataMatrixEncoder
from reportlab.lib.units import inch
from reportlab.lib.utils import ImageReader
from reportlab.pdfgen import canvas

PAGE_WIDTH_IN = 0.4861 # 35pt native tape-width canvas, 12mm (max) D1 tape
_SHORT_PAGE_IN = 2.0
_LONG_PAGE_IN = 3.5
_PAGE_MARGIN_IN = 0.833 # fixed on both DYMO "Label" presets
_SHORT_PRINTABLE_IN = _SHORT_PAGE_IN - _PAGE_MARGIN_IN
_CROSS_TAPE_IN = 0.32 # printable width for 12mm tape, both presets
_DM_SIZE_IN = 0.32
_GAP_IN = 0.05
_PN_FONT = ("Courier-Bold", 8)
_NAME_FONT = ("Courier-Bold", 6.5)
_META_FONT = ("Courier-Bold", 5.5)


def _lines_and_fonts(
pn: str, name: str, meta: str | None
) -> tuple[list[str], list[tuple[str, float]]]:
lines = [pn, name]
fonts = [_PN_FONT, _NAME_FONT]
if meta:
lines.append(meta)
fonts.append(_META_FONT)
return lines, fonts


def page_height_in(pn: str, name: str, meta: str | None) -> float:
"""Pick the shortest DYMO page preset that fits this label's content."""
lines, fonts = _lines_and_fonts(pn, name, meta)
widths_in = [
canvas.Canvas(BytesIO()).stringWidth(text, font, size) / 72
for text, (font, size) in zip(lines, fonts, strict=True)
]
needed_in = _DM_SIZE_IN + _GAP_IN + max(widths_in) + _GAP_IN
return _SHORT_PAGE_IN if needed_in <= _SHORT_PRINTABLE_IN else _LONG_PAGE_IN


def render_dymo_label_pdf(pn: str, name: str, meta: str | None, datamatrix_data: str) -> bytes:
"""Render a DYMO LabelManager 280 label as a print-ready PDF.

datamatrix_data is the identifier encoded in the Data Matrix — the
part number for a part label, the OPAL number for an inventory label.
"""
page_h_in = page_height_in(pn, name, meta)
page_w = PAGE_WIDTH_IN * inch
page_h = page_h_in * inch

buf = BytesIO()
c = canvas.Canvas(buf, pagesize=(page_w, page_h))
# The driver's native page is portrait (narrow = tape width, long =
# feed direction); rotate +90deg so content reads correctly once the
# tape is peeled off — confirmed against physical hardware.
c.translate(page_w / 2, page_h / 2)
c.rotate(90)

lines, fonts = _lines_and_fonts(pn, name, meta)
text_width_in = max(
c.stringWidth(text, font, size) / 72 for text, (font, size) in zip(lines, fonts, strict=True)
)
content_width_in = _DM_SIZE_IN + _GAP_IN + text_width_in
x0 = -(content_width_in * inch) / 2
band_top = (_CROSS_TAPE_IN * inch) / 2

# Data Matrix, flush left within the content block. Rendered via
# pystrich's own image encoder rather than its `.matrix` attribute,
# which turned out to be stale/wrong — get_pilimage() is the same
# code path that produced the SVG confirmed scannable on hardware.
dm_image = DataMatrixEncoder(datamatrix_data).get_pilimage(cellsize=4)
c.drawImage(
ImageReader(dm_image),
x0,
band_top - _DM_SIZE_IN * inch,
width=_DM_SIZE_IN * inch,
height=_DM_SIZE_IN * inch,
)

# Text stack, vertically split across the cross-tape band
text_x = x0 + (_DM_SIZE_IN + _GAP_IN) * inch
line_height_in = _CROSS_TAPE_IN / len(lines)
for i, (text, (font_name, font_size)) in enumerate(zip(lines, fonts, strict=True)):
baseline_y = band_top - (i + 0.8) * line_height_in * inch
c.setFont(font_name, font_size)
c.drawString(text_x, baseline_y, text)

c.showPage()
c.save()
return buf.getvalue()
72 changes: 72 additions & 0 deletions src/opal/core/printing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
"""Local CUPS printer discovery and job dispatch.

OPAL runs on one machine (see CLAUDE.md); direct-print means printing to
whatever's attached to that machine, so this shells out to the system's
own `lpstat`/`lp` rather than speaking a print protocol directly.
"""

import subprocess
import tempfile
from pathlib import Path


class PrinterError(Exception):
"""Raised when listing printers or dispatching a print job fails."""


def list_printers() -> list[dict[str, str]]:
"""List CUPS printers known to this machine, with their status."""
try:
result = subprocess.run(
["lpstat", "-p"], capture_output=True, text=True, timeout=5, check=False
)
except (OSError, subprocess.TimeoutExpired) as exc:
raise PrinterError(f"Could not query printers: {exc}") from exc

printers = []
for line in result.stdout.splitlines():
if not line.startswith("printer "):
continue
parts = line.split()
if len(parts) < 2:
continue
if " is idle" in line:
status = "idle"
elif " is printing" in line or " now printing" in line:
status = "printing"
elif " disabled " in line:
status = "disabled"
else:
status = "unknown"
printers.append({"name": parts[1], "status": status})
return printers


def print_file(printer: str, pdf_bytes: bytes) -> None:
"""Send a PDF to a named CUPS printer.

Validates the printer name against the live `lpstat -p` list rather
than trusting the caller — the name ends up as a CLI argument to
`lp`, so this also guards against an unexpected/mistyped target.
"""
known = {p["name"] for p in list_printers()}
if printer not in known:
raise PrinterError(f"Unknown printer: {printer}")

with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
f.write(pdf_bytes)
tmp_path = Path(f.name)
try:
result = subprocess.run(
["lp", "-d", printer, str(tmp_path)],
capture_output=True,
text=True,
timeout=10,
check=False,
)
if result.returncode != 0:
raise PrinterError(result.stderr.strip() or "lp failed")
except (OSError, subprocess.TimeoutExpired) as exc:
raise PrinterError(f"Could not send print job: {exc}") from exc
finally:
tmp_path.unlink(missing_ok=True)
Loading