Labeldruck und was draus folgt

Du hast eine Idee für ein Projekt?
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Also, das war jetzt schon ein wenig seltsam.
Hab in dem Code mal bei jeder einzelnen Funktion ein Print reingeschrieben um zu sehen wo der Fehler ausgelöst wurde, dabei kam dann eine Tracebackmeldung.

Code: Alles auswählen

loadconfig
lade daten
bin in der loop vom usb
warte auf usb
remove
remove
add
add
add OLEBIRD
warte auf usb
copyFile <sqlite3.Connection object at 0x7fd692626b90> OLEBIRD
<sqlite3.Connection object at 0x7fd692626b90>
config arbeitsdatei
Exception in Tkinter callback
Traceback (most recent call last):
  File "/home/earl/projekt/Zusammen2.05/alles5.01.py", line 117, in config_arbeitsdb
    cursor.execute("insert or ignore into numbers(ID) select ID from knopfdaten")
AttributeError: 'builtin_function_or_method' object has no attribute 'execute'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/earl/projekt/Zusammen2.05/alles5.01.py", line 135, in copy_files
    ergebnis = config_arbeitsdb(db_engine, name_of_stick)
  File "/home/earl/projekt/Zusammen2.05/alles5.01.py", line 117, in config_arbeitsdb
    cursor.execute("insert or ignore into numbers(ID) select ID from knopfdaten")
  File "/usr/lib/python3.6/contextlib.py", line 185, in __exit__
    self.thing.close()
AttributeError: 'builtin_function_or_method' object has no attribute 'close'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.6/tkinter/__init__.py", line 1705, in __call__
    return self.func(*args)
  File "/usr/lib/python3.6/tkinter/__init__.py", line 749, in callit
    func(*args)
  File "/home/earl/projekt/Zusammen2.05/alles5.01.py", line 292, in queue_loop
    copy_files(self.connection, name_of_stick)
  File "/home/earl/projekt/Zusammen2.05/alles5.01.py", line 142, in copy_files
    print("hier war das der Error beim Kopieren", message)
NameError: name 'message' is not defined
Da konnte ich jetzt ein wenig suchen und glaube ich hab es gefunden.
Es war die Zeile mit closing(connection.cursor)...
Das hatte ich in einem früheren Versuch die USB-Version auch schon mal. Hier geht das nicht mit dem cursor sondern mit der connection.execute.
Das hab ich jetzt mal umgesetzt und ich glaube, es geht.
Die prints muss ich noch rauswerfen und vielleicht noch was einbauen einen evtl. Fehler abzufangen, denn dafür war vermutlich das closing gedacht, oder?
So sieht es zumindest aus als würde alles klappen.
Danke für alles auf jeden Fall!!!

Code: Alles auswählen

 #!/usr/bin/env python3
# -*- coding: utf8 -*-
#import csv
import queue
import threading
import sqlite3
import time
import subprocess
import tkinter as tk
from collections import namedtuple
from tkinter import messagebox
from functools import partial
from contextlib import closing
from pathlib import Path
from datetime import datetime as DateTime
import pandas as pd
#import usbconfig
import pyudev


PFAD = Path.home() / ".DruckData"
BARCODE_DB_FILENAME = PFAD / "sqlite/db"
DATABASE = BARCODE_DB_FILENAME / "config8.db"

MEDIA_PFAD = Path("/media/earl/")
CONFIG_ON_STICK = Path("TB_Ausgabe_8iii.txt")
CSV_DATEI_TO_STICK = Path("gedruckte_nummern.csv")

ORT = "PZ Dingsbums 14"
CHECKPOINT = "42"
SQL_UPDATE_AKTUELLE_NR = """UPDATE numbers
    SET letzte_nr = (letzte_nr + 1) % 10000 
    WHERE id = ?"""
SQL_SELECT_KNOPFDATEN = """SELECT knd.ISPT_Nr,
                           num.Letzte_Nr
                           FROM knopfdaten as knd,
                           numbers as num
                           WHERE knd.ID = ? 
                       AND knd.ID = num.ID"""

SQL_OUT = """select distinct
             knopfdaten.ID,
             knopfdaten.E_St,
             knopfdaten.ISPT_Nr,
             numbers.Letzte_Nr
             from knopfdaten,
             numbers
             where
             numbers.ID = knopfdaten.ID"""

USED_COLS = [
        'ID',
        'E_St',
        'Zeile3',
        'Zeile5',
        'Zeile6',
        'Zeile7',
        'Zeilex',
        'ISPT_Nr',
        'HallenPos',
        'Aktuell_Nr',
        'Barcode'
        ]

