"""Astitvaams Print Helper — the small utility that lets a hosted Astitvaams print labels. Why it exists ------------- The label is rendered by the Astitvaams server and sent to the printer over TCP:9100. That works while the server sits on the same network as the printer. A hosted server does not: the printer answers to a private address behind NAT, which is unreachable from outside. Nothing in the browser can bridge that either — there is no raw-socket API in a page or an extension, by design. So a small program runs on a machine that CAN see the printer, and the browser talks to it on localhost. It does two things and nothing else: GET /ping am I here, which version, what OS GET /discover find printers on this network (port 9100) GET /status which roll is loaded in one printer, and is it happy POST /print send a rendered label to one Everything else — label design, sizes, colours, job history — stays in Astitvaams. This utility holds no settings and no state, so a second machine that installs it prints identical labels with no setup at all. The same file runs on Windows, macOS, Linux and a Raspberry Pi. Only the installer differs. Run: python ams_print_helper.py """ from __future__ import annotations import base64 import io import json import logging import os import platform import socket import sys import threading import time import urllib.parse import webbrowser from pathlib import Path # setup_ui.py and selfinstall.py sit next to this file and are imported lazily. # Normally Python puts a script's own folder on sys.path and that just works — # but the Windows embeddable build does not: it takes sys.path ENTIRELY from its # python3xx._pth file, which lists the interpreter's folder, not the script's. # The helper then starts, answers /ping, and fails only when the setup page is # opened, which reads as "this page isn't working" with no clue why. _HERE = str(Path(__file__).resolve().parent) if _HERE not in sys.path: sys.path.insert(0, _HERE) from concurrent.futures import ThreadPoolExecutor from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer __version__ = "1.4.2" # ── Port ───────────────────────────────────────────────────────────────────── # Chosen deliberately, not casually: # * below 32768, so it cannot collide with a Linux ephemeral port (the default # range starts there, and a helper that intermittently fails to bind because # an outbound socket took its port is a support nightmare) # * outside the ranges the usual suspects claim — 3000/5000/8000/8080/8888 # (dev servers), 9100 (the printer itself), 631 (CUPS), 8181/8182 (QZ Tray) # * not registered to anything by IANA # Override with AMS_HELPER_PORT if it ever does clash. DEFAULT_PORT = 18726 PORT = int(os.environ.get("AMS_HELPER_PORT", DEFAULT_PORT)) # Bound to loopback only. The helper must never be reachable from the network: # it can print, and nothing outside this machine has any business doing that. HOST = "127.0.0.1" # Browsers send Origin on cross-origin fetches. Only Astitvaams should be able to # drive the printer, so the list is explicit rather than "*". Add your live # domain via AMS_HELPER_ORIGINS (comma separated). # Only the local dev servers are built in. Every deployment has its own address # — staging, production, a customer's own domain — so the site the helper serves # is configuration, never a constant in this file. # # With localhost alone, an install worked on a developer's machine and silently # failed for everyone else: the browser drops the request before it reaches this # process, so the log shows nothing at all and the printer looks broken when it # is fine. That is why the installer asks for the address and writes it here. DEV_ORIGINS = [ "http://localhost:5173", "http://127.0.0.1:5173", "http://localhost:4173", "http://localhost", ] def _config_dir() -> Path: """Where the installer put origins.txt — which is NOT the same place on every platform. install.sh uses $HOME/.ams-print-helper install.ps1 uses %LOCALAPPDATA%\\ams-print-helper (Windows convention) This used to be Path.home()/".ams-print-helper" on every OS, so on Windows the installer asked for the Astitvaams address, said "Astitvaams address saved", wrote it to LOCALAPPDATA — and the helper then read a different directory that did not exist. The result was a helper answering 200 on /ping while the browser refused every response with No 'Access-Control-Allow-Origin' header is present i.e. correctly installed, correctly configured, and permanently unusable, with nothing in any log to say why. """ env = os.environ.get("AMS_HELPER_DIR") if env: return Path(env) if os.name == "nt": local = os.environ.get("LOCALAPPDATA") if local: return Path(local) / "ams-print-helper" return Path.home() / ".ams-print-helper" CONFIG_DIR = _config_dir() ORIGINS_FILE = CONFIG_DIR / "origins.txt" _origins_cache: tuple[float, list[str]] = (0.0, []) def _normalise(url: str) -> str: """A browser's Origin header is scheme://host[:port] with no path or slash. Accept whatever an admin pastes — "staging.example.com/", "https://x/labels" — and reduce it to that, so a trailing slash is not the reason printing silently fails.""" url = url.strip().lstrip("\ufeff").strip() if not url: return "" if "//" not in url: url = "https://" + url parts = urllib.parse.urlsplit(url) if not parts.scheme or not parts.netloc: return "" return f"{parts.scheme}://{parts.netloc}" def _origin_lines() -> list[str]: """The real entries in origins.txt — comments and blank lines removed. Read as utf-8-sig, and the mark stripped again per line, because PowerShell's `Set-Content -Encoding UTF8` writes UTF-8 *with* a byte-order mark. That mark lands in front of the first character, and "\ufeff# comment".lstrip() does not remove it — \ufeff is not whitespace as far as str.lstrip() is concerned. So the installer's own comment line was read back as a site. _normalise() turned it into the nonsense origin "https://\ufeff", the real address was never allowed, and every browser call was refused before it arrived. The helper answered /ping perfectly from a terminal, the log stayed empty, and Astitvaams said "Not installed on this computer" about a process that was running. """ try: text = ORIGINS_FILE.read_text(encoding="utf-8-sig") except OSError: return [] out = [] for line in text.splitlines(): line = line.lstrip("\ufeff").strip() if line and not line.startswith("#"): out.append(line) return out def allowed_origins() -> list[str]: """Dev defaults + whatever the installer or admin configured. Re-read when origins.txt changes, so adding a site takes effect without a restart — the person fixing it is usually not the person who can restart a background service. """ global _origins_cache env = os.environ.get("AMS_HELPER_ORIGINS", "") configured = [_normalise(o) for o in env.split(",")] if env else [] try: mtime = ORIGINS_FILE.stat().st_mtime except OSError: mtime = 0.0 if mtime and mtime != _origins_cache[0]: from_file = [_normalise(l) for l in _origin_lines()] _origins_cache = (mtime, [o for o in from_file if o]) elif not mtime: _origins_cache = (0.0, []) seen, out = set(), [] for o in DEV_ORIGINS + [o for o in configured if o] + _origins_cache[1]: if o and o not in seen: seen.add(o) out.append(o) return out # ── The printer Astitvaams already has registered ──────────────────────────── # Written into the download by the server from Label Workspace > Printers. The # roll type is configured there and nowhere else: this file is how the helper's # own setup page learns it, so the test label uses the same roll as every real # label instead of a second default that nobody set. PRINTER_FILE = CONFIG_DIR / "printer.json" def configured_printer() -> dict: """What Astitvaams says about this site's printer. Empty when nothing is registered there — the helper still works, it just has nothing to prefer.""" try: data = json.loads(PRINTER_FILE.read_text(encoding="utf-8-sig")) except (OSError, ValueError): return {} return data if isinstance(data, dict) else {} DISCOVER_PORT = 9100 DISCOVER_TIMEOUT = 0.35 DISCOVER_WORKERS = 64 log = logging.getLogger("ams-print-helper") # ── Printer discovery ──────────────────────────────────────────────────────── def _primary_lan_ip() -> str | None: """This machine's address on the network the printer is on. Uses a UDP connect, which picks the right interface without sending anything.""" s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except OSError: return None finally: s.close() def _reachable(ip: str, port: int = DISCOVER_PORT, timeout: float = DISCOVER_TIMEOUT) -> bool: try: with socket.create_connection((ip, port), timeout=timeout): return True except OSError: return False def _printer_name(ip: str) -> str: """Model from the printer's own status page, so the list shows something recognisable rather than four numbers.""" import re import urllib.request try: with urllib.request.urlopen(f"http://{ip}/general/status.html", timeout=2) as r: body = r.read().decode("utf-8", "replace") m = re.search(r"QL-\d+\w*", body) return m.group(0) if m else "Network printer" except Exception: return "Network printer" # ── What roll is actually in the printer ───────────────────────────────────── # Brother's raster protocol answers a status request with a fixed 32-byte reply. # Only a few bytes matter here: # [8] error information 1 [10] media width (mm) # [9] error information 2 [11] media type (0x0A continuous, 0x0B die-cut) # [17] media length (mm; 0 on a continuous roll) # Source: Brother QL Series Command Reference, "Status information". # # This exists because the test label was drawn 62mm wide and printed with # label="62" no matter what was loaded. Put a 29mm roll in and the printer # rejects the job; brother_ql reports "Label mismatch", which tells the person # holding the printer nothing they can act on. Reading the roll first means the # helper prints the right size, and when it cannot, names the roll that is in # there and the roll the design wants. _ERRORS_1 = { 0x01: "No roll is loaded", 0x02: "The roll has run out", 0x04: "The cutter is jammed", 0x10: "The printer is busy with another job", 0x20: "The printer is switched off", } _ERRORS_2 = { 0x01: "Wrong roll loaded for this job", 0x02: "Printer memory is full", 0x04: "Communication error", 0x08: "Printer buffer is full", 0x10: "The printer cover is open", 0x20: "Printing was cancelled at the printer", 0x40: "The roll will not feed — check it is straight and the cover is shut", 0x80: "Printer system error", } def _label_for_media(width: int, length: int, media_type: int) -> str | None: """Turn the reported roll into the identifier brother_ql uses — "62" for a 62mm continuous roll, "62x29" for die-cut, "d24" for round. Checked against the library's own list rather than a table kept here, so a brother_ql update cannot leave the two quietly out of step. """ try: from brother_ql.labels import ALL_LABELS known = {l.identifier for l in ALL_LABELS} except Exception: known = set() candidates: list[str] = [] if media_type == 0x0B or length: if width == length: candidates.append(f"d{width}") # round die-cut candidates.append(f"{width}x{length}") candidates.append(str(width)) if not known: return candidates[0] for c in candidates: if c in known: return c return None # A Brother printer accepts one session at a time on port 9100. A single trip # through the setup page asks for status during discovery, again before the test # print, and again inside print_label — three connections in a few seconds to a # port that allows one. Cache briefly so one user action is one question. _STATUS_TTL = 15.0 _status_cache: dict[str, tuple[float, dict]] = {} def printer_status(ip: str, timeout: float = 5.0, attempts: int = 2) -> dict: """Ask the printer what roll is in it and whether it is happy. Talks the protocol directly rather than through brother_ql, so a discovery sweep does not pay for a library import per address. Best-effort by design. Not every Brother model and firmware answers a status request over the network port — a QL-820NWB that prints perfectly can still time out here — so a failure means "roll unknown", never "do not print". Everything that calls this must still work when it returns ok: false. """ hit = _status_cache.get(ip) if hit and time.monotonic() - hit[0] < _STATUS_TTL: return hit[1] request = b"\x00" * 200 + b"\x1b\x40" + b"\x1b\x69\x53" # invalidate, init, status buf = b"" last = "no reply" for attempt in range(attempts): if attempt: time.sleep(0.4) # the port may still be held by the last probe try: with socket.create_connection((ip, DISCOVER_PORT), timeout=timeout) as sock: sock.sendall(request) sock.settimeout(timeout) buf = b"" while len(buf) < 32: chunk = sock.recv(32 - len(buf)) if not chunk: break buf += chunk if len(buf) >= 32: break except OSError as e: last = str(e) buf = b"" if len(buf) < 32: # Something is listening on 9100 but it did not answer as a Brother # printer. Do not guess a roll from a short reply — say it is unknown. out = {"ok": False, "error": f"Could not read the roll from the printer at {ip} ({last}). " f"Printing still works; the roll size has to be set in Astitvaams."} _status_cache[ip] = (time.monotonic(), out) return out width, media_type, length = buf[10], buf[11], buf[17] faults = [msg for bit, msg in _ERRORS_1.items() if buf[8] & bit] faults += [msg for bit, msg in _ERRORS_2.items() if buf[9] & bit] out = { "ok": True, "media_width_mm": width, "media_length_mm": length, "continuous": media_type == 0x0A, "label": _label_for_media(width, length, media_type), "roll": f"{width}mm x {length}mm" if length else f"{width}mm continuous", "errors": faults, } # Never cache a fault: the whole point of "the cover is open" is that # somebody shuts it and presses the button again. if not faults: _status_cache[ip] = (time.monotonic(), out) return out def _printer_model(ip: str, fallback: str = "QL-820NWB") -> str: """The model brother_ql should raster for. A QL-700 handed QL-820NWB instructions prints a blank label and reports no error at all.""" name = _printer_name(ip) try: from brother_ql.models import ALL_MODELS if name in {m.identifier for m in ALL_MODELS}: return name except Exception: pass return fallback def discover(subnet: str | None = None) -> list[dict]: """Every device on this network answering on the printer port.""" ip = _primary_lan_ip() if not ip and not subnet: return [] base = subnet or ".".join(ip.split(".")[:3]) candidates = [f"{base}.{n}" for n in range(1, 255)] found: list[dict] = [] with ThreadPoolExecutor(max_workers=DISCOVER_WORKERS) as pool: for addr, ok in zip(candidates, pool.map(_reachable, candidates)): if ok: info = {"ip": addr, "name": _printer_name(addr), "online": True} # So Astitvaams can show "62mm continuous" next to the printer, and # say "the cover is open" instead of just failing to print. st = printer_status(addr) if st.get("ok"): info["roll"] = st["roll"] info["label"] = st["label"] info["errors"] = st["errors"] found.append(info) return found # ── Printing ───────────────────────────────────────────────────────────────── def print_label(image_b64: str, ip: str, label: str | None = None, model: str | None = None, red: bool = False, copies: int = 1, cut: bool = True) -> dict: """Send an already-rendered label to the printer. Astitvaams renders the PNG so the output is identical to the on-screen preview and to what a directly-connected server would produce — this only rasterises and sends. `cut` is why the utility exists at all: a PDF through the operating system's print queue cannot ask the printer to cut the tape. `label` defaults to the roll Astitvaams has registered for this printer (Label Workspace > Printers), then to whatever the printer reports, then 62. `model` defaults to the model the printer reports. Both used to be fixed at "62" and QL-820NWB — right for one roll on one machine, silently wrong for every other combination. """ from brother_ql.backends.helpers import send from brother_ql.conversion import convert from brother_ql.raster import BrotherQLRaster from PIL import Image # Ask the printer before sending anything. A jammed cutter or an open cover # is worth reporting as itself, not as a failed print. status = printer_status(ip) loaded = status.get("label") if status.get("ok") else None if status.get("ok") and status["errors"]: raise RuntimeError( "; ".join(status["errors"]) + f". Roll in the printer: {status['roll']}." ) if not label: # Astitvaams' registered roll before the printer's own report: an admin # who set 62x29 in Label Workspace must not be overruled by a printer # that reports something else. The report is the fallback and, below, # the cross-check. label = (configured_printer().get("label_size") or "").strip() or loaded or "62" if loaded and label != loaded: raise RuntimeError( f"This label is designed for the {label} roll, but the printer has " f"{status['roll']} in it. Either load the {label} roll, or pick a " f"template for {loaded} in Astitvaams." ) if not model: model = _printer_model(ip) raw = image_b64.split(",", 1)[-1] # tolerate a data: URL img = Image.open(io.BytesIO(base64.b64decode(raw))).convert("RGB") qlr = BrotherQLRaster(model) qlr.exception_on_warning = True instructions = convert( qlr=qlr, images=[img] * max(1, int(copies)), label=label, rotate="auto", threshold=70, dither=False, compress=False, red=red, cut=bool(cut), ) send(instructions=instructions, printer_identifier=f"tcp://{ip}", backend_identifier="network", blocking=True) return {"printed": max(1, int(copies)), "printer": ip, "label": label, "model": model, "roll": status.get("roll") if status.get("ok") else None} # ── HTTP ───────────────────────────────────────────────────────────────────── class Handler(BaseHTTPRequestHandler): server_version = f"AMSPrintHelper/{__version__}" def log_message(self, fmt, *args): # noqa: A003 - stdlib signature log.info("%s %s", self.address_string(), fmt % args) # -- CORS / Private Network Access ------------------------------------- def _cors(self) -> None: origin = self.headers.get("Origin", "") if origin in allowed_origins(): self.send_header("Access-Control-Allow-Origin", origin) elif not origin: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Vary", "Origin") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") # Chrome's Private Network Access: a public page reaching localhost must # be granted it explicitly, or the request never arrives. self.send_header("Access-Control-Allow-Private-Network", "true") def _json(self, code: int, body: dict) -> None: payload = json.dumps(body).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self._cors() self.end_headers() self.wfile.write(payload) def do_OPTIONS(self): # noqa: N802 - stdlib signature self.send_response(204) self._cors() self.end_headers() def handle_one_request(self): # noqa: N802 - stdlib signature """Turn any unhandled error into a readable 500. BaseHTTPRequestHandler lets an exception escape, which closes the socket with nothing written — the browser shows ERR_EMPTY_RESPONSE and the person has no idea what happened. An import that fails on one platform and not another (setup_ui on the Windows embeddable build) is exactly the case that produced a blank page and no clue. """ try: super().handle_one_request() except Exception as e: log.exception("request failed: %s", self.path) try: self._json(500, {"ok": False, "error": f"{type(e).__name__}: {e}"}) except Exception: pass # connection already gone; the log still has it def do_GET(self): # noqa: N802 path = self.path.split("?", 1)[0].rstrip("/") or "/" if path in ("/", "/ping"): self._json(200, { "ok": True, "name": "Astitvaams Print Helper", "version": __version__, "os": platform.system(), "os_release": platform.release(), "host": socket.gethostname(), "lan_ip": _primary_lan_ip(), # So the Label Workspace can say "this site is not allowed to # print" instead of the user seeing nothing happen. A caller # whose Origin is not allowed never receives this body, but a # same-origin or no-Origin probe does. "origin_allowed": (self.headers.get("Origin", "") in allowed_origins() if self.headers.get("Origin") else True), "origins_file": str(ORIGINS_FILE), }) elif path == "/setup": try: from setup_ui import SETUP_HTML except Exception as e: # A blank browser page ("this page isn't working") is the worst # possible answer here — say what is missing instead. log.exception("setup page unavailable") self._json(500, {"ok": False, "error": f"Setup page could not load: {e}. " f"setup_ui.py should sit next to {__file__}."}) return body = SETUP_HTML.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) elif path == "/setup/state": site = next(iter(_origin_lines()), "") printer = configured_printer() self._json(200, {"ok": True, "site": site, "allowed": allowed_origins(), "printer": printer, "label_size": printer.get("label_size") or ""}) elif path == "/status": # Astitvaams asks this before printing so it can show which roll is in the # printer, rather than letting the user discover the mismatch by # pressing Print and getting a rejection. qs = urllib.parse.parse_qs(self.path.partition("?")[2]) ip = (qs.get("ip") or [""])[0] if not ip: self._json(400, {"ok": False, "error": "ip is required"}) else: self._json(200, printer_status(ip)) elif path == "/discover": try: self._json(200, {"ok": True, "printers": discover()}) except Exception as e: log.exception("discover failed") self._json(500, {"ok": False, "error": str(e)}) else: self._json(404, {"ok": False, "error": "Not found"}) def do_POST(self): # noqa: N802 path = self.path.split("?", 1)[0].rstrip("/") if path.startswith("/setup"): self._setup_post(path) return if path != "/print": self._json(404, {"ok": False, "error": "Not found"}) return try: length = int(self.headers.get("Content-Length") or 0) body = json.loads(self.rfile.read(length) or b"{}") except (ValueError, TypeError): self._json(400, {"ok": False, "error": "Body must be JSON"}) return image = body.get("image") ip = body.get("ip") if not image or not ip: self._json(400, {"ok": False, "error": "image and ip are required"}) return try: result = print_label( image, ip, # No defaults here on purpose — print_label reads the roll and # the model off the printer when Astitvaams does not pin them. label=body.get("label") or None, model=body.get("model") or None, red=bool(body.get("red", False)), copies=int(body.get("copies", 1)), cut=bool(body.get("cut", True)), ) self._json(200, {"ok": True, **result}) except Exception as e: # The printer's own reason is far more useful than "print failed", # and Astitvaams shows it verbatim. log.exception("print failed") self._json(502, {"ok": False, "error": str(e)}) def _setup_post(self, path: str) -> None: """Everything the setup page needs. Loopback only, like the rest of the helper, so this is not a remote-configuration surface.""" try: length = int(self.headers.get("Content-Length") or 0) body = json.loads(self.rfile.read(length) or b"{}") except (ValueError, TypeError): body = {} if path == "/setup/site": site = _normalise(str(body.get("site", ""))) if not site: self._json(400, {"ok": False, "error": "That does not look like a web address."}) return CONFIG_DIR.mkdir(parents=True, exist_ok=True) ORIGINS_FILE.write_text(site + "\n", encoding="utf-8") log.info("Astitvaams address set to %s", site) self._json(200, {"ok": True, "saved": site}) elif path == "/setup/test-print": ip = body.get("ip") if not ip: self._json(400, {"ok": False, "error": "Find the printer first."}) return status = printer_status(ip) if status.get("ok") and status["errors"]: # "The printer cover is open" is something the person standing # next to it can fix. "Print failed" is not. self._json(502, {"ok": False, "error": "; ".join(status["errors"]), "roll": status["roll"]}) return # Order of authority, and it matters: # 1. Astitvaams > Label Workspace > Printers - the ONE place a site # configures its roll. It beats even an explicit request from # this page, so the setup screen cannot become a second place # to set the same thing. # 2. what the page asked for, only when nothing is registered # 3. what the printer reports # 4. 62, the commonest roll # Reading the printer first was wrong: it quietly overrode a roll an # admin had already set in Astitvaams, so the same printer could take # two different sizes depending on which button was pressed. registered = (configured_printer().get("label_size") or "").strip() reported = status.get("label") or "" chosen = registered or (body.get("label") or "").strip() or reported or "62" unknown_roll = not registered and not reported # Disagreement is not something to print through. A 62x29 design on a # 29 roll is rejected by the printer anyway, so the only useful # outcome is the sentence that says which two things disagree. if registered and reported and registered != reported: self._json(409, { "ok": False, "mismatch": True, "registered": registered, "reported": reported, "error": (f"Astitvaams has this printer set to {registered}, but the " f"printer has {status['roll']} loaded. Load the {registered} " f"roll, or change the roll in Astitvaams > Label Workspace " f"> Printers."), }) return try: out = print_label(_test_label_png(chosen), ip, label=chosen) note = None if unknown_roll: note = (f"No roll is set for this printer in Astitvaams and the " f"printer did not report one, so {chosen} was used. Set the " f"roll in Astitvaams > Label Workspace > Printers.") self._json(200, {"ok": True, **out, "guessed": unknown_roll, "mismatch": False, "registered": registered, "reported": reported, "note": note}) except Exception as e: log.exception("test print failed") self._json(502, {"ok": False, "error": str(e), "roll": status.get("roll")}) elif path == "/setup/rolls": # For the dropdown the setup page shows when the roll cannot be read. try: from brother_ql.labels import ALL_LABELS rolls = [{"id": l.identifier, "name": l.name} for l in ALL_LABELS] except Exception: rolls = [{"id": i, "name": i} for i in ("12", "29", "38", "50", "54", "62", "62x29", "62x100", "102")] self._json(200, {"ok": True, "rolls": rolls}) elif path == "/setup/install": from selfinstall import install_autostart self._json(200, install_autostart(CONFIG_DIR)) else: self._json(404, {"ok": False, "error": "Not found"}) def _test_label_png(label: str = "62") -> str: """A small label the setup page can print without Astitvaams being involved — proving printer, roll and cutter before the user goes anywhere near the app. Drawn here rather than shipped as a file so the binary stays one file. Sized from the roll that is actually loaded. It was fixed at 696x271 (62mm at 300dpi), so the one step meant to prove the printer works was itself the step that failed on any other roll. """ from PIL import Image, ImageDraw width, height = 696, 271 try: from brother_ql.labels import ALL_LABELS spec = next((l for l in ALL_LABELS if l.identifier == label), None) if spec: width = spec.dots_printable[0] or width # A continuous roll reports height 0 — it is cut to whatever is sent. height = spec.dots_printable[1] or max(120, width // 3) except Exception: pass img = Image.new("RGB", (width, height), "white") d = ImageDraw.Draw(img) pad = max(4, width // 90) d.rectangle([pad, pad, width - pad - 1, height - pad - 1], outline="black", width=max(2, pad // 2)) d.text((pad * 4, height // 3), "Astitvaams", fill="black") d.text((pad * 4, height // 3 + 22), f"Printer test - {label} roll", fill="black") buf = io.BytesIO() img.save(buf, format="PNG") return base64.b64encode(buf.getvalue()).decode() def serve(port: int = PORT) -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") try: httpd = ThreadingHTTPServer((HOST, port), Handler) except OSError as e: log.error("Cannot listen on %s:%s — %s", HOST, port, e) log.error("Another copy may already be running. Set AMS_HELPER_PORT to use a different port.") sys.exit(1) log.info("Astitvaams Print Helper %s listening on http://%s:%s", __version__, HOST, port) log.info("Allowed origins: %s", ", ".join(allowed_origins())) log.info("Configured in %s — add a line per site, no restart needed", ORIGINS_FILE) try: httpd.serve_forever() except KeyboardInterrupt: log.info("Stopping") httpd.shutdown() def main() -> None: """Double-click runs setup; the autostart entry runs `--serve`. A non-technical user gets a browser window that walks them through it. The background service, registered during that setup, passes --serve and never opens anything. One binary, two behaviours, no separate installer to ship or to keep in step. """ threading.current_thread().name = "main" args = set(sys.argv[1:]) if "--uninstall" in args: from selfinstall import remove_autostart r = remove_autostart() print("Removed." if r.get("ok") else f"Could not remove: {r.get('error')}") return if "--serve" in args: serve() return # Double-clicked. Open setup in the browser, then serve so the page works. logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") url = f"http://{HOST}:{PORT}/setup" try: httpd = ThreadingHTTPServer((HOST, PORT), Handler) except OSError: # Already running as a service — just show the page from that copy # rather than failing with a port clash the user cannot act on. log.info("Helper already running — opening setup") webbrowser.open(url) return log.info("Astitvaams Print Helper %s — setup at %s", __version__, url) threading.Timer(0.7, lambda: webbrowser.open(url)).start() try: httpd.serve_forever() except KeyboardInterrupt: httpd.shutdown() if __name__ == "__main__": main()