add initial version of ptouch

diff --git a/.gitignore b/.gitignore
new file mode 100644 (file)
index 0000000..bee8a64
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+__pycache__

diff --git a/requirements.txt b/requirements.txt
new file mode 100644 (file)
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 (file)
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 (file)
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