def warte_auf_usb_stick(udev_context):
    print("warte auf usb")
    monitor = pyudev.Monitor.from_netlink(udev_context)
    monitor.filter_by("block")
    for device in iter(monitor.poll, None):
        print(device.action)
        if "ID_FS_TYPE" in device and device.action == "add":
            name_of_stick = Path(device.get("ID_FS_LABEL"))
            print(device.action, name_of_stick)
            time.sleep(2)
            return name_of_stick
    raise AssertionError("unreachable code")  # das soll auch nie in Aktion sein, wenn es richtig funktioniert

def usb_stick_loop(queue):
    print("bin in der loop vom usb")
    udev_context = pyudev.Context()
    while True:
        name_of_stick = warte_auf_usb_stick(udev_context)
        queue.put(name_of_stick)

def anzahl_drucke_dokumentieren(connection, name_of_stick, dateiname_roh):
    print("ducke doku")
    dateiname_ziel = (
        MEDIA_PFAD
        / name_of_stick
        / dateiname_roh.with_name(
            f"{dateiname_roh.stem}_{DateTime.now():%Y-%m-%d_%H_%M}.csv"
        )
    )
    pd.read_sql(SQL_OUT, connection).to_csv(
        dateiname_ziel, sep=";", decimal=",", index=False
    )

def config_arbeitsdb(connection, name_of_stick):
    print("config arbeitsdatei")
    config_datei = MEDIA_PFAD / name_of_stick / CONFIG_ON_STICK
    datensaetze = pd.read_csv(        
        config_datei,
        usecols=USED_COLS,
        na_values="x",
        quotechar='"',
        sep=";",
        encoding="utf-8",
        decimal=",",
        dtype={"ID": int, "E_St": int, "ISPT_Nr": str, "Barcode": str},
    )
    if len(datensaetze) != 8:
        raise RuntimeError("es waren keine 8 Datensätze")
    datensaetze.to_sql(
        "knopfdaten", connection, if_exists="replace", index=False
    )
    with connection as conn:
        conn.execute("insert or ignore into numbers(ID) select ID from knopfdaten")
    
    return "Daten erfolgreich übertragen"

def auswerfen(name_of_stick):
    print("werfe DS aus")
    subprocess.run(
        ["umount", MEDIA_PFAD / name_of_stick],
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )

def copy_files(connection, name_of_stick):
    print("copyFile", connection, name_of_stick)
    try:
        db_engine = connection
        print(db_engine)
        ergebnis = config_arbeitsdb(db_engine, name_of_stick)
        print(ergebnis)
        messagebox.showinfo(message=ergebnis, icon='info')
        anzahl_drucke_dokumentieren(
            db_engine, name_of_stick, CSV_DATEI_TO_STICK
        )
    except Exception as error:
        print("hier war das der Error beim Kopieren", message)
        messagebox.showinfo(message=str(error), icon='error')
    try:
        auswerfen(name_of_stick)
    except subprocess.CalledProcessError as error:
        error_text = f"Fehler {error.returncode} beim Auswerfen: {error.stdout}"
        messagebox.showinfo(message=str(error_text), icon='error')


# und bitte bessere Namen für die Spalten!
KnopfDaten = namedtuple("KnopfDaten", "id, e_st, zeile3, zeile5, zeile6, zeile7")

def lade_daten(connection):
    print("lade daten")
    with closing(connection.cursor()) as cursor:
        cursor.execute(f"select {','.join(KnopfDaten._fields)} from knopfdaten")
        return [KnopfDaten(*row) for row in cursor]

def prepare_rows_for_print(count, row, first_line):
    return "\n".join([
        "^XA",
        "^FXUnicode:",
        "^CI28",
        "^FXBox oben:",
        "^FO690,20^GB0,1150,8,^FS",
        "^FXgroßeBox:",
        "^FO15,20^GB790,1150,8^FS",
        "^FXZwischenlinie",
        "^FO495,20^GB0,1150,8,^FS",
        "^FXZwischenlinie unten",
        "^FO125,20^GB0,1150,8,^FS",
        "^FO720,200^A0R,50,50^FB800,1,0,C^FD",
        first_line,
        "^FS^FO620,200^A0R,60,60^FB800,1,0,C^FD",
        str(row.e_st),
        "^FS^FO500,200^A0R,95,95^FB800,1,0,C^FD",
        str(row.zeile3),
        "^FS^FO0,180^BY3",
        "^BCR,90,Y,N,N",
        "^FO385,370^BY4^FD",
        str(count),
        "^FS^FO260,200^A0R,70,70^FB800,,0,C^FD",
        str(row.zeile5),
        "^FS^FO120,200^A0R,60,60^FB800,2,0,C^FD",
        str(row.zeile6),
        "^FS^FO40,200^A0R,70,70^FB800,1,0,C^FD",
        str(row.zeile7),
        "^FS",
        "^XZ",
        ""
    ])

