From: Don Armstrong Date: Sun, 3 Mar 2024 23:45:32 +0000 (-0800) Subject: add initial version of ptouch X-Git-Url: https://git.donarmstrong.com/?a=commitdiff_plain;h=0a25979dd911d50464e90036ae67e74225ec346c;p=ptouch_web.git add initial version of ptouch --- 0a25979dd911d50464e90036ae67e74225ec346c diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bee8a64 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__ diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a19fd91 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastapi +uvicorn +dash diff --git a/src/ptouch_web/api.py b/src/ptouch_web/api.py new file mode 100644 index 0000000..8b56a6a --- /dev/null +++ b/src/ptouch_web/api.py @@ -0,0 +1,142 @@ +from fastapi import FastAPI, BackgroundTasks, HTTPException +from pydantic import BaseModel, Field +from fastapi.responses import RedirectResponse +from fastapi.middleware.wsgi import WSGIMiddleware +from .dashapp import app as dash_app +import uuid +import io +import subprocess +import asyncio +from datetime import datetime + +from tempfile import NamedTemporaryFile +from typing import Optional, Union + + +class QueueBase(BaseModel): + """Base model for queues""" + + id: uuid.UUID = Field(default_factory=uuid.uuid4) + type: str = "base" + printed_at: datetime | None = None + failed: bool = False + job_result: str = "" + copies: int = 1 + + def ptouch_arguments(self) -> list[str]: + raise NotImplementedError("Don't call this method on the QueueBase class") + + def cleanup(self): + """Clean up any temporary files after printing""" + pass + + +class QueueText(QueueBase): + """Text Queue Item""" + + type: str = "text" + text: list[str] = Field(default_factory=list) + fontsize: Optional[int] = None + + def ptouch_arguments(self) -> list[str]: + args = [] + if self.fontsize is not None: + args.extend(["--fontsize", str(self.fontsize)]) + args.extend(["--text", *self.text]) + + return args + + +class QueueImage(QueueBase): + """Image (PNG) Queue Item""" + + type: str = "image" + image: bytes + _temp_file: Optional[NamedTemporaryFile] = None + + def ptouch_arguments(self) -> list[str]: + if self._temp_file is None: + # write the image out to temp file + self._temp_file = NamedTemporaryFile() + self._temp_file.write(self.image) + return ["--image", self._temp_file.name] + + def cleanup(self): + """Close the temporary file created when this item is printed.""" + if self._temp_file is not None: + self._temp_file.close() + self._temp_file = None + + +async def print_queue_item(queue_item: QueueBase) -> QueueBase: + """Print a queue item, capturing the results of the call. + + :param queue_item: The queue item to print + + :returns: The queue item passed, after running it and marking the + job printed + """ + for i in range(0, queue_item.copies): + stdout = io.StringIO + proc = await asyncio.create_subprocess_exec( + "ptouch-print", + *queue_item.ptouch_arguments(), + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + (stdout, stderr) = await proc.communicate() + queue_item.cleanup() + queue_item.printed_at = datetime.now() + if proc.returncode != 0: + queue_item.failed = True + break + queue_item.job_result += stdout.decode() + return queue_item + + +app = FastAPI() + +app.mount("/app", WSGIMiddleware(dash_app.server)) + +label_queue = dict() + + +@app.get("/") +async def redirect_dash(): + "Redirect to the web application" + return RedirectResponse("/app") + + +@app.get("/v1/queue", response_model=list[Union[QueueText, QueueImage]]) +async def list_printer_queue(): + """Return the items in the printer queue""" + return list(label_queue.values()) + + +@app.post("/v1/queue", response_model=Union[QueueText, QueueImage]) +async def add_to_printer_queue( + queue_item: Union[QueueText, QueueImage], background_tasks: BackgroundTasks +): + """Add an item to the printer queue for printing""" + # validate the item + if queue_item.id in label_queue: + # This queue item already exists(?) + raise HTTPException( + status_code=400, detail=f"Duplicate queue ID {queue_item.id}" + ) + + # add it to the queue + label_queue[queue_item.id] = queue_item + background_tasks.add_task(print_queue_item, queue_item) + + return queue_item + + +@app.get("/v1/queue/{queue_id}", response_model=Union[QueueText, QueueImage]) +async def get_printer_queue_item_by_id(queue_id: uuid.UUID): + """Return one item in the printer queue""" + if queue_id not in label_queue: + raise HTTPException( + status_code=404, detail=f"Queue id {queue_id} does not exist" + ) + return label_queue[queue_id] diff --git a/src/ptouch_web/dashapp.py b/src/ptouch_web/dashapp.py new file mode 100644 index 0000000..c3ff627 --- /dev/null +++ b/src/ptouch_web/dashapp.py @@ -0,0 +1,47 @@ +from dash import Dash, html, dcc, Input, Output, State, ctx +import requests + +app = Dash(__name__, requests_pathname_prefix="/app/") + +app.layout = html.Div( + [ + html.Div( + children=[ + html.Label("Text input"), + dcc.Textarea( + id="label-text", + style={"width": "100%"}, + placeholder="Text for label", + ), + html.Label("Font size"), + dcc.Input(id="font-size", type="number", min=0, max=100), + html.Label("Copies"), + dcc.Input(id="copies", value=1, type="number", min=0, max=100), + html.Button("Print", name="print", id="print"), + html.Div(id="print-result"), + ] + ), + ] +) + + +@app.callback( + Output("print-result", "children"), + Output("copies", "value"), + Input("print", "n_clicks"), + State("label-text", "value"), + State("font-size", "value"), + State("copies", "value"), + prevent_initial_call=True, +) +def print_button(print, label_text, font_size, copies): + if ctx.triggered_id != "print": + return + # fix up the label text + queue_item = dict(text=label_text.split("\n"), type="text", copies=copies) + + if font_size is not None: + queue_item["fontsize"] = int(font_size) + + res = requests.post("http://localhost:8000/v1/queue", json=queue_item) + return res.text, 1