nothing works, shits broken
This commit is contained in:
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,3 +1,4 @@
|
|||||||
*.env
|
*.env
|
||||||
*venv
|
*venv
|
||||||
__pycache__
|
__pycache__
|
||||||
|
server_config.json
|
||||||
88
config.py
88
config.py
@@ -1,38 +1,86 @@
|
|||||||
import json
|
import json
|
||||||
import printer
|
import printer
|
||||||
|
|
||||||
|
config_filename = "server_config.json"
|
||||||
|
|
||||||
|
|
||||||
class PrinterConfig:
|
class PrinterConfig:
|
||||||
editables = {"printer_address", "printer_port"}
|
editables = {"address", "port"}
|
||||||
required_params = {"printer_address"}
|
required_params = {"address"}
|
||||||
|
_address = None
|
||||||
def __init__(self):
|
_port = 9100
|
||||||
self.printer_address = None
|
_profile = "RP-F10-80mm"
|
||||||
self.printer_port = 9100
|
_printer = None
|
||||||
self.printer_profile = "RP-F10-80mm"
|
_dirty = True
|
||||||
self.__printer = None
|
|
||||||
|
|
||||||
def save(self):
|
|
||||||
with open("config.json", "w", encoding="utf-8") as ofd:
|
|
||||||
json.dump(self.__dict__, ofd, ensure_ascii=False)
|
|
||||||
|
|
||||||
def load(self):
|
def load(self):
|
||||||
with open("config.json", "r", encoding="utf-8") as ifd:
|
"""Load printer configuration from config.json file."""
|
||||||
self.__dict__ = json.load(ifd)
|
with open(config_filename, "r", encoding="utf-8") as ifd:
|
||||||
|
loaded_config = json.load(ifd)
|
||||||
|
for param_name in self.editables:
|
||||||
|
if param_name in loaded_config:
|
||||||
|
setattr(self, param_name, loaded_config[param_name])
|
||||||
|
|
||||||
|
def save(self):
|
||||||
|
"""Save current printer configuration to config.json file."""
|
||||||
|
with open(config_filename, "w", encoding="utf-8") as ofd:
|
||||||
|
json.dump(
|
||||||
|
{"address": self.address, "port": self.port, "profile": self.profile},
|
||||||
|
ofd,
|
||||||
|
ensure_ascii=False,
|
||||||
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def complete(self):
|
def complete(self):
|
||||||
|
"""Check if all required parameters are set.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True if all required parameters are set, False otherwise.
|
||||||
|
"""
|
||||||
for param in self.required_params:
|
for param in self.required_params:
|
||||||
if getattr(self, param) is None:
|
if getattr(self, param) is None:
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
@property
|
||||||
|
def port(self):
|
||||||
|
return self._port
|
||||||
|
|
||||||
|
@port.setter
|
||||||
|
def port(self, new_value: int):
|
||||||
|
if not isinstance(new_value, int) or 0 > new_value < 65535:
|
||||||
|
new_value = 9100
|
||||||
|
self._port = new_value
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self):
|
||||||
|
return self._address
|
||||||
|
|
||||||
|
@address.setter
|
||||||
|
def address(self, new_value: str):
|
||||||
|
if not isinstance(new_value, str):
|
||||||
|
new_value = ""
|
||||||
|
self._address = new_value
|
||||||
|
|
||||||
|
@property
|
||||||
|
def profile(self):
|
||||||
|
return self._profile
|
||||||
|
|
||||||
|
@profile.setter
|
||||||
|
def profile(self, new_value: str):
|
||||||
|
raise AttributeError("Printer profile is a fixed value. Check config.py.")
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def printer(self):
|
def printer(self):
|
||||||
if self.__printer is None:
|
if self._printer is not None:
|
||||||
self.__printer = printer.Printer(
|
self._printer.printer.close()
|
||||||
address=self.printer_address,
|
|
||||||
port=self.printer_port,
|
if self._dirty or self.printer is None:
|
||||||
profile=self.printer_profile,
|
self._printer = printer.Printer(
|
||||||
|
address=self.address,
|
||||||
|
port=self.port,
|
||||||
|
profile=self.profile,
|
||||||
)
|
)
|
||||||
return self.printer
|
self._dirty = False
|
||||||
|
|
||||||
|
return self._printer
|
||||||
|
|||||||
13
printer.py
13
printer.py
@@ -1,17 +1,24 @@
|
|||||||
from escpos.printer import Network
|
from escpos.printer import Network
|
||||||
|
|
||||||
|
|
||||||
class Printer:
|
class Printer:
|
||||||
def __init__(self, address: str, port: int = 9070, profile="RP-F10-80mm"):
|
def __init__(self, address: str, port: int = 9070,
|
||||||
|
profile: str = "RP-F10-80mm"):
|
||||||
self.address = address
|
self.address = address
|
||||||
self.port = port
|
self.port = port
|
||||||
self.profile = profile
|
self.profile = profile
|
||||||
self.printer = Network(host = self.address, port=self.port, profile=profile)
|
self.printer = Network(
|
||||||
|
host=self.address,
|
||||||
|
port=self.port,
|
||||||
|
profile=self.profile,
|
||||||
|
timeout=5,
|
||||||
|
)
|
||||||
|
|
||||||
def test_connection(self):
|
def test_connection(self):
|
||||||
return self.printer.is_online()
|
return self.printer.is_online()
|
||||||
|
|
||||||
def print_text(self, text: str):
|
def print_text(self, text: str):
|
||||||
# todo text properties via printer.set()
|
# todo web editor rich text to printer.set() properties
|
||||||
self.printer.text(text)
|
self.printer.text(text)
|
||||||
self.printer.ln(2)
|
self.printer.ln(2)
|
||||||
self.printer.cut()
|
self.printer.cut()
|
||||||
|
|||||||
67
server.py
67
server.py
@@ -1,4 +1,4 @@
|
|||||||
from os import environ
|
# from os import environ
|
||||||
import argparse
|
import argparse
|
||||||
from flask import Flask, request, make_response, send_from_directory, jsonify
|
from flask import Flask, request, make_response, send_from_directory, jsonify
|
||||||
from config import PrinterConfig
|
from config import PrinterConfig
|
||||||
@@ -9,6 +9,8 @@ argument_parser.add_argument("--port", type=int, required=False, default=8005)
|
|||||||
|
|
||||||
args = argument_parser.parse_args()
|
args = argument_parser.parse_args()
|
||||||
|
|
||||||
|
# todo read parameters from env
|
||||||
|
|
||||||
printer_config = PrinterConfig()
|
printer_config = PrinterConfig()
|
||||||
app = Flask(__name__)
|
app = Flask(__name__)
|
||||||
|
|
||||||
@@ -20,21 +22,22 @@ def index_route():
|
|||||||
|
|
||||||
@app.route("/config", methods=["GET", "POST"])
|
@app.route("/config", methods=["GET", "POST"])
|
||||||
def config_route():
|
def config_route():
|
||||||
request_type = None
|
json_request = request.args.get("json", False)
|
||||||
if "Accept" in request.headers:
|
print(f"{json_request=}")
|
||||||
if "text/html" in request.headers["Accept"]:
|
if not json_request and json_request != "":
|
||||||
request_type = "html"
|
json_request = False
|
||||||
elif "application/json" in request.headers["Accept"]:
|
elif json_request == "":
|
||||||
request_type = "json"
|
json_request = True
|
||||||
|
|
||||||
print(request_type)
|
if request.method == "GET" and not json_request:
|
||||||
if request.method == "GET" and request_type == "html":
|
|
||||||
# return config html page
|
# return config html page
|
||||||
return send_from_directory("static", "config.html")
|
return send_from_directory("static", "config.html")
|
||||||
elif request.method == "GET" and request_type == "json":
|
|
||||||
|
elif request.method == "GET" and json_request:
|
||||||
# render config as JSON
|
# render config as JSON
|
||||||
print(get_config())
|
print(get_config())
|
||||||
return jsonify(get_config())
|
return jsonify(get_config())
|
||||||
|
|
||||||
elif request.method == "POST" and request.is_json:
|
elif request.method == "POST" and request.is_json:
|
||||||
# accept JSON form with new parameters
|
# accept JSON form with new parameters
|
||||||
errors = {"critical": False, "errors": {}}
|
errors = {"critical": False, "errors": {}}
|
||||||
@@ -44,18 +47,21 @@ def config_route():
|
|||||||
except Exception:
|
except Exception:
|
||||||
errors["errors"]["json"] = "Could not decode message as json"
|
errors["errors"]["json"] = "Could not decode message as json"
|
||||||
errors["critical"] = True
|
errors["critical"] = True
|
||||||
|
|
||||||
request_params = {}
|
request_params = {}
|
||||||
if "address" in request_data:
|
if "address" in request_data:
|
||||||
request_params["address"] = request_data["address"]
|
request_params["address"] = request_data["address"]
|
||||||
if "port" in request_data:
|
if "port" in request_data:
|
||||||
port = request_data["port"]
|
port = request_data["port"]
|
||||||
|
# todo remove redundant checks
|
||||||
if isinstance(port, int) and 0 < port <= 65535:
|
if isinstance(port, int) and 0 < port <= 65535:
|
||||||
request_params["port"] = port
|
request_params["port"] = port
|
||||||
else:
|
else:
|
||||||
errors["errors"]["parseint"] = "Could not parse port number as an integer"
|
errors["errors"][
|
||||||
|
"parseint"
|
||||||
|
] = "Could not parse port number as an integer"
|
||||||
errors["critical"] = True
|
errors["critical"] = True
|
||||||
|
|
||||||
response = jsonify(errors)
|
response = jsonify(errors)
|
||||||
if errors["critical"]:
|
if errors["critical"]:
|
||||||
response.status_code = 422
|
response.status_code = 422
|
||||||
@@ -67,36 +73,43 @@ def config_route():
|
|||||||
|
|
||||||
def get_config():
|
def get_config():
|
||||||
return {
|
return {
|
||||||
"address": printer_config.printer_address,
|
"address": printer_config.address,
|
||||||
"port": printer_config.printer_port,
|
"port": printer_config.port,
|
||||||
"profile": printer_config.printer_profile,
|
"profile": printer_config.profile,
|
||||||
"can_print": can_print(),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def set_config(address: str = None, port: int = None):
|
def set_config(address: str = None, port: int = None):
|
||||||
if address is None:
|
printer_config.address = address
|
||||||
printer_config.printer_address = None
|
printer_config.port = port
|
||||||
else:
|
# printer_config.profile = profile # fixed default in config.py
|
||||||
printer_config.printer_address = address
|
printer_config.save()
|
||||||
if port is None:
|
|
||||||
printer_config.printer_port = 9100
|
|
||||||
else:
|
|
||||||
printer_config.printer_port = port
|
|
||||||
|
|
||||||
|
|
||||||
def can_print():
|
def can_print():
|
||||||
return printer_config.complete
|
return printer_config.complete
|
||||||
|
|
||||||
|
|
||||||
|
@app.route("/status", methods=["GET"])
|
||||||
|
def printer_status_route():
|
||||||
|
return jsonify(
|
||||||
|
{
|
||||||
|
"configured": can_print(),
|
||||||
|
"connected": printer_config.printer.test_connection(),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
try:
|
try:
|
||||||
printer_config.load()
|
printer_config.load()
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
pass
|
printer_config.address = "localhost"
|
||||||
|
printer_config.port = 9100
|
||||||
|
# printer_config.profile = ""
|
||||||
|
|
||||||
try:
|
try:
|
||||||
app.run(host=args.address, port=args.port)
|
app.run(host=args.address, port=args.port, threaded=False)
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
async function get_config() {
|
async function get_config() {
|
||||||
response = await fetch("/config", {method: "GET", headers: {"Accept": "application/json"}})
|
response = await fetch("/config?json", {method: "GET", headers: {"Accept": "application/json"}})
|
||||||
if (! response.ok){
|
if (! response.ok){
|
||||||
throw Error("config data request error: " + await response.text())
|
throw Error("config data request error: " + await response.text())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -16,6 +16,14 @@
|
|||||||
</div>
|
</div>
|
||||||
|
|
||||||
<div class="container" display="block">
|
<div class="container" display="block">
|
||||||
|
<div class="status-block">
|
||||||
|
<div class="status-block configuration">
|
||||||
|
<h2>Configuration:</h2>
|
||||||
|
</div>
|
||||||
|
<div class="status-block connection">
|
||||||
|
<h2>Connection:</h2>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
</div>
|
</div>
|
||||||
</body>
|
</body>
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user