def datei_drucken(text):    
    subprocess.run(["lp", "-"], input=text.encode('utf-8'), check=True)

def zaehl_ausdrucke(connection, id):
    print("zahl ausdrucke")
    with closing(connection.cursor()) as cursor:
        cursor.execute(SQL_UPDATE_AKTUELLE_NR, [id])
        cursor.execute(SQL_SELECT_KNOPFDATEN, [id])
        ispt_nr, letzte_nr = cursor.fetchone()
        barcode = f"{ispt_nr}{CHECKPOINT}{letzte_nr}"
        cursor.execute("update knopfdaten SET Barcode = ? where id = ?",  (barcode, id))
        print(barcode, letzte_nr)
        connection.commit()
    return barcode


class MainWindow(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("Auswahl der Label")
        self["background"] = "#f2c618"
        self.connection = sqlite3.connect(f"{DATABASE}")
        button_frame = tk.Frame(self, width=1800, height=800)
        button_frame.grid(row=0, column=0, padx=0, pady=0, sticky='NSWE')
        self.buttons = []
        for index in range(8):
            row_index, column_index = divmod(index, 4)
            button = tk.Button(
                button_frame,
                bg="#f2c618",
                width=33,
                height=18,
                command=partial(self.on_click, index),
                )
            button.grid(row=row_index, column=column_index, padx=0, pady=0, sticky='NSWE')
            self.buttons.append(button)
        self.load_config()
        self.bind('<ButtonPress-1>', self.start_druck)
        self.bind('<ButtonRelease-1>', self.stopp_druck)
        self.knopf_drueck_zeit = None
        self.queue = queue.Queue()
        
        usb_looper = threading.Thread(target=usb_stick_loop, args = (self.queue, ), daemon=True)
        usb_looper.start()
        self.after(500, self.queue_loop)
        self.wait_visibility()
        self._center()

    def _center(self, *args):

        xpos = (self.winfo_screenwidth() - self.winfo_width()) / 2
        ypos = ((self.winfo_screenheight() - self.winfo_height()) / 2) / 100 * 90
        self.wm_geometry("+%d+%d" % (xpos,ypos))

        

    def load_config(self):
        print("loadconfig")
        self.config = lade_daten(self.connection)
        for button, entry in zip(self.buttons, self.config):
            button['text'] = f"{entry.e_st}\n{entry.zeile3}\n{entry.zeile5}"

    def start_druck(self, event):
        self.knopf_drueck_zeit = time.monotonic()
        print("start_druck", self.knopf_drueck_zeit)

    def stopp_druck(self, event):
        knopf_loslass_zeit = time.monotonic()
        print("stopp_druck", knopf_loslass_zeit)
        if knopf_loslass_zeit - self.knopf_drueck_zeit > 5:
            print("schalte jetzt aus")
            self.exit_application()
    
    def exit_application(self):
        result = messagebox.askyesno(
            title="Ausschalten/Neustarten",
            message="Soll der Rechner ausgeschaltet werden?\n Mit »Ja» ausschalten, mit »Nein« zurück zum Programm"
        )
        if result:
            self.destroy()
        else:
            messagebox.showinfo(
                'Zurück',
                'Das Hauptfenster wird wieder gezeigt'
            )
        
    def on_click(self, index):
        entry = self.config[index]
        count = zaehl_ausdrucke(self.connection, entry.id)
        print(count)
        string_zum_druck = prepare_rows_for_print(count, entry, ORT)
        datei_drucken(string_zum_druck)

    def queue_loop(self):
        #print("ich bin die queue")
        if not self.queue.empty():
            
            name_of_stick = self.queue.get()
            copy_files(self.connection, name_of_stick)
            self.load_config()
        self.after(500, self.queue_loop)

def main():
    BARCODE_DB_FILENAME.mkdir(parents=True, exist_ok=True)
    root = MainWindow()
    root.mainloop()
        
if __name__ == '__main__':
    main()
Sirius3
User
Beiträge: 17710
Registriert: Sonntag 21. Oktober 2012, 17:20

Es fehlten nur die Klammern

Code: Alles auswählen

with closing(connection.cursor()) as cursor:
execute von der connection-Instanz sollte man nicht benutzen, da es nicht zum Standard gehört.
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Hab ich mal probiert, das geht nicht. Es kommt keine Fehlermeldung, keine Erfolgsmeldung.
Da ich ja in jedem Teil ein print hab, kann ich dir nur sagen, wo er zuletzt was getan hat:
'warte auf usb' war das letzte was mir angezeigt wird, das steht in dem Teil:

Code: Alles auswählen

def warte_auf_usb_stick(udev_context):
    print("warte auf usb")
    monitor = pyudev.Monitor.from_netlink(udev_context)
    monitor.filter_by("block")
    for device in iter(monitor.poll, None):
        print(device.action)
        if "ID_FS_TYPE" in device and device.action == "add":
            name_of_stick = Path(device.get("ID_FS_LABEL"))
            print(device.action, name_of_stick)
            time.sleep(2)
            return name_of_stick
    raise AssertionError("unreachable code")  # das soll auch nie in Aktion sein, wenn es richtig funktioniert
Der Stick wird auch nicht ausgeworfen. Die Oberfläche läuft weiter, das Drucken geht auch, aber das Nachladen bleibt aus.

Aber, Pandas macht zudem ja auch einen Autocommit, wahrscheinlich ist das dann in dem Fall die Engine, nicht die Connection, auch wenn das so gesehen das gleiche ist - also das gleiche wahrscheinlich nicht, aber es funktioniert genauso mit der engine die sqlalchemy erzeugt und die connection von sqlite3. Das hab ich mal an einem Testmodul ausprobiert. Nur mit Pandas halt ohne Commit.
Vielleicht sollte ich das mal umbenennen in engine, auch wenn das nicht viel an der Funktionalität ändert, kann sein, dass ich den Code selbst mal wieder lesen muss in ein paar Tagen :o)
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Kurze Zwischenmeldung: Es klappt auf dem Raspberry auch.
Hab inzwischen noch rausgefunden wie ich die Knöpfe in den Vollbildmodus bringe und mit dem Ausschalten der GUI auch gleich den Rechner runterfahre.

