weaponizzata la logica di controllo della stampante
Guido Longoni
parent
336e47216a
commit
c9de984608
|
|
@ -1 +1,6 @@
|
|||
MkpWIFI/__pycache__
|
||||
# VS Codium user settings directory
|
||||
.vscode
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
|
|
@ -1,21 +1,33 @@
|
|||
{
|
||||
"ip":"192.168.40.214",
|
||||
"port": 8899,
|
||||
"name":"MKPsouplesse",
|
||||
"autoconnect":true,
|
||||
"keepalive":5000,
|
||||
"timeout": 0.5,
|
||||
"inkstat" : 2,
|
||||
"printcmd": {
|
||||
"filename":"TM1.00I",
|
||||
"dateformat": "%d%m%y-",
|
||||
"handshake" : "*",
|
||||
"printmode" : "11",
|
||||
"headstatus" : [
|
||||
"ink", "htmr", "iuse", "xclk", "bat",
|
||||
"txt", "pause", "headOk", "headTyp",
|
||||
"labelCnt", "ztv", "cp", "lx", "exInk",
|
||||
"scStat", "inkStat"
|
||||
]
|
||||
}
|
||||
}
|
||||
"ip": "192.168.40.30",
|
||||
"port": 8899,
|
||||
"name": "MKPsouplesse",
|
||||
"autoconnect": true,
|
||||
"keepalive": 5000,
|
||||
"timeout": 0.5,
|
||||
"inkstat": 2,
|
||||
"printcmd": {
|
||||
"filename": "TM1.00I",
|
||||
"dateformat": "%d%m%y-",
|
||||
"handshake": "*",
|
||||
"printmode": "11",
|
||||
"headstatus": [
|
||||
"ink",
|
||||
"htmr",
|
||||
"iuse",
|
||||
"xclk",
|
||||
"bat",
|
||||
"txt",
|
||||
"pause",
|
||||
"headOk",
|
||||
"headTyp",
|
||||
"labelCnt",
|
||||
"ztv",
|
||||
"cp",
|
||||
"lx",
|
||||
"exInk",
|
||||
"scStat",
|
||||
"inkStat"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
|
|||
217
MkpWIFI/main.py
217
MkpWIFI/main.py
|
|
@ -5,186 +5,87 @@ Created on 8 apr 2019
|
|||
'''
|
||||
|
||||
import sys
|
||||
import time
|
||||
import datetime
|
||||
import socket
|
||||
import json
|
||||
import PyQt5.QtWidgets
|
||||
from MkpWIFI.MKPrint_GUI import Ui_MainWindow
|
||||
from MKPrint_GUI import Ui_MainWindow
|
||||
from printer import MarkoPrinter
|
||||
|
||||
|
||||
class main(PyQt5.QtWidgets.QMainWindow):
|
||||
|
||||
def __init__(self):
|
||||
class MainWindow(PyQt5.QtWidgets.QMainWindow):
|
||||
def __init__(self, configFilePath):
|
||||
try:
|
||||
self.printer = MarkoPrinter(configFilePath)
|
||||
except FileNotFoundError as e:
|
||||
print("Configuration file error: {}".format(e))
|
||||
sys.exit(1)
|
||||
|
||||
PyQt5.QtWidgets.QMainWindow.__init__(self)
|
||||
self.ui = Ui_MainWindow()
|
||||
self.ui.setupUi(self)
|
||||
self.setFixedSize(self.maximumSize())
|
||||
self.setWindowTitle("Manda Etichette a Marcatrice")
|
||||
|
||||
self.inkStatCounter = 0
|
||||
self.printer = None
|
||||
self.headStatus = dict()
|
||||
self.connected = False
|
||||
self.sending = False
|
||||
self.timer = PyQt5.QtCore.QTimer()
|
||||
|
||||
self.ui.btn_connect.clicked.connect(self.connetti)
|
||||
self.ui.btn_connect.clicked.connect(self.toggla_connessione)
|
||||
self.ui.btn_upload.clicked.connect(self.upload)
|
||||
|
||||
self.ui.btn_upload.setEnabled(False)
|
||||
self.ui.rad_dataOn.setChecked(True)
|
||||
|
||||
self.ui.prg_inkLevel.setTextVisible(True)
|
||||
|
||||
try:
|
||||
self.setFile = open("conf.json", "r")
|
||||
self.settings = json.load(self.setFile)
|
||||
self.data = self.settings["printcmd"]
|
||||
except FileNotFoundError as e:
|
||||
print("Configuration file error: {}".format(e))
|
||||
sys.exit(1)
|
||||
pass
|
||||
|
||||
def connetti(self):
|
||||
if not self.connected:
|
||||
if self.printer is None:
|
||||
try:
|
||||
self.printer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.printer.connect((self.settings["ip"], self.settings["port"]))
|
||||
self.printer.settimeout(0.5)
|
||||
self.ui.btn_connect.setText("DISCONNETTI")
|
||||
self.ui.lbl_stat.setText("Connesso")
|
||||
self.ui.btn_upload.setEnabled(True)
|
||||
self.connected = True
|
||||
self.status()
|
||||
self.timer.start(self.settings["keepalive"])
|
||||
self.timer.timeout.connect(self.keepalive)
|
||||
except socket.error as e :
|
||||
print ("Connessione Fallita: {}".format(e))
|
||||
self.printer = None
|
||||
self.connected = False
|
||||
pass
|
||||
|
||||
def updateStatusView(self):
|
||||
connessa = self.printer.connected
|
||||
self.ui.btn_upload.setEnabled(connessa)
|
||||
if connessa:
|
||||
self.ui.prg_inkLevel.setValue(self.printer.headStatus["iuse"])
|
||||
self.ui.lcd_printCount.display(
|
||||
str(self.printer.headStatus["labelCnt"]))
|
||||
else:
|
||||
self.ui.prg_inkLevel.setValue(0)
|
||||
self.ui.lcd_printCount.display('0')
|
||||
|
||||
def toggla_connessione(self):
|
||||
connessa = self.printer.connected
|
||||
if connessa:
|
||||
connessa = self.printer.disconnetti()
|
||||
else:
|
||||
connessa = self.printer.connetti()
|
||||
if connessa:
|
||||
self.ui.btn_connect.setText("DISCONNETTI")
|
||||
self.ui.lbl_stat.setText("Connesso")
|
||||
self.timer.start(self.printer.settings["keepalive"])
|
||||
self.timer.timeout.connect(self.keepalive)
|
||||
else:
|
||||
self.disconnetti()
|
||||
pass
|
||||
pass
|
||||
|
||||
def disconnetti(self):
|
||||
if self.connected:
|
||||
self.timer.stop()
|
||||
self.connected = False
|
||||
self.ui.btn_connect.setText("CONNETTI")
|
||||
self.ui.lbl_stat.setText("Disconnesso")
|
||||
self.printer.close()
|
||||
self.printer = None
|
||||
pass
|
||||
|
||||
def upload(self):
|
||||
self.sending=True
|
||||
if self.connected:
|
||||
if self.ui.rad_dataOn.isChecked():
|
||||
date = datetime.datetime.now()
|
||||
date = date.strftime(self.data["dateformat"])
|
||||
else:
|
||||
date = ""
|
||||
|
||||
cmdstr = "\x02TZ" + self.data["filename"] + ";" + self.data["printmode"] + "\r"
|
||||
descr = str(self.ui.txt_descr.toPlainText()).capitalize()
|
||||
print(bytearray(descr,"ascii"))
|
||||
descr = descr.replace("\n", "\t")
|
||||
datastr = str(self.ui.txt_model.text()).upper() + "\r" + \
|
||||
date + descr + "\r"
|
||||
outstr = cmdstr + datastr + "\03\r"
|
||||
self.printer.settimeout(1.0)
|
||||
try:
|
||||
self.printer.send(bytearray(outstr, "ascii"))
|
||||
retval = self.printer.recv(1)
|
||||
print("ACK: ", retval)
|
||||
if retval == b'\x06':
|
||||
retval = self.printer.recv(6)
|
||||
if retval == b'\x021OK\x03':
|
||||
print("RESP: ", retval)
|
||||
self.ui.lbl_stat.setText("Invio Riuscito!")
|
||||
print("Printer OK")
|
||||
else:
|
||||
self.ui.lbl_stat.setText("Invio Fallito!")
|
||||
print("Printer Error")
|
||||
else:
|
||||
print("Send Error")
|
||||
self.printer.settimeout(0.5)
|
||||
|
||||
self.status()
|
||||
|
||||
if self.headStatus["ink"] == 2:
|
||||
self.ui.lbl_stat.setText("CARTUCCIA ESAURITA\nSostituire!")
|
||||
elif self.headStatus["ink"] == 1:
|
||||
self.ui.lbl_stat.setText("Cartuccia in Esaurimento")
|
||||
|
||||
except socket.error as e:
|
||||
print("Socket error: ", e)
|
||||
self.printer.settimeout(0.5)
|
||||
self.ui.lbl_stat.setText("Invio Fallito!")
|
||||
self.printer.sendall(b"\r"*5)
|
||||
self.printer.recv(5)
|
||||
self.sending = False
|
||||
pass
|
||||
|
||||
|
||||
self.ui.btn_connect.setText("CONNETTI")
|
||||
self.updateStatusView()
|
||||
|
||||
def keepalive(self):
|
||||
if not self.sending:
|
||||
print ("Keepalive")
|
||||
if self.connected:
|
||||
try:
|
||||
self.printer.send(b"\x1B*\r")
|
||||
print(str(self.printer.recv(5)))
|
||||
self.ui.lbl_stat.setText("Connesso")
|
||||
if self.inkStatCounter <= self.settings["inkstat"]:
|
||||
self.inkStatCounter +=1
|
||||
else:
|
||||
self.status()
|
||||
self.inkStatCounter = 0
|
||||
except:
|
||||
self.disconnetti()
|
||||
else:
|
||||
self.ui.lbl_stat.setText("Disconnesso")
|
||||
pass
|
||||
|
||||
def status(self):
|
||||
if self.connected:
|
||||
try:
|
||||
self.printer.send(b"\x1BS1\r")
|
||||
time.sleep(1)
|
||||
c = self.printer.recv(1)
|
||||
inData = b""
|
||||
while c != b"\r":
|
||||
inData += c
|
||||
if len(inData) > 200:
|
||||
print("Status Message Length Error")
|
||||
raise
|
||||
c=self.printer.recv(1)
|
||||
inData = str(inData,"ascii")
|
||||
inData = inData.split(":")
|
||||
for val in enumerate(self.data["headstatus"]):
|
||||
if val[1] != "txt":
|
||||
self.headStatus[val[1]] = int(inData[val[0]])
|
||||
else:
|
||||
self.headStatus[val[1]] = str(inData[val[0]])
|
||||
print("Status: {}".format(self.headStatus))
|
||||
self.ui.prg_inkLevel.setValue(self.headStatus["iuse"])
|
||||
self.ui.lcd_printCount.display(str(self.headStatus["labelCnt"]))
|
||||
except:
|
||||
print("Status retreive error!")
|
||||
pass
|
||||
|
||||
# TODO: usare il primo argomento di keepalive (stato connessione) per aggiornare la ui
|
||||
_, statusUpdated = self.printer.keepalive()
|
||||
if statusUpdated:
|
||||
self.updateStatusView()
|
||||
|
||||
def upload(self):
|
||||
if self.ui.rad_dataOn.isChecked():
|
||||
date = datetime.datetime.now()
|
||||
date = date.strftime(self.printer.data["dateformat"])
|
||||
else:
|
||||
date = ""
|
||||
descr = str(self.ui.txt_descr.toPlainText()).capitalize()
|
||||
datastr = str(self.ui.txt_model.text()).upper() + "\r" + \
|
||||
date + descr + "\r"
|
||||
response, statusUpdated = self.printer.upload(descr, datastr)
|
||||
if statusUpdated:
|
||||
self.updateStatusView()
|
||||
self.ui.lbl_stat.setText(response)
|
||||
|
||||
def quit(self):
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
app = PyQt5.QtWidgets.QApplication(sys.argv)
|
||||
window = main()
|
||||
configFilePath = "MkpWIFI/conf.json"
|
||||
window = MainWindow(configFilePath)
|
||||
window.show()
|
||||
sys.exit(app.exec_())
|
||||
pass
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,132 @@
|
|||
import socket
|
||||
import json
|
||||
import time
|
||||
|
||||
|
||||
class MarkoPrinter:
|
||||
def __init__(self, configFilePath):
|
||||
self.inkStatCounter = 0
|
||||
self.printSock = None
|
||||
self.headStatus = dict()
|
||||
self.connected = False
|
||||
self.sending = False
|
||||
with open(configFilePath, "r") as setFile:
|
||||
self.settings = json.load(setFile)
|
||||
self.data = self.settings["printcmd"]
|
||||
self.debug = True
|
||||
|
||||
def log(self, *logtxt):
|
||||
if self.debug:
|
||||
print(*logtxt)
|
||||
|
||||
def connetti(self):
|
||||
if not self.connected:
|
||||
if self.printSock is None:
|
||||
try:
|
||||
self.printSock = socket.socket(
|
||||
socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.printSock.connect(
|
||||
(self.settings["ip"], self.settings["port"]))
|
||||
self.connected = True
|
||||
self.printSock.settimeout(0.5)
|
||||
self.checkStatus()
|
||||
except socket.error as e:
|
||||
self.printSock = None
|
||||
self.connected = False
|
||||
self.log("Connessione Fallita: {}".format(e))
|
||||
else:
|
||||
self.printSock.close()
|
||||
self.printSock = None
|
||||
self.connetti()
|
||||
return self.connected
|
||||
|
||||
def disconnetti(self):
|
||||
if self.connected:
|
||||
self.printSock.close()
|
||||
self.printSock = None
|
||||
self.connected = False
|
||||
return self.connected
|
||||
|
||||
def checkStatus(self):
|
||||
statusUpdated = False
|
||||
if self.connected:
|
||||
try:
|
||||
self.printSock.send(b"\x1BS1\r")
|
||||
time.sleep(1)
|
||||
c = self.printSock.recv(1)
|
||||
inData = b""
|
||||
while c != b"\r":
|
||||
inData += c
|
||||
if len(inData) > 200:
|
||||
self.log("Status Message Length Error")
|
||||
raise IOError
|
||||
c = self.printSock.recv(1)
|
||||
inData = str(inData, "ascii")
|
||||
inData = inData.split(":")
|
||||
for index, key in enumerate(self.data["headstatus"]):
|
||||
if key != "txt":
|
||||
self.headStatus[key] = int(inData[index])
|
||||
else:
|
||||
self.headStatus[key] = str(inData[index])
|
||||
self.log("Status: {}".format(self.headStatus))
|
||||
statusUpdated = True
|
||||
except:
|
||||
self.log("Status retreive error!")
|
||||
return statusUpdated
|
||||
|
||||
def upload(self, descr, datastr):
|
||||
self.sending = True
|
||||
response = ""
|
||||
if self.connected:
|
||||
cmdstr = "\x02TZ" + self.data["filename"] + \
|
||||
";" + self.data["printmode"] + "\r"
|
||||
self.log(bytearray(descr, "ascii"))
|
||||
descr = descr.replace("\n", "\t")
|
||||
outstr = cmdstr + datastr + "\03\r"
|
||||
self.printSock.settimeout(1.0)
|
||||
try:
|
||||
self.printSock.send(bytearray(outstr, "ascii"))
|
||||
retval = self.printSock.recv(1)
|
||||
self.log("ACK: ", retval)
|
||||
if retval == b'\x06':
|
||||
retval = self.printSock.recv(6)
|
||||
if retval == b'\x021OK\x03':
|
||||
self.log("RESP: ", retval)
|
||||
response = "Invio Riuscito!"
|
||||
self.log("Printer OK")
|
||||
else:
|
||||
response = "Invio Fallito!"
|
||||
self.log("Printer Error")
|
||||
else:
|
||||
self.log("Send Error")
|
||||
self.printSock.settimeout(0.5)
|
||||
statusUpdated = self.checkStatus()
|
||||
if self.headStatus["ink"] == 2:
|
||||
response = "CARTUCCIA ESAURITA\nSostituire!"
|
||||
elif self.headStatus["ink"] == 1:
|
||||
response = "Cartuccia in Esaurimento"
|
||||
except socket.error as e:
|
||||
self.log("Socket error: ", e)
|
||||
self.printSock.settimeout(0.5)
|
||||
response = "Invio Fallito!"
|
||||
self.printSock.sendall(b"\r"*5)
|
||||
self.printSock.recv(5)
|
||||
self.sending = False
|
||||
return response, statusUpdated
|
||||
|
||||
def keepalive(self):
|
||||
statusUpdated = False
|
||||
if not self.sending:
|
||||
self.log("Keepalive")
|
||||
if self.connected:
|
||||
try:
|
||||
self.printSock.send(b"\x1B*\r")
|
||||
self.log(str(self.printSock.recv(5)))
|
||||
if self.inkStatCounter <= self.settings["inkstat"]:
|
||||
self.inkStatCounter += 1
|
||||
else:
|
||||
statusUpdated = self.checkStatus()
|
||||
self.inkStatCounter = 0
|
||||
except:
|
||||
self.disconnetti()
|
||||
return self.connected, statusUpdated
|
||||
Loading…
Reference in New Issue