littleprynter/src/main.py

"""
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.
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,
Then we build the API around Flask,
Then we build the web interface, using the simple Jinja2 templating.

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 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,
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_limiter import Limiter
from flask_limiter.util import get_remote_address
from printer import Printer  # The wrapper for the printer class
from raspberry import Raspberry  # The Raspberry pi control Class
from web import Web  # Wrapper for the web routes and API

# Variables

app = Flask(__name__)
socketio = SocketIO(app)
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}

# Load the configuration file
try:
    app.logger.debug("Loading config file...")
    with open("configuration/config.toml", "r", encoding="utf-8") as f:
        configuration_file = toml.load(f)
except TypeError:
    app.logger.error(
        "Unable to load the config file: invalid type or is a list containing invalid types"
    )
    sys.exit(-1)
except toml.TomlDecodeError as e:
    app.logger.error(
        "An error occured while decoding the file %s ,  error at %s:%s",
        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 !")

# Define the USB connections here.
vendor_id = configuration_file["printer"]["vendor_id"]
device_id = configuration_file["printer"]["device_id"]
UPLOAD_FOLDER = str(configuration_file["printer"]["upload_folder"])

try:
    os.mkdir(UPLOAD_FOLDER)
    app.logger.debug("Directory %s created successfully.", UPLOAD_FOLDER)
except FileExistsError:
    app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
except PermissionError:
    app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
    exit(77)

# Output the config file
if os.getenv("LIPY_DEBUG") is True:
    pprint.pprint(configuration_file)

# We define the app module used by Flask
app.secret_key = configuration_file["secrets"]["flask_secret_key"]
app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
app.config["ALLOWED_EXTENSIONS"] = ALLOWED_EXTENSIONS
app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000  # Maximum 3Mb for a file upload
app.config["TEMPLATES_AUTO_RELOAD"] = True

# Printer connection
# Uses the class defined in the printer.py file
printer = Printer(app, 0x04B8, 0x0E28)
printer.init_printer()

# Find out if we are running on a Raspberry Pi
rpi = Raspberry(
    printer,
    app,
    socketio,
    configuration_file["rpi"]["button_gpio_port_number"],
    configuration_file["rpi"]["indicator_gpio_port_number"],
    configuration_file["rpi"]["flash_gpio_port_number"],
    configuration_file["rpi"]["flash"],
)

RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()


#  Web & API  routes

web = Web(app, printer)

if __name__ == "__main__":
    app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

limiter = Limiter(
    get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
)


@app.route("/")
@limiter.limit("1/second", override_defaults=False)
def index():
    """Return the web interface index"""
    app.logger.debug("Loading index")
    return render_template("index.html")


@app.route("/webcam")
@limiter.limit("1/second", override_defaults=False)
def webcam():
    """Returns the webcam web interface"""
    app.logger.debug("Loading webcam interface")
    return render_template("webcam.html")


# API routes
# The api has the following methods
# api/print/{sms,img,letter,qr,barcode}
# api/auth/{login,logout}
# api/status/{paper,ping,stats}


# If you just call the api route, you get a help back.
@app.route("/api")
@app.route("/api/print")
@limiter.limit("1/second", override_defaults=False)
def api_index():
    """Returns a how-to for the API"""
    app.logger.debug("Loading API")
    return render_template("api.html")


@app.route("/api/print/sms", methods=["POST"])
@limiter.limit("6/minute", override_defaults=False)
def api_print_sms():
    """Prints a short message on a printer"""
    app.logger.debug("Printing an sms")
    try:
        txt = request.form["txt"]
    except werkzeug.exceptions.BadRequestKeyError as e:
        app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e))
        flash("Whoops, no forms submitted or missing signature : %s", str(e))
        return redirect(url_for("index"))

    try:
        # comment: We try to get a signature
        sign = request.form["signature"]
    except werkzeug.exceptions.BadRequestKeyError as e:
        app.logger.warning(
            "No signature found for this print, using default signature.", str(e)
        )
        sign = configuration_file["defaults"]["signature"]

    web.print_sms(txt, sign)
    return redirect(url_for("index"))


@app.route("/api/print/img", methods=["POST"])
@limiter.limit("6/minute", override_defaults=False)
def api_print_image():
    """Prints an image on a printer"""
    app.logger.debug("Printing an image")

    try:
        # comment: We try to get a signature
        sign = request.form["signature"]
    except werkzeug.exceptions.BadRequestKeyError as e:
        app.logger.warning(
            "No signature found for this print, using default signature.", str(e)
        )
        sign = configuration_file["defaults"]["signature"]

    if request.method == "POST":
        # check if the post request has the file part
        if "img" not in request.files:
            app.logger.error("Whoops, no images submitted  : %s ", str(e))
            app.logger.error("Error getting the files : %s", str(e))
            flash("Whoops, no images submitted  : %s", str(e))
            return redirect(url_for("index"))

        file = request.files["img"]
        # If the user does not select a file, the browser submits an
        # empty file without a filename.
        if file.filename == "":
            app.logger.error("Submitted file has no filename !")
            flash("Submitted file has no filename !")
            return redirect(url_for("index"))

        try:
            app.logger.debug("Sending the image to the printer.")
            web.print_image(file, sign)
        except Exception as e:
            app.logger.error("The image could not be printed because : %s ", str(e))
            flash("The image could not be printed because : %s ", str(e))
            return redirect(url_for("index"))

    else:
        app.logger.error("Method not allowed")
        flash("Method not allowed")
        return redirect(url_for("index"))

    flash("Picture printed ! ")
    return redirect(url_for("index"))


@app.route("/api/camera/picture")
def camera_picture():
    """Returns a picture taken by the camera"""
    if RASPBERRY_PI_CONNECTED:
        try:
            return rpi.camera_picture()
        except Exception as e:
            return jsonify({"message": "Error getting the stream : " + e}), 500
    else:
        return jsonify({"message": "No camera present"}), 500


@app.route("/login")
@limiter.limit("1/second", override_defaults=False)
def login_page():
    """Unsued, logins"""
    # web.login(username,password)
    return redirect(url_for("index"))


@app.route("/logout")
@limiter.limit("1/second", override_defaults=False)
def logout_page():
    """Unused, logout"""
    # web.logout(username, password)
    return redirect(url_for("index"))


@app.errorhandler(429)
def ratelimit_handler(e):
    """Handle rate limits"""
    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"))


@app.route("/ping")
@limiter.exempt
def ping():
    """Returns a pong"""
    flash("🏓 Pong !", "info")
    app.logger.debug("🏓 Pong !")
    return redirect(url_for("index"))


@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():
    """Returns camera status to a socket"""
    app.logger.debug("Client asked if we had a camera")
    if RASPBERRY_PI_CONNECTED:
        socketio.emit("camera_status", True)
    else:
        socketio.emit("camera_status", False)