Mein größtes Problem ist grad, dass ich die Anwendung nicht in den "Autostart" bringe. Tkinter kommt beim Start des service über systemd nicht an das Display ran. Konnte die Fehlermeldung noch nicht kopieren, vielleicht kennt die ja jemand und was man dagegen tun kann...
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Nun habe ich die Fehlermeldung aus dem jourmalctl ausgelesen, das hatte gestern nicht mehr geklappt. Da war der Text in der Datei nur noch halb da.
Hier der Teil in der der Service gestartet wird, und zwar zum Testen von mir manuell:

Code: Alles auswählen

Mai 09 22:15:50 raspberrypi systemd[1]: Started labeldrucker Service.
Mai 09 22:15:50 raspberrypi sudo[1217]: pam_unix(sudo:session): session closed for user root
Mai 09 22:15:51 raspberrypi python3[1224]: Traceback (most recent call last):
Mai 09 22:15:51 raspberrypi python3[1224]:   File "/home/pi/Labeldruck/alles5.02.py", line 294, in <module>
Mai 09 22:15:51 raspberrypi python3[1224]:     main()
Mai 09 22:15:51 raspberrypi python3[1224]:   File "/home/pi/Labeldruck/alles5.02.py", line 290, in main
Mai 09 22:15:51 raspberrypi python3[1224]:     root = MainWindow()
Mai 09 22:15:51 raspberrypi python3[1224]:   File "/home/pi/Labeldruck/alles5.02.py", line 204, in __init__
Mai 09 22:15:51 raspberrypi python3[1224]:     super().__init__()
Mai 09 22:15:51 raspberrypi python3[1224]:   File "/usr/lib/python3.7/tkinter/__init__.py", line 2023, in __init__
Mai 09 22:15:51 raspberrypi python3[1224]:     self.tk = _tkinter.create(screenName, baseName, className, interactive, wantobjects, useTk, sync, use)
Mai 09 22:15:51 raspberrypi python3[1224]: _tkinter.TclError: no display name and no $DISPLAY environment variable
Mai 09 22:15:52 raspberrypi systemd[1]: labeldrucker.service: Main process exited, code=exited, status=1/FAILURE
Mai 09 22:15:52 raspberrypi systemd[1]: labeldrucker.service: Failed with result 'exit-code'.
Mai 09 22:16:00 raspberrypi sudo[1228]:       pi : TTY=pts/0 ; PWD=/home/pi ; USER=root ; COMMAND=/bin/journalctl
Mai 09 22:16:00 raspberrypi sudo[1228]: pam_unix(sudo:session): session opened for user root by (uid=0)
Mai 09 22:16:00 raspberrypi sudo[1228]: pam_unix(sudo:session): session closed for user root
Mai 09 22:17:01 raspberrypi CRON[1273]: pam_unix(cron:session): session opened for user root by (uid=0)
Mai 09 22:17:01 raspberrypi CRON[1277]: (root) CMD (   cd / && run-parts --report /etc/cron.hourly)
Mai 09 22:17:01 raspberrypi CRON[1273]: pam_unix(cron:session): session closed for user root
Dabei habe ich die Software auf 2 Pi's getestet indem ich sie mit IDLE und im Terminal gestartet hab, in dem ich das Skript ausführbar gemacht und doppelt geklickt sowie durch ein Shellskript gestartet.
Jedesmal läuft das alles super, lediglich die Druckgeschwindigkeit ist auf einem der beiden Test-Pi's ein wenig lahmer, aber es tut was es soll ohne Fehlermeldung.
Nur als "service" gibt es diese Meldung im systemctl.
Falls jemand eine Idee hat, woran das liegen könnten, noch meine Datei labeldrucker.service die hier gestartet werden soll:

Code: Alles auswählen

[Unit]
Description=labeldrucker Service
After=multi-user.target

[Service]
User=pi
Group=pi
Type=idle
WorkingDirectory=/home/pi/Labeldruck
ExecStart=python3 /home/pi/Labeldruck/alles5.02.py &

[Install]
WantedBy=multi-user.target
Das Einzige was mir dazu einfällt ist, dass die Software vielleicht al root gestartet wird der keinen eigenen Bildschirm hat. Das ist aber nur gestochert...

Danke für Tipps, die Software zu starten mit Klick auf ein Symbol ginge auch, aber die sind auf dem Touchscreen so winzig.
Benutzeravatar
__blackjack__
User
Beiträge: 13003
Registriert: Samstag 2. Juni 2018, 10:21
Wohnort: 127.0.0.1
Kontaktdaten:

@theoS: Das wird als Benutzer `pi` gestartet, das steht ja so in der *.service-Datei. Das Problem ist das für ein grafisches Programm die grafische Oberfläche laufen muss und der Benutzer dort angemeldet sein muss. Ich würde das nicht mit systemd versuchen sondern mit der Autostart-Funktionalität der Desktop-Umgebung.
“Most people find the concept of programming obvious, but the doing impossible.” — Alan J. Perlis
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

@__blackjack__ Danke für den Tipp, dachte mir schon so ähnliches. Habe da jetzt auch was gefunden, aber das ist ja offenbar von Version zu Version anders geregelt.
Einzig ein Shellskript in /etc/profile.d tut bisher was beim Start vom Raspi, aber das muss scheinbar dann auch irgendwie als Root ausgeführt werden, beim Start. Merkwürdig ist, dass es funktionieren würde wenn ich es manuell starte.
im profile.d hab ich dieses Skript als '.sh ausführbar:

Code: Alles auswählen

Excec=/bin/bash /home/pi/Labeldruck/startlabeldruck.sh
Das Skript geht sogar noch diesen Umweg über dieses Skript:

Code: Alles auswählen

#!/bin/bash
#
sudo -H -u pi /home/pi/Labeldruck/alles5.02.py
Wie gesagt, das Progg startet aber es lässt sich nicht mehr mit F4 beenden und der Teil mit dem USB-Stick geht nicht. Da findet er den Pfad zum Stick nicht, der korrekt ausgelesen wird. Manuell über eines der beiden Skripte geht das jedesmal zuverlässig.
Bin da ein wenig irritiert, dass das bei einem Pi so kompliziert ist ein Programm als User automatisch zu starten. Das ist bei Mate ein wenig einfacher :)
Benutzeravatar
sparrow
User
Beiträge: 4164
Registriert: Freitag 17. April 2009, 10:28

Welche Desktopumgebung läuft denn auf dem Gerät?
Wenn du das weißt sollte die Antwort, wie man da ein Programm automatisch starten lässt genau 1 Google Suche weit weg sein.
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Ja, wenn das so einfach wäre. Bei den Anleitungen die von vielen netten Menschen ins Netz gestellt wird, steht bloß Rhasbian. Das was bei mir drauf ist, sollte Jessie sein. Dazu findest du dann 5 verschiedene von denen 6 nicht so funktionieren wie beschrieben.
Ich meine, es startet ja, aber mit den verkehrten Rechten.
Und wenn du danach googlest?
Dann kommst du wieder bei den Services raus.
Benutzeravatar
sparrow
User
Beiträge: 4164
Registriert: Freitag 17. April 2009, 10:28

