printer.py

Go to the documentation of this file.

"""!
********************************************************************************
@file   printer.py
@brief  create bon text and send to printer
********************************************************************************
"""
 
import sys
import os
import time
import enum
import logging
import queue
from typing import Any, TYPE_CHECKING
import re
import csv
 
 
import socket
import serial
from escpos.printer import Dummy
 
from Source.version import __title__
from Source.Util.app_data import EPaper, read_com_port_settings, read_paper_width_settings, \
    write_com_port_settings, write_paper_width_settings, thread_loop, S_UNIT, get_computer_name, EUser, get_serial_ports, \
    I_DEFAULT_PAPER_WIDTH
from Source.Util.gui_toolkit import THREAD, group_available_menu, config_label
 
from Source.Model.language import L_YES, L_CANCEL
from Source.Model.report import Item
 
if TYPE_CHECKING:
    from Source.Controller.main_window import MainWindow
 
log = logging.getLogger(__title__)
 
 
class EPrinter(str, enum.Enum):
    """!
    @brief Supported printers
    """
    EPSON = "EPSON"
    PROLIFIC = "Prolific"
    DEFAULT = "Default"
 
 
D_PRINTER = {
    EPrinter.EPSON: {EPaper.WIDTH_58_MM.name: 34,
                     EPaper.WIDTH_80_MM.name: 48},
    EPrinter.PROLIFIC: {EPaper.WIDTH_58_MM.name: 30,
                        EPaper.WIDTH_80_MM.name: 44},
    EPrinter.DEFAULT: {EPaper.WIDTH_58_MM.name: 30,  # max length of 58 mm paper
                       EPaper.WIDTH_80_MM.name: 44}  # auto line break or max length of 80mm paper
}
 
F_PRINTER_SLEEP_TIME = 0.2
F_PRINTER_PRINT_DELAY = 0.75  # delay for every bon to set to prevent buffer overflow of printer
S_PRINT_ERROR_FILE = "FailedPrints.csv"
 
 
def is_ip_address(s: str) -> bool:
    """!
    @brief Checks if a string is a network address.
    @param s : string to check
    @return status if is ip address
    """
    return re.fullmatch(r'(\d{1,3}\.){3}\d{1,3}(:\d+)?', s) is not None  # IPv4 with optional port
 
 
