Lint project
This commit is contained in:
227
src/main.py
227
src/main.py
@@ -1,46 +1,72 @@
|
|||||||
# Welcome to the LittlePrynter's source code.
|
"""
|
||||||
# This program expose a web interface, with user authentification, that makes it possible to print messages from the web.
|
Welcome to the LittlePrynter's source code.
|
||||||
# It also exposes a API, making it possible to print and interface with much of the printer's abilities.
|
This program expose a web interface, with user authentification,
|
||||||
|
that makes it possible to print messages from the web.
|
||||||
|
It also exposes a API, making it possible to print and interface
|
||||||
|
with much of the printer's abilities.
|
||||||
|
|
||||||
# We first define the connection to the printer itself,
|
We first define the connection to the printer itself,
|
||||||
# Then we build the API around Flask,
|
Then we build the API around Flask,
|
||||||
# Then we build the web interface, using the simple Jinja2 templating.
|
Then we build the web interface, using the simple Jinja2 templating.
|
||||||
|
|
||||||
# We support two modes :
|
We support two modes :
|
||||||
# The first is a simple mode, where a computer, connected to a thermal printer, runs this program and exposes a web interface that makes use of the client's camera
|
The first is a simple mode, where a computer, connected to a thermal printer,
|
||||||
# The seconde is booth mode, where a Raspberry Pi is connected to a thermal printer, a button and a flash. The web interface exists but may not be used, as the press of the button with take a picture and activate the flash while simply informing the web page.
|
runs this program and exposes a web interface that makes use of the client's camera
|
||||||
|
The seconde is booth mode, where a Raspberry Pi is connected to a thermal printer,
|
||||||
|
a button and a flash.
|
||||||
|
The web interface exists but may not be used, as the press of the button with take
|
||||||
|
a picture and activate the flash while simply informing the web page.
|
||||||
|
"""
|
||||||
|
|
||||||
# Following are the librairies we import,
|
# Following are the librairies we import,
|
||||||
from flask import Flask, request, render_template, flash, abort, redirect, url_for, make_response, jsonify # Used for the web framework
|
import sys
|
||||||
|
import os # For VARS from the shell.
|
||||||
|
import pprint # To pretty print JSON
|
||||||
|
import toml # Used for the config file parsing
|
||||||
|
from flask import (
|
||||||
|
Flask,
|
||||||
|
request,
|
||||||
|
render_template,
|
||||||
|
flash,
|
||||||
|
redirect,
|
||||||
|
url_for,
|
||||||
|
jsonify,
|
||||||
|
) # Used for the web framework
|
||||||
|
import werkzeug.exceptions
|
||||||
from flask_socketio import SocketIO
|
from flask_socketio import SocketIO
|
||||||
from flask_limiter import Limiter
|
from flask_limiter import Limiter
|
||||||
from flask_limiter.util import get_remote_address
|
from flask_limiter.util import get_remote_address
|
||||||
from printer import Printer # The wrapper for the printer class
|
from printer import Printer # The wrapper for the printer class
|
||||||
from raspberry import Raspberry # The Raspberry pi control Class
|
from raspberry import Raspberry # The Raspberry pi control Class
|
||||||
from web import Web # Wrapper for the web routes and API
|
from web import Web # Wrapper for the web routes and API
|
||||||
import toml # Used for the config file parsing
|
|
||||||
import pprint # To pretty print JSON
|
|
||||||
import time # To sleep
|
|
||||||
import os # For VARS from the shell.
|
|
||||||
# Variables
|
# Variables
|
||||||
|
|
||||||
app = Flask(__name__)
|
app = Flask(__name__)
|
||||||
socketio = SocketIO(app)
|
socketio = SocketIO(app)
|
||||||
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
|
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif"}
|
||||||
|
|
||||||
# Load the configuration file
|
# Load the configuration file
|
||||||
try:
|
try:
|
||||||
app.logger.debug("Loading config file...")
|
app.logger.debug("Loading config file...")
|
||||||
configuration_file = toml.load("configuration/config.toml")
|
with open("configuration/config.toml", "r", encoding="utf-8") as f:
|
||||||
|
configuration_file = toml.load(f)
|
||||||
except TypeError:
|
except TypeError:
|
||||||
app.logger.error("Unable to load the config file: invalid type or is a list containing invalid types")
|
app.logger.error(
|
||||||
exit(-1)
|
"Unable to load the config file: invalid type or is a list containing invalid types"
|
||||||
except toml.TomlDecodeError:
|
)
|
||||||
app.logger.error("An error occured while decoding the file")
|
sys.exit(-1)
|
||||||
exit(-1)
|
except toml.TomlDecodeError as e:
|
||||||
except Exception as e:
|
app.logger.error(
|
||||||
app.logger.error("Error while loading file : " + str(e))
|
"An error occured while decoding the file %s , error at %s:%s",
|
||||||
exit(-1)
|
str(e.doc),
|
||||||
|
str(e.colno),
|
||||||
|
str(e.lineno),
|
||||||
|
)
|
||||||
|
sys.exit(-1)
|
||||||
|
except OSError as e:
|
||||||
|
app.logger.error("Error while loading file %s ", str(e))
|
||||||
|
sys.exit(-1)
|
||||||
|
|
||||||
app.logger.debug("Config file loaded !")
|
app.logger.debug("Config file loaded !")
|
||||||
|
|
||||||
@@ -51,69 +77,69 @@ UPLOAD_FOLDER = str(configuration_file["printer"]["upload_folder"])
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
os.mkdir(UPLOAD_FOLDER)
|
os.mkdir(UPLOAD_FOLDER)
|
||||||
app.logger.debug(f"Directory '{UPLOAD_FOLDER}' created successfully.")
|
app.logger.debug("Directory %s created successfully.", UPLOAD_FOLDER)
|
||||||
except FileExistsError:
|
except FileExistsError:
|
||||||
app.logger.debug(f"Directory '{UPLOAD_FOLDER}' already exists.")
|
app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
|
||||||
except PermissionError:
|
except PermissionError:
|
||||||
app.logger.error(f"Permission denied: Unable to create '{UPLOAD_FOLDER}'.")
|
app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
|
||||||
except Exception as e:
|
|
||||||
app.logger.error(f"An error occurred: {e}")
|
|
||||||
|
|
||||||
|
|
||||||
# Output the config file
|
# Output the config file
|
||||||
if os.getenv('LIPY_DEBUG') == True:
|
if os.getenv("LIPY_DEBUG") is True:
|
||||||
pprint.pprint(configuration_file)
|
pprint.pprint(configuration_file)
|
||||||
|
|
||||||
# We define the app module used by Flask
|
# We define the app module used by Flask
|
||||||
app.secret_key = configuration_file["secrets"]["flask_secret_key"]
|
app.secret_key = configuration_file["secrets"]["flask_secret_key"]
|
||||||
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
|
app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
|
||||||
app.config['ALLOWED_EXTENSIONS'] = ALLOWED_EXTENSIONS
|
app.config["ALLOWED_EXTENSIONS"] = ALLOWED_EXTENSIONS
|
||||||
app.config['MAX_CONTENT_LENGTH'] = 10 * 1000 * 1000 # Maximum 3Mb for a file upload
|
app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000 # Maximum 3Mb for a file upload
|
||||||
app.config['TEMPLATES_AUTO_RELOAD'] = True
|
app.config["TEMPLATES_AUTO_RELOAD"] = True
|
||||||
|
|
||||||
# Printer connection
|
# Printer connection
|
||||||
# Uses the class defined in the printer.py file
|
# Uses the class defined in the printer.py file
|
||||||
printer = Printer(app,0x04b8, 0x0e28)
|
printer = Printer(app, 0x04B8, 0x0E28)
|
||||||
printer.init_printer()
|
printer.init_printer()
|
||||||
|
|
||||||
# Find out if we are running on a Raspberry Pi
|
# Find out if we are running on a Raspberry Pi
|
||||||
rpi = Raspberry(printer,
|
rpi = Raspberry(
|
||||||
|
printer,
|
||||||
app,
|
app,
|
||||||
socketio,
|
socketio,
|
||||||
configuration_file['rpi']['button_gpio_port_number'], configuration_file['rpi']['indicator_gpio_port_number'],
|
configuration_file["rpi"]["button_gpio_port_number"],
|
||||||
configuration_file['rpi']['flash_gpio_port_number'],
|
configuration_file["rpi"]["indicator_gpio_port_number"],
|
||||||
configuration_file['rpi']['flash'] )
|
configuration_file["rpi"]["flash_gpio_port_number"],
|
||||||
|
configuration_file["rpi"]["flash"],
|
||||||
|
)
|
||||||
|
|
||||||
RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
|
RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
|
||||||
|
|
||||||
|
|
||||||
#############################################################
|
|
||||||
# Web & API routes
|
# Web & API routes
|
||||||
#############################################################
|
|
||||||
|
|
||||||
|
|
||||||
web = Web(app, printer)
|
web = Web(app, printer)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
app.run(debug=True, use_reloader=False, host='0.0.0.0', ssl_context='adhoc')
|
app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
|
||||||
|
|
||||||
limiter = Limiter(
|
limiter = Limiter(
|
||||||
get_remote_address,
|
get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
|
||||||
app=app,
|
|
||||||
default_limits=["1500 per day", "500 per hour"]
|
|
||||||
)
|
)
|
||||||
|
|
||||||
@app.route('/')
|
|
||||||
|
@app.route("/")
|
||||||
@limiter.limit("1/second", override_defaults=False)
|
@limiter.limit("1/second", override_defaults=False)
|
||||||
def index():
|
def index():
|
||||||
|
"""Return the web interface index"""
|
||||||
app.logger.debug("Loading index")
|
app.logger.debug("Loading index")
|
||||||
return render_template('index.html')
|
return render_template("index.html")
|
||||||
|
|
||||||
@app.route('/webcam')
|
|
||||||
|
@app.route("/webcam")
|
||||||
@limiter.limit("1/second", override_defaults=False)
|
@limiter.limit("1/second", override_defaults=False)
|
||||||
def webcam():
|
def webcam():
|
||||||
|
"""Returns the webcam web interface"""
|
||||||
app.logger.debug("Loading webcam interface")
|
app.logger.debug("Loading webcam interface")
|
||||||
return render_template('webcam.html')
|
return render_template("webcam.html")
|
||||||
|
|
||||||
|
|
||||||
# API routes
|
# API routes
|
||||||
# The api has the following methods
|
# The api has the following methods
|
||||||
@@ -121,71 +147,71 @@ def webcam():
|
|||||||
# api/auth/{login,logout}
|
# api/auth/{login,logout}
|
||||||
# api/status/{paper,ping,stats}
|
# api/status/{paper,ping,stats}
|
||||||
|
|
||||||
|
|
||||||
# If you just call the api route, you get a help back.
|
# If you just call the api route, you get a help back.
|
||||||
@app.route('/api')
|
@app.route("/api")
|
||||||
@app.route('/api/print')
|
@app.route("/api/print")
|
||||||
@limiter.limit("1/second", override_defaults=False)
|
@limiter.limit("1/second", override_defaults=False)
|
||||||
def api_index():
|
def api_index():
|
||||||
|
"""Returns a how-to for the API"""
|
||||||
app.logger.debug("Loading API")
|
app.logger.debug("Loading API")
|
||||||
return render_template("api.html")
|
return render_template("api.html")
|
||||||
|
|
||||||
|
|
||||||
@app.route('/api/print/sms', methods=['POST'])
|
@app.route("/api/print/sms", methods=["POST"])
|
||||||
@limiter.limit("6/minute", override_defaults=False)
|
@limiter.limit("6/minute", override_defaults=False)
|
||||||
def api_print_sms():
|
def api_print_sms():
|
||||||
|
"""Prints a short message on a printer"""
|
||||||
app.logger.debug("Printing an sms")
|
app.logger.debug("Printing an sms")
|
||||||
try:
|
try:
|
||||||
txt = request.form["txt"]
|
txt = request.form["txt"]
|
||||||
sign = request.form["signature"]
|
sign = request.form["signature"]
|
||||||
except Exception as e:
|
except werkzeug.exceptions.BadRequestKeyError as e:
|
||||||
app.logger.error("Whoops, no forms submitted or missing signature :" + str(e))
|
app.logger.error(
|
||||||
flash("Whoops, no forms submitted or missing signature : " + str(e))
|
"Whoops, no forms submitted or missing signature : %s ", str(e)
|
||||||
|
)
|
||||||
|
flash("Whoops, no forms submitted or missing signature : %s", str(e))
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
web.print_sms(txt, sign)
|
web.print_sms(txt, sign)
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
@app.route('/api/print/img', methods=['POST'])
|
|
||||||
|
@app.route("/api/print/img", methods=["POST"])
|
||||||
@limiter.limit("6/minute", override_defaults=False)
|
@limiter.limit("6/minute", override_defaults=False)
|
||||||
def api_print_image():
|
def api_print_image():
|
||||||
|
"""Prints an image on a printer"""
|
||||||
app.logger.debug("Printing an image")
|
app.logger.debug("Printing an image")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
sign = request.form["signature"]
|
sign = request.form["signature"]
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
app.logger.error("Whoops, no forms submitted or missing signature :" + str(e))
|
app.logger.error("Whoops, no forms submitted or missing signature : %s", str(e))
|
||||||
flash("Whoops, no forms submitted or missing signature : " + str(e))
|
flash("Whoops, no forms submitted or missing signature : %s ", str(e))
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
|
if request.method == "POST":
|
||||||
|
|
||||||
if request.method == 'POST':
|
|
||||||
# check if the post request has the file part
|
# check if the post request has the file part
|
||||||
try:
|
if "img" not in request.files:
|
||||||
if 'img' not in request.files:
|
app.logger.error("Whoops, no images submitted : %s ", str(e))
|
||||||
app.logger.error("Whoops, no images submitted :" + str(e))
|
app.logger.error("Error getting the files : %s", str(e))
|
||||||
flash("Whoops, no images submitted : " + str(e))
|
flash("Whoops, no images submitted : %s", str(e))
|
||||||
else:
|
|
||||||
file = request.files['img']
|
|
||||||
except Exception as e:
|
|
||||||
app.logger.error('Error getting the files :' + str(e))
|
|
||||||
flash('Error getting the files :' + str(e))
|
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
|
file = request.files["img"]
|
||||||
# If the user does not select a file, the browser submits an
|
# If the user does not select a file, the browser submits an
|
||||||
# empty file without a filename.
|
# empty file without a filename.
|
||||||
if file.filename == '':
|
if file.filename == "":
|
||||||
app.logger.error("Submitted file has no filename !")
|
app.logger.error("Submitted file has no filename !")
|
||||||
flash("Submitted file has no filename !")
|
flash("Submitted file has no filename !")
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
app.logger.debug("Sending the image to the printer.")
|
app.logger.debug("Sending the image to the printer.")
|
||||||
web.print_image(file, sign)
|
web.print_image(file, sign)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
app.logger.error("The image could not be printed because : " + str(e) )
|
app.logger.error("The image could not be printed because : %s ", str(e))
|
||||||
flash("The image could not be printed because : " + str(e))
|
flash("The image could not be printed because : %s ", str(e))
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
else:
|
else:
|
||||||
@@ -193,56 +219,69 @@ def api_print_image():
|
|||||||
flash("Method not allowed")
|
flash("Method not allowed")
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
flash('Picture printed ! '),
|
flash("Picture printed ! ")
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
|
|
||||||
@app.route('/api/camera/picture')
|
@app.route("/api/camera/picture")
|
||||||
def camera_picture():
|
def camera_picture():
|
||||||
# Returns a picture taken by the camera
|
"""Returns a picture taken by the camera"""
|
||||||
if RASPBERRY_PI_CONNECTED:
|
if RASPBERRY_PI_CONNECTED:
|
||||||
try:
|
try:
|
||||||
return rpi.camera_picture()
|
return rpi.camera_picture()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
return jsonify({'message': 'Error getting the stream : ' + e}), 500
|
return jsonify({"message": "Error getting the stream : " + e}), 500
|
||||||
else:
|
else:
|
||||||
return jsonify({'message': 'No camera present'}), 500
|
return jsonify({"message": "No camera present"}), 500
|
||||||
|
|
||||||
|
|
||||||
|
@app.route("/login")
|
||||||
@app.route('/login')
|
|
||||||
@limiter.limit("1/second", override_defaults=False)
|
@limiter.limit("1/second", override_defaults=False)
|
||||||
def login_page():
|
def login_page():
|
||||||
|
"""Unsued, logins"""
|
||||||
# web.login(username,password)
|
# web.login(username,password)
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
@app.route('/logout')
|
|
||||||
|
@app.route("/logout")
|
||||||
@limiter.limit("1/second", override_defaults=False)
|
@limiter.limit("1/second", override_defaults=False)
|
||||||
def logout_page():
|
def logout_page():
|
||||||
|
"""Unused, logout"""
|
||||||
# web.logout(username, password)
|
# web.logout(username, password)
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
|
|
||||||
@app.errorhandler(429)
|
@app.errorhandler(429)
|
||||||
def ratelimit_handler(e):
|
def ratelimit_handler(e):
|
||||||
flash("Rate limit reached, please slow down :) ( Currently at "+ e.description + ")", 'error')
|
"""Handle rate limits"""
|
||||||
app.logger.debug('Rate limit reached ' + str(e.description))
|
flash(
|
||||||
|
"Rate limit reached, please slow down :) ( Currently at " + e.description + ")",
|
||||||
|
"error",
|
||||||
|
)
|
||||||
|
app.logger.debug("Rate limit reached %s " , str(e.description))
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
|
|
||||||
@app.route("/ping")
|
@app.route("/ping")
|
||||||
@limiter.exempt
|
@limiter.exempt
|
||||||
def ping():
|
def ping():
|
||||||
flash("🏓 Pong !",'info')
|
"""Returns a pong"""
|
||||||
app.logger.debug('🏓 Pong !')
|
flash("🏓 Pong !", "info")
|
||||||
|
app.logger.debug("🏓 Pong !")
|
||||||
return redirect(url_for("index"))
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
@socketio.on('ping')
|
|
||||||
def handle_message(data):
|
|
||||||
app.logger.debug('Received : ' + str(data))
|
|
||||||
socketio.emit('pong',"Pong !")
|
|
||||||
|
|
||||||
@socketio.on('get_camera_status')
|
@socketio.on("ping")
|
||||||
|
def handle_message(data):
|
||||||
|
"""Handle sockets pings"""
|
||||||
|
app.logger.debug("Received : %s " , str(data))
|
||||||
|
socketio.emit("pong", "Pong !")
|
||||||
|
|
||||||
|
|
||||||
|
@socketio.on("get_camera_status")
|
||||||
def camera_status():
|
def camera_status():
|
||||||
app.logger.debug('Client asked if we had a camera')
|
"""Returns camera status to a socket"""
|
||||||
|
app.logger.debug("Client asked if we had a camera")
|
||||||
if RASPBERRY_PI_CONNECTED:
|
if RASPBERRY_PI_CONNECTED:
|
||||||
socketio.emit("camera_status", True)
|
socketio.emit("camera_status", True)
|
||||||
else:
|
else:
|
||||||
|
|||||||
Reference in New Issue
Block a user