Es ist ganz einfach. Man muss nur seine Werkzeuge beim Namen kennen. Ich habe nach der Desktopumgebung gefragt. Du hast doch weiter oben schon Mate genannt - du solltest also wissen was eine Desktopumgebung ist.

"Jessie" ist ein Debian-Release. Debian ist eine Linux-Distribution und hat nichts mit der verwendeten Desktopumgebung zu tun. Debian kann man mit KDE, Gnome, LXDE, was weiß ich benutzen.

"Rhasbian" kenne ich nicht. "Raspbian" hingegen schon.
Und wenn man nach "Raspbian" googelt, dann findet man heraus, dass die nomralerweise verwendete Desktopumgebung "LXDE" ist.
Wenn man nach "LXDE Autostart" sucht findet man als ersten Treffen deren Wiki
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Yep, danke für den Link. Das war eine der ersten die ich ausprobiert hab und die nicht geklappt hat.
Sonst hätte ich nicht nach krampfigen Alternativen gesucht. Auch was man dabei verkehrt machen kann hab ich gefunden. Der .desktop Eintrag geklickt geht aber halt nicht automatisch.
Benutzeravatar
sparrow
User
Beiträge: 4164
Registriert: Freitag 17. April 2009, 10:28

Dann machst du wohl etwas anders als dort beschrieben.
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

sparrow hat geschrieben: Montag 11. Mai 2020, 06:06 Dann machst du wohl etwas anders als dort beschrieben.
Nur ein klitzekleinwenig.
In ungefähr jeder Anleitung steht, dass man einfach eine *.desktop-Datei in den Ordner legen muss, den es nicht gibt, das reicht.
Dass dort eine Datei drin sein muss wo man diese Datei dann namentlich einträgt steht dann im Wiki verborgen in einem weiterführenden Link.
Das war der eine Grund, der zweite war, dass ich immer die Endung .desktop eingegeben hab, die ist aber auf dem Pi immer verschwunden. Also nicht nur einmal, ich hab diese Datei ja mindestens 5 mal neu angelegt an verschiedenen Orten.
Das sind so die kleinen Stolperfallen. Aber jetzt geht das.
Danke für die Hilfe von allen Seiten.

Das einzige was jetzt momentan noch »stört« ist, dass der Druck sehr langsam angestoßen wird. Was ich jetzt nicht so recht verstehe, denn ich hab das in einem zweiten Pi getestet, da ging das zwar nicht Turbo aber schon schneller. Der eine hat einen Touchscreen, der andere läuft über den großen Bildschirm. Das ist der Unterschied.
Ach ja, das ist jetzt der Code, wie er auf dem Pi läuft.

Code: Alles auswählen

#!/usr/bin/env python3
# -*- coding: utf8 -*-

import queue
import threading
import sqlite3
import time
import subprocess
import tkinter as tk
from collections import namedtuple
from tkinter import messagebox
from functools import partial
from contextlib import closing
from pathlib import Path
from datetime import datetime as DateTime
import pandas as pd
import pyudev


PFAD = Path.home() / ".DruckData"
BARCODE_DB_FILENAME = PFAD / "sqlite/db"
DATABASE = BARCODE_DB_FILENAME / "config8.db"

MEDIA_PFAD = Path("/media/pi/")
CONFIG_ON_STICK = Path("TB_Ausgabe_8.csv")
CSV_DATEI_TO_STICK = Path("gedruckte_nummern.csv")

ORT = "DAZ Dingensocken 13"
CHECKPOINT = "42"
SQL_UPDATE_AKTUELLE_NR = """UPDATE numbers
    SET letzte_nr = (letzte_nr + 1) % 10000 
    WHERE id = ?"""
SQL_SELECT_KNOPFDATEN = """SELECT knd.ISPT_Nr,
                           num.Letzte_Nr
                           FROM knopfdaten as knd,
                           numbers as num
                           WHERE knd.ID = ? 
                       AND knd.ID = num.ID"""

SQL_OUT = """select distinct
             knopfdaten.ID,
             knopfdaten.E_St,
             knopfdaten.ISPT_Nr,
             numbers.Letzte_Nr
             from knopfdaten,
             numbers
             where
             numbers.ID = knopfdaten.ID"""