class Printer(THREAD):
    """!
    @brief Class create bons and printout.
    @param ui : main window object
    """
 
    def __init__(self, ui: "MainWindow") -> None:
        super().__init__()
        self.ui = ui
        self.b_open_drawer = False
        self.i_line_break = I_DEFAULT_PAPER_WIDTH
        self.l_available_ports: list[str] = []
        self.l_available_port_names: list[str] = []
        self.s_header1 = ""
        self.s_header2 = ""
        self.bon_queue: queue.Queue[Item] = queue.Queue()
        self.report_queue: queue.Queue[str] = queue.Queue()
        self.b_select_com_port_available = False
        self.e_printer = EPrinter.DEFAULT
        self.s_select_com_port = read_com_port_settings()
        self.i_paper_width = read_paper_width_settings()
        self.init_com_port()
 
    def set_header(self, s_header1: str = "", s_header2: str = "") -> None:
        """!
        @brief Set bon header
        @param s_header1 : first line in bon header
        @param s_header2 : second line in bon header
        """
        self.s_header1 = s_header1
        self.s_header2 = s_header2
 
    def init_com_port(self) -> None:
        """!
        @brief Initialize COM Port.
        """
        l_present_ports = get_serial_ports()
 
        self.b_select_com_port_available = False
        self.l_available_ports = []
        for port, name, _ in l_present_ports:
            log.debug("Present Port: %s (%s)", port, name)
            b_com_port_available = True
            if port == self.s_select_com_port:
                if b_com_port_available:
                    self.b_select_com_port_available = True
                for printer, _value in D_PRINTER.items():
                    if name.startswith(printer.name):
                        self.e_printer = printer
            if b_com_port_available:
                self.l_available_ports.append(port)
                self.l_available_port_names.append(name)
                log.debug("Available Port: %s (%s)", port, name)
 
        self.update_com_port_menu()
 
        if (self.s_select_com_port is not None) and (not self.b_select_com_port_available):
            self.ui.set_status([f"Select COM Port not available: {self.s_select_com_port}",
                                f"Ausgewählter COM Port nicht verfügbar: {self.s_select_com_port}"], True)
        else:
            self.update_paper_width(self.i_paper_width, False)  # call after set self.e_printer
 
    def update_com_port_menu(self) -> None:
        """!
        @brief Update COM port menu
        """
        self.ui.ag_port, all_actions_add = group_available_menu(self.ui, self.ui.l_action_com_port, self.s_select_com_port, self.l_available_ports, self.ui.action_none)
        if not all_actions_add:
            self.ui.set_status(["Too much COM Ports available", "Zu viele COM Ports verfügbar"], True)
 
    def update_com_port(self, s_com_port: str) -> None:
        """!
        @brief Update COM Port of printer
        @param s_com_port : selected COM port
        """
        if not (not self.bon_queue.empty() or not self.report_queue.empty() or self.b_open_drawer):
            b_update_port = True
 
            if b_update_port:
                self.s_select_com_port = s_com_port
                write_com_port_settings(self.s_select_com_port)
                if s_com_port is None:
                    self.ui.set_status(["Printer disabled", "Drucker deaktiviert"])
                else:
                    self.ui.set_status([f"Select Printer: {s_com_port}", f"Ausgewählter Drucker: {s_com_port}"])
 
                self.init_com_port()
            else:
                pass
 
        else:
            self.ui.set_status(["Can not change COM port. Print queue is not empty.",
                                "COM Port kann nicht geändert werden. Es befinden sich Ausdrucke in der Warteschlange."], True)
            self.update_com_port_menu()
 
    def update_paper_width(self, i_paper_width: int, b_statusbar_info: bool = True) -> None:
        """!
        @brief Update paper width.
        @param i_paper_width : paper width in "mm"
        @param b_statusbar_info : [True] show update info on status bar; [False] show not
        """
        self.i_paper_width = i_paper_width
        if i_paper_width == EPaper.WIDTH_80_MM.value:
            self.i_line_break = D_PRINTER[self.e_printer][EPaper.WIDTH_80_MM.name]
        else:
            self.i_line_break = D_PRINTER[self.e_printer][EPaper.WIDTH_58_MM.name]
        log.debug("Set line Break to %s", self.i_line_break)
        write_paper_width_settings(self.i_paper_width)
        if b_statusbar_info:
            self.ui.set_status([f"Select Paper width: {i_paper_width} mm",
                                f"Gewählte Papierbreite: {i_paper_width} mm"])
 
    def add_to_queue(self, l_items: list[Item], i_user_pos: int, f_total_price: float) -> None:
        """!
        @brief Add item to bon queue.
        @param l_items : list of items to add
        @param i_user_pos : user position
        @param f_total_price : total price
        """
        s_user = l_items[0].user
        b_save_report = True
        if self.s_select_com_port is not None:
            if self.b_select_com_port_available:
                b_print_article = self.ui.model.c_config.get_print_article_status()
                for item in l_items:
                    b_print_actual_article = self.ui.model.c_config.get_item_print_status(item.pos) if b_print_article else False
                    b_print = b_print_actual_article
                    if b_print:
                        self.bon_queue.put(item)
                self.ui.set_status([f"Print Articles for user {s_user}",
                                    f"Artikel werden gedruckt für Benutzer {s_user}"])
            else:
                b_save_report = False
                self.ui.set_status([f"{self.s_select_com_port} will print and is not available",
                                    f"{self.s_select_com_port} ist zum Drucken nicht verfügbar"], True)
        else:
            self.ui.set_status([f"Articles saved to user {s_user}",
                                f"Artikel gesichert für Benutzer {s_user}"])
        if b_save_report:
            # TODO other printer for kitchen printer and for print_article possible
            self.ui.model.c_report.write_data_to_print_file(l_items, i_user_pos, f_total_price, self.s_select_com_port)
 
    def add_to_report(self, s_text: str) -> None:
        """!
        @brief Add text to report queue.
        @param s_text : text to print out
        """
        if self.s_select_com_port is not None:
            self.report_queue.put(s_text)
 
    def open_drawer(self) -> None:
        """!
        @brief Open drawer trigger
        """
        if self.s_select_com_port is not None:
            self.b_open_drawer = True
 
    def network_print(self, network_printer_ip: str, print_content: bytes) -> bool:
        """!
        @brief Network print
        @param network_printer_ip : IP address of printer
        @param print_content : content to print
        @return success status
        """
        b_success = False
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(5)
                s.connect((network_printer_ip, 9100))
                s.sendall(print_content)
            b_success = True
        except socket.timeout:
            self.ui.set_status(["Printout to network printer failed", "Ausdruck auf Netzwerkdrucker fehlgeschlagen"], True, b_thread=True)
            b_success = False
        except Exception:
            self.ui.set_status(["Printout to network printer failed - Error unknown", "Ausdruck auf Netzwerkdrucker fehlgeschlagen - Fehler unbekannt"], True, b_thread=True)
            b_success = False
        return b_success
 
    def loop(self) -> None:
        """!
        @brief Printer loop to check for items to print out.
        """
        s_actual_used_com_port = None
        c_serial_port = None
        while True:
            if not self.bon_queue.empty() or not self.report_queue.empty() or self.b_open_drawer:
                if self.b_select_com_port_available and (s_actual_used_com_port != self.s_select_com_port):
                    if self.s_select_com_port is not None:
                        if (c_serial_port is not None) and c_serial_port.isOpen():
                            c_serial_port.close()
                        s_actual_used_com_port = self.s_select_com_port
                        c_serial_port = serial.Serial(self.s_select_com_port)
                if c_serial_port is not None:
                    if not c_serial_port.isOpen():
                        c_serial_port.open()
                    self.check_open_drawer(c_serial_port)
                    if not self.bon_queue.empty():
                        item = self.bon_queue.get()
                        network_printer_ip = None
                        bon = self.create_bon(item)
                        log.debug("Create Bon: %s", item)
                        for _ in range(item.amount):
                            if network_printer_ip:
                                b_success = self.network_print(network_printer_ip, bon)
                                if not b_success:
                                    # write failed network prints to separate log file
                                    with open(S_PRINT_ERROR_FILE, mode="a+", encoding="utf-8", newline="") as file:
                                        writer = csv.writer(file, delimiter=";")
                                        price = float(item.price_total) / item.amount
                                        s_price = f"{price:.2f}"
                                        l_item = [1, item.name, item.pos, item.group, s_price, item.user, item.date, network_printer_ip]
                                        writer.writerow(l_item)
                            else:
                                c_serial_port.write(bon)
                            log.debug("Print Bon")
                            time.sleep(F_PRINTER_PRINT_DELAY)
                            b_check_open_drawer = True
                            if b_check_open_drawer:
                                self.check_open_drawer(c_serial_port)  # open after print same items is possible
                    elif not self.report_queue.empty():
                        report = self.report_queue.get()
                        c_serial_port.write(self.create_report_bon(report))
                else:
                    self.ui.set_status("None printer will print. Disable printer and clear queue.", True, b_thread=True)  # state not possible
                    self.s_select_com_port = None
                    self.bon_queue = queue.Queue()
                    self.report_queue = queue.Queue()
                    self.b_open_drawer = False
            else:
                if (c_serial_port is not None) and c_serial_port.isOpen():
                    c_serial_port.close()
                time.sleep(F_PRINTER_SLEEP_TIME)
 
    def run(self) -> None:
        """!
        @brief Printer Thread to check for items to print out.
        """
        thread_loop(self, "Printer")
 
    def create_bon(self, item: Item) -> Any:
        """!
        @brief Create bon code for printer.
        @param item : item to create bon
        @return return bon code for printer
        """
        c_dummy = Dummy()
        s_pay = f"{S_UNIT} {item.price}".rjust(self.i_line_break)
        s_info = f"{item.date} {item.user}"[:self.i_line_break]
        s_info = s_info[:self.i_line_break]
        s_product = f"{item.name}"[:int(self.i_line_break / 2)]  # edit in btn_article_preview too if line break changed
        c_dummy.set_with_default()
        if self.s_header1:
            s_header1 = self.s_header1.center(self.i_line_break)[:self.i_line_break]
            c_dummy.text(f"{s_header1}\n")
        if self.s_header2:
            s_header2 = self.s_header2.center(self.i_line_break)[:self.i_line_break]
            c_dummy.text(f"{s_header2}\n")
        c_dummy.text(f"{s_info}\n\n")
        c_dummy.set(double_width=True, double_height=True)
        c_dummy.text(f"{s_product}")
        c_dummy.set_with_default()
        if (item.user != EUser.FREE) or self.ui.model.c_config.get_print_free_price_status():
            c_dummy.text(f"\n{s_pay}")
        if self.e_printer != EPrinter.EPSON:  # EPSON write empty lines at end as default
            c_dummy.text("\n")
        c_dummy.cut(mode="PART")
        return c_dummy.output
 
    def create_report_bon(self, s_text: str) -> Any:
        """!
        @brief Create report text for printer
        @param s_text : text
        @return return report code for printer
        """
        c_dummy = Dummy()
        c_dummy.set_with_default()
        c_dummy.text(s_text)
        c_dummy.cut(mode="PART")
        return c_dummy.output
 
    def create_open_drawer(self) -> Any:
        """!
        @brief Create open drawer code for printer
        @return return open drawer code for printer
        """
        c_dummy = Dummy()
        c_dummy.cashdraw(pin=2)  # first connector = pin2 ; second connector = pin5
        return c_dummy.output
 
    def check_open_drawer(self, serial_port: serial.Serial) -> None:
        """!
        @brief Open drawer if required
        @param serial_port : serial port
        """
        if self.b_open_drawer:
            log.debug("Open Cash Drawer")
            serial_port.write(self.create_open_drawer())
            self.b_open_drawer = False