USED_COLS = [
        'ID',
        'E_St',
        'Zeile3',
        'Zeile5',
        'Zeile6',
        'Zeile7',
        'Zeilex',
        'ISPT_Nr',
        'HallenPos',
        'Aktuell_Nr',
        'Barcode'
        ]

def warte_auf_usb_stick(udev_context):    
    monitor = pyudev.Monitor.from_netlink(udev_context)
    monitor.filter_by("block")
    for device in iter(monitor.poll, None):
        print(device.action)
        if "ID_FS_TYPE" in device and device.action == "add":
            name_of_stick = Path(device.get("ID_FS_LABEL"))
            print(device.action, name_of_stick)
            time.sleep(2)
            return name_of_stick
    raise AssertionError("unreachable code")  

def usb_stick_loop(queue):    
    udev_context = pyudev.Context()
    while True:
        name_of_stick = warte_auf_usb_stick(udev_context)
        queue.put(name_of_stick)

def anzahl_drucke_dokumentieren(connection, name_of_stick, dateiname_roh):    
    dateiname_ziel = (
        MEDIA_PFAD
        / name_of_stick
        / dateiname_roh.with_name(
            f"{dateiname_roh.stem}_{DateTime.now():%Y-%m-%d_%H_%M}.csv"
        )
    )
    pd.read_sql(SQL_OUT, connection).to_csv(
        dateiname_ziel, sep=";", decimal=",", index=False
    )

def config_arbeitsdb(connection, name_of_stick):    
    config_datei = MEDIA_PFAD / name_of_stick / CONFIG_ON_STICK
    datensaetze = pd.read_csv(        
        config_datei,
        usecols=USED_COLS,
        na_values="x",
        quotechar='"',
        sep=";",
        encoding="utf-8",
        decimal=",",
        dtype={"ID": int, "E_St": int, "ISPT_Nr": str, "Barcode": str},
    )
    if len(datensaetze) != 8:
        raise RuntimeError("es waren keine 8 Datensätze")
    datensaetze.to_sql(
        "knopfdaten", connection, if_exists="replace", index=False
    )
    with connection as conn:
        conn.execute("insert or ignore into numbers(ID) select ID from knopfdaten")
    
    return "Daten erfolgreich übertragen"

def auswerfen(name_of_stick):    
    subprocess.run(
        ["umount", MEDIA_PFAD / name_of_stick],
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )

def copy_files(connection, name_of_stick):    
    try:
        db_engine = connection
        
        ergebnis = config_arbeitsdb(db_engine, name_of_stick)
        
        messagebox.showinfo(message=ergebnis, icon='info')
        anzahl_drucke_dokumentieren(
            db_engine, name_of_stick, CSV_DATEI_TO_STICK
        )
    except Exception as error:        
        messagebox.showinfo(message=str(error), icon='error')
    try:
        auswerfen(name_of_stick)
    except subprocess.CalledProcessError as error:
        error_text = f"Fehler {error.returncode} beim Auswerfen: {error.stdout}"
        messagebox.showinfo(message=str(error_text), icon='error')



KnopfDaten = namedtuple("KnopfDaten", "id, e_st, zeile3, zeile5, zeile6, zeile7")
## war immer noch nicht überzeugend genug für sprechendere Spaltenüberschriften in der DB :(

def lade_daten(connection):    
    with closing(connection.cursor()) as cursor:
        cursor.execute(f"select {','.join(KnopfDaten._fields)} from knopfdaten")
        return [KnopfDaten(*row) for row in cursor]

def prepare_rows_for_print(count, row, first_line):
    return "\n".join([
        "^XA",
        "^FXUnicode:",
        "^CI28",
        "^FXBox oben:",
        "^FO690,20^GB0,1150,8,^FS",
        "^FXgroßeBox:",
        "^FO20,20^GB790,1150,8^FS",
        "^FXZwischenlinie",
        "^FO495,20^GB0,1150,8,^FS",
        "^FXZwischenlinie unten",
        "^FO125,20^GB0,1150,8,^FS",
        "^FO720,200^A0R,50,50^FB800,1,0,C^FD",
        first_line,
        "^FS^FO620,200^A0R,60,60^FB800,1,0,C^FD",
        str(row.e_st),
        "^FS^FO500,200^A0R,95,95^FB800,1,0,C^FD",
        str(row.zeile3),
        "^FS^FO0,180^BY3",
        "^BCR,90,Y,N,N",
        "^FO385,370^BY4^FD",
        str(count),
        "^FS^FO260,200^A0R,70,70^FB800,,0,C^FD",
        str(row.zeile5),
        "^FS^FO120,200^A0R,60,60^FB800,2,0,C^FD",
        str(row.zeile6),
        "^FS^FO40,200^A0R,70,70^FB800,1,0,C^FD",
        str(row.zeile7),
        "^FS",
        "^XZ",
        ""
    ])

def datei_drucken(text):
    #print(text) #zum Ausgeben ZLP-Code falls mal was dran geändert werden muss
    subprocess.run(["lp", "-"], input=text.encode('utf-8'), check=True)

def zaehl_ausdrucke(connection, id):    
    with closing(connection.cursor()) as cursor:
        cursor.execute(SQL_UPDATE_AKTUELLE_NR, [id])
        cursor.execute(SQL_SELECT_KNOPFDATEN, [id])
        ispt_nr, letzte_nr = cursor.fetchone()
        barcode = f"{ispt_nr}{CHECKPOINT}{letzte_nr}"
        cursor.execute("update knopfdaten SET Barcode = ? where id = ?",  (barcode, id))        
        connection.commit()
    return barcode


class MainWindow(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("Auswahl der Label")
        self["background"] = "#f2c618"
        self.attributes('-fullscreen', True)
        self.fullscreenState = True
        self.connection = sqlite3.connect(f"{DATABASE}")
        button_frame = tk.Frame(self, width=1024, height=720)
        button_frame.grid(row=0, column=0, padx=0, pady=0, sticky='NSWE')
        self.buttons = []
        for index in range(8):
            row_index, column_index = divmod(index, 4)
            button = tk.Button(
                button_frame,
                bg="#f2c618",
                width=22,
                height=13,
                command=partial(self.on_click, index),
                )
            button.grid(row=row_index, column=column_index, padx=0, pady=0, sticky='NSWE')
            self.buttons.append(button)
        self.load_config()
        self.bind('<ButtonPress-1>', self.start_druck)
        self.bind('<ButtonRelease-1>', self.stopp_druck)
        self.knopf_drueck_zeit = None
        self.queue = queue.Queue()        
        usb_looper = threading.Thread(target=usb_stick_loop, args = (self.queue, ), daemon=True)
        usb_looper.start()
        self.after(500, self.queue_loop)
        self.wait_visibility()
        self._center()

    def _center(self, *args):

        xpos = (self.winfo_screenwidth() - self.winfo_width()) / 2
        ypos = ((self.winfo_screenheight() - self.winfo_height()) / 2) / 100 * 90
        self.wm_geometry("+%d+%d" % (xpos,ypos))


        

    def load_config(self):        
        self.config = lade_daten(self.connection)
        for button, entry in zip(self.buttons, self.config):
            button['text'] = f"{entry.e_st}\n{entry.zeile3}\n{entry.zeile5}"

    def start_druck(self, event):
        self.knopf_drueck_zeit = time.monotonic()
        

    def stopp_druck(self, event):
        knopf_loslass_zeit = time.monotonic()
        
        if knopf_loslass_zeit - self.knopf_drueck_zeit > 5:            
            self.exit_application()
    
    def exit_application(self):
        result = messagebox.askyesno(
            title="Ausschalten/Neustarten",
            message="Soll der Rechner ausgeschaltet werden?\n Mit »Ja» ausschalten, mit »Nein« zurück zum Programm"
        )
        if result:
            self.destroy()
            subprocess.run(['shutdown', '-h', 'now'], shell=False) # Rechner Ausschalten (nur für Pi)
        else:
            messagebox.showinfo(
                'Zurück',
                'Das Hauptfenster wird wieder gezeigt'
            )
        
    def on_click(self, index):
        entry = self.config[index]
        count = zaehl_ausdrucke(self.connection, entry.id)        
        string_zum_druck = prepare_rows_for_print(count, entry, ORT)
        datei_drucken(string_zum_druck)

    def queue_loop(self):        
        if not self.queue.empty():            
            name_of_stick = self.queue.get()
            copy_files(self.connection, name_of_stick)
            self.load_config()
        self.after(500, self.queue_loop)

def main():
    BARCODE_DB_FILENAME.mkdir(parents=True, exist_ok=True)
    root = MainWindow()
    root.mainloop()
        
if __name__ == '__main__':
    main()
theoS
User
Beiträge: 108
Registriert: Dienstag 5. November 2019, 21:44

Das mit dem Druck habe ich jetzt auch hingekriegt. Das Ding druckt jetzt brav über lprng ganz ohne Cups und sonst.
Wenn es jetzt noch in der Praxis funktioniert und nicht noch 100 Änderungswünsche hinzukommen, mache ich ein Fass auf!
Danke an alle!
Antworten