diff --git a/requirements.txt b/requirements.txt index fe5e0f2..3c2534d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,36 +1,36 @@ -Adafruit_Thermal==1.1.0 -appdirs==1.4.4 -argcomplete==2.0.6 -cffi==1.17.1 -click==8.1.8 -commonmark==0.9.1 -cryptography==45.0.4 -Deprecated==1.2.18 -escpos==2.0.0 -Flask==2.1.3 -Flask-Limiter==2.4.5.1 -future==0.18.3 -itsdangerous==2.1.2 -Jinja2==3.1.6 -limits==2.6.3 -MarkupSafe==2.1.5 -numpy==2.3.0 -packaging==21.3 -pillow==11.2.1 -pycparser==2.22 -Pygments==2.12.0 -pyparsing==3.0.9 -pyserial==3.5 -python-barcode==0.13.1 -pyusb==1.2.1 -PyYAML==6.0.2 -qrcode==7.3.1 -rich==12.4.4 -setuptools==80.9.0 -six==1.16.0 -toml==0.10.2 -typing_extensions==4.2.0 -Unidecode==1.3.8 -viivakoodi==0.8.0 -Werkzeug==2.1.2 -wrapt==1.14.1 +Adafruit_Thermal>=1.1.0 +appdirs>=1.4.4 +argcomplete>=2.0.6 +cffi>=1.17.1 +click>=8.1.8 +commonmark>=0.9.1 +cryptography>=45.0.4 +Deprecated>=1.2.18 +escpos>=2.0.0 +Flask>=2.1.3 +Flask-Limiter>=2.4.5.1 +future>=0.18.3 +itsdangerous>=2.1.2 +Jinja2>=3.1.6 +limits>=2.6.3 +MarkupSafe>=2.1.5 +numpy>=2.3.0 +packaging>=21.3 +pillow>=11.2.1 +pycparser>=2.22 +Pygments>=2.12.0 +pyparsing>=3.0.9 +pyserial>=3.5 +python-barcode>=0.13.1 +pyusb>=1.2.1 +PyYAML>=6.0.2 +qrcode>=7.3.1 +rich>=12.4.4 +setuptools>=80.9.0 +six>=1.16.0 +toml>=0.10.2 +typing_extensions>=4.2.0 +Unidecode>=1.3.8 +viivakoodi>=0.8.0 +Werkzeug>=2.1.2 +wrapt>=1.14.1 diff --git a/src/main.py b/src/main.py index 8658e4b..2127f0c 100644 --- a/src/main.py +++ b/src/main.py @@ -6,11 +6,17 @@ # 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, from flask import Flask, request, render_template, flash, abort, redirect, url_for, make_response, jsonify # Used for the web framework +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 import toml # Used for the config file parsing import pprint # To pretty print JSON @@ -19,6 +25,7 @@ import os # For VARS from the shell. # Variables app = Flask(__name__) +socketio = SocketIO(app) ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'} # Load the configuration file @@ -36,6 +43,7 @@ except Exception as e: 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"] @@ -68,18 +76,28 @@ app.config['TEMPLATES_AUTO_RELOAD'] = True printer = Printer(app,0x04b8, 0x0e28) printer.init_printer() -# Web routes +# 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'] ) +RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi() +if RASPBERRY_PI_CONNECTED: + rpi.initialise_gpio() + + +############################################################# +# Web & API routes +############################################################# + + web = Web(app, printer) if __name__ == "__main__": app.run(ssl_context='adhoc') - limiter = Limiter( - app, - key_func=get_remote_address, + get_remote_address, + app=app, default_limits=["1500 per day", "500 per hour"] -) + ) @app.route('/') @limiter.limit("1/second", override_defaults=False) @@ -116,16 +134,12 @@ def api_print_sms(): txt = request.form["txt"] sign = request.form["signature"] except Exception as e: - app.logger.error(str(e) + " - Whoops, no forms submitted or missing signature.") - return jsonify({'message': 'Error getting the information from the form :' + e}), 500 + app.logger.error("Whoops, no forms submitted or missing signature :" + str(e)) + flash("Whoops, no forms submitted or missing signature : " + str(e)) + return redirect(url_for("index")) - try: - web.print_sms(txt,sign) - except Exception as e: - return jsonify({'message': 'Error printing the SMS:' + e}), 500 - - - return jsonify({'message': 'Message printed'}), 200 + web.print_sms(txt,sign) + return redirect(url_for("index")) @app.route('/api/print/img', methods=['POST']) @limiter.limit("6/minute", override_defaults=False) @@ -135,40 +149,61 @@ def api_print_image(): try: sign = request.form["signature"] except Exception as e: - app.logger.error(str(e) + " - Whoops, no forms submitted or missing signature.") - return jsonify({'message': 'Error getting the information from the form :' + e}), 500 + app.logger.error("Whoops, no forms submitted or missing signature :" + str(e)) + flash("Whoops, no forms submitted or missing signature : " + str(e)) + return redirect(url_for("index")) + if request.method == 'POST': # check if the post request has the file part try: if 'img' not in request.files: - app.logger.error("No file found. Did you use the good form ?") - return jsonify({'message': 'No file found. Did you use the good form ?'}), 500 + app.logger.error("Whoops, no images submitted :" + str(e)) + flash("Whoops, no images submitted : " + str(e)) else: file = request.files['img'] except Exception as e: - app.logger.error('Error getting the files :' + e) - return jsonify({'message': 'Error getting the files :' + e}), 500 + app.logger.error('Error getting the files :' + str(e)) + flash('Error getting the files :' + str(e)) + return redirect(url_for("index")) # 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 !") - return jsonify({'message': "Submitted file has no filename !" + e}), 500 + 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 : " + e ) - return jsonify({'message': "The image could not be printed :" + e}), 500 + app.logger.error("The image could not be printed because : " + str(e) ) + flash("The image could not be printed because : " + str(e)) + return redirect(url_for("index")) else: - return jsonify({'message': "Method Not Allowed, please POST"}), 403 - app.logger.debug('Bad access type to this API, please POST') + 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 - return jsonify({'message': 'Message printed'}), 200 @app.route('/login') @@ -195,3 +230,16 @@ def ping(): flash("🏓 Pong !",'info') app.logger.debug('🏓 Pong !') 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') +def camera_status(): + 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) diff --git a/src/printer.py b/src/printer.py index 77e4faa..5dc40e8 100644 --- a/src/printer.py +++ b/src/printer.py @@ -48,16 +48,14 @@ class Printer(object): self.app.logger.error('Printer has no more paper, aborting...') flash("No more paper on the printer. Sorry.",category='error') self.printer.close() - return False + raise Exception("No more paper in the printer") case 1: self.app.logger.warning('Printer needs paper to be changed very soon ! ') flash('Printer needs paper to be changed very soon ! ', category='info') self.printer.close() - return True case 2: self.app.logger.debug('Printer has paper, good to go') self.printer.close() - return True def init_printer(self): @@ -99,57 +97,55 @@ class Printer(object): self.ready = True; self.printer.close(); - if not self.check_paper(): - return False + self.check_paper() return True - def print_sms(self, msg, signature) -> bool: + def print_sms(self, msg, signature) -> None: clean_msg = str(msg) clean_signature = str(signature) - if not self.check_paper(): - return False - if len(clean_msg) > 256 or len(clean_msg) < 3 : self.app.logger.warning("Could not print message of this length: " + str(len(clean_msg))) - flash("Could not print message of this length :" + str(len(clean_msg)) + ", needs to between 3 and 256 caracters long.",category='error') - return False + raise Exception("Could not print message of this length :" + str(len(clean_msg)) + ", needs to between 3 and 256 caracters long.") if len(signature) > 256 or len(signature) < 3: - self.app.logger.warning("Could not print message without a signature.") - flash("Could not print message without a signature.",category='error') - return False + self.app.logger.warning("Could not print signature of this length: " + str(len(clean_signature))) + raise Exception("Could not print signature of this length :" + str(len(clean_signature)) + ", needs to between 3 and 256 caracters long.") - if not os.getenv('LIPY_DEBUG') == True: - try: - self.printer.open(self.usb_args); - self.printer.set(align='left', font='a', bold=False, underline=0, width=1, height=1, density=8, invert=False, smooth=True, flip=False, double_width=False, double_height=False, custom_size=False) - self.printer.textln("> Printed by LittlePrinter | https://n07070.xyz/articles/littleprynter") - self.printer.textln(clean_msg) - self.printer.set(align='left', font='b', bold=False, underline=1, width=1, height=1, density=9, invert=False, smooth=True, flip=False, double_width=False, double_height=False, custom_size=False) - self.printer.textln("> " + clean_signature + " @ " + strftime("%Y-%m-%d %H:%M:%S", gmtime())) - self.printer.cut() - self.printer.close - except Exception as e: - flash("Unable to print because : " + e) + self.check_paper() + + try: + self.printer.open(self.usb_args); + self.printer.set(align='left', font='a', bold=False, underline=0, width=1, height=1, density=8, invert=False, smooth=True, flip=False, double_width=False, double_height=False, custom_size=False) + self.printer.set(align='left', font='a', bold=True, underline=1, width=1, height=1, density=6, invert=False, smooth=True, flip=False, double_width=False, double_height=False, custom_size=False) + self.printer.textln(clean_msg) + self.printer.textln("") + self.printer.set(align='left', font='b', bold=False, underline=1, width=1, height=1, density=9, invert=False, smooth=True, flip=False, double_width=False, double_height=False, custom_size=False) + self.printer.textln("> " + clean_signature + " @ " + strftime("%Y-%m-%d %H:%M:%S", gmtime())) + self.printer.textln("") + self.printer.textln("Printed by LittlePrinter ") + self.printer.textln("n07070.xyz/articles/littleprynter") + self.printer.textln("") + self.printer.cut() + self.printer.close() + except Exception as e: + self.app.logger.error("Unable to print because : " + str(e)) + raise e flash("Message printed : " + clean_msg ,category='info') return True - def print_img(self, path, sign): + def print_img(self, path, sign) -> None: clean_signature = str(sign) - if len(clean_signature) > 256 or len(clean_signature) < 3: - self.app.logger.warning("Could not print message without a signature.") - flash("Could not print message without a signature.",category='error') - return False - + if len(sign) > 256 or len(sign) < 3: + self.app.logger.warning("Could not print signature of this length: " + str(len(clean_signature))) + raise Exception("Could not print signature of this length :" + str(len(clean_signature)) + ", needs to between 3 and 256 caracters long.") if not os.path.isfile(str(path)): self.app.logger.warning("File does not exist : " + str(path)) - flash('The file path for this image :' + str(path) + " wasn't found. Please try again.", 'error') - return False + raise Exception('The file path for this image :' + str(path) + " wasn't found. Please try again.") else: self.app.logger.debug("Printing file from " + str(path)) @@ -157,12 +153,15 @@ class Printer(object): self.app.logger.debug("Proccessing the image") path = process_image(self, path) except Exception as e: - flash(str(e)) self.app.logger.error(str(e)) + raise e + try: + self.check_paper() self.printer.open(self.usb_args) - self.printer.textln("> Printed by LittlePrinter | https://n07070.xyz/articles/littleprynter") + self.printer.textln("Printed by LittlePrinter ") + self.printer.textln("n07070.xyz/articles/littleprynter") self.printer.textln("> " + clean_signature + " @ " + strftime("%Y-%m-%d %H:%M:%S", gmtime())) self.printer.image(path) self.printer.cut() @@ -170,11 +169,10 @@ class Printer(object): self.app.logger.debug("Printed an image : " + str(path)) os.remove(path) self.app.logger.debug("Removed image.") - return True except Exception as e: self.printer.close() - flash(str(e),'error') - return False + self.app.logger.error(str(e)) + raise e def process_image(self, path): brightness_factor = 1.5 # Used only if image is too dark diff --git a/src/raspberry.py b/src/raspberry.py index aaa7017..fbfc632 100644 --- a/src/raspberry.py +++ b/src/raspberry.py @@ -1,4 +1,106 @@ -from printer import Printer +from flask_socketio import SocketIO +from gpiozero import Button, LED +from time import sleep +import io # To check if we are on a Raspberry Pi +import subprocess class Raspberry(object): - pass \ No newline at end of file + """ + This class will manage three things : + - Connecting to a USB webcam + - Managing a push button + - Activating a flash ( or light ) + - Flash an indicator light + + """ + def __init__(self, printer, app, socketio, button_gpio_port_number, indicator_gpio_port_number, is_flash_present,): + self.printer = printer + self.socketio = socketio + self.app = app + + self.is_flash_present = is_flash_present + self.button_gpio = button_gpio_port_number + self.led_gpio = indicator_gpio_port_number + self.image_path = self.app.config['UPLOAD_FOLDER'] + '/image.jpg' + + def initialise_gpio(self): + + self.button = Button(self.button_gpio) + self.led = LED(self.led_gpio) + + button.when_pressed = self.on_button_pressed() + + + def is_raspberry_pi(self, raise_on_errors=False): + # Check if we are running on a raspberry pi + try: + with io.open('/proc/cpuinfo', 'r') as cpuinfo: + found = False + for line in cpuinfo: + if line.startswith('Hardware'): + found = True + label, value = line.strip().split(':', 1) + value = value.strip() + if value not in ( + 'BCM2708', + 'BCM2709', + 'BCM2711', + 'BCM2835', + 'BCM2836' + ): + self.app.logger.debug('This system does not appear to be a Raspberry Pi.') + return False + + if not found: + self.app.logger.error('Couldn\'t get sufficient hardware information from /proc/cpuinfo, Unable to determine if we are on a Raspberry Pi.') + return False + except IOError: + self.app.logger.error('Unable to open `/proc/cpuinfo`.') + return False + + self.app.logger.debug('It seems we are on a Raspberry Pi') + return True + + def take_picture(self): + # return a path to a picture + try: + subprocess.run(['fswebcam', '--no-banner', self.image_path]) + except Exception as e: + self.app.logger.error('Unable to take a picture :' + e) + raise e + return True + + def flash(state): + if state: + # turn on the flash + pass + else: + # turn off the flash + pass + + def flash_indicator(self): + # The indicator will flash three times before return true + self.led.on() + sleep(0.1) + self.led.off() + sleep(1) + self.led.on() + sleep(0.1) + self.led.off() + sleep(1) + self.led.on() + sleep(0.1) + self.led.off() + + def on_button_pressed(): + self.socketio.emit('button_pressed') # Notify clients that a button has been pressed on the raspberry pi + self.flash_indicator() # The indicator will flash a countdown LED + self.flash(True) + try: + self.take_picture() + self.printer.print_image(self.image_path) + except Exception as e: + app.logger.error("Could not take a picture after the button press : " + e) + self.led.on() + self.flash(False) + self.socketio.emit('picture_taken') # Notify clients that a picture has been taken diff --git a/src/static/js/webcam.js b/src/static/js/webcam.js index 36f8e2b..1fa8cef 100644 --- a/src/static/js/webcam.js +++ b/src/static/js/webcam.js @@ -4,38 +4,72 @@ var current_camera_is; var width = document.getElementById("video").parentNode.parentElement.clientWidth; var height = width / (4 / 3); +var socket = io(); -function startup(){ +socket.on('connect', function() { + socket.emit('ping', {data: 'I\'m connected!'}); + console.log("Sent a ping to the server :)"); +}); + + +socket.on('pong', () => { + console.log('Received pong back ! '); +}); + +socket.on('new_image', () => { + console.log("Received new image event"); + const img = document.getElementById('snapshot'); + img.src = '/image?' + new Date().getTime(); // Bust cache +}); + + +async function startup(){ video = document.getElementById('video'); canvas = document.getElementById('canvas'); photo = document.getElementById('photo'); switch_cameras = document.getElementById('flip') printButton = document.getElementById('print_button'); - - if (check_webcam() === true ){ + if ( await check_webcam() === true ){ setup_events(); clear_canvas(); + } else if ( await check_server_camera() === true){ + console.log("The server has a camera, using it."); } else { no_webcam_error(); console.log("Seems like it's impossible to get a webcam."); } } -function check_webcam(){ +async function check_webcam(){ console.log("Cheking for a camera..."); - if (get_front_webcam()) { + if (await get_front_webcam()) { console.log("Got front camera !"); return true; } - if (get_any_webcam()) { + if (await get_any_webcam()) { console.log("Got a webcam !"); return true; } - console.log("Nope"); + + console.log("Found no usable cameras on client device"); + return false; } +async function check_server_camera(){ + console.log("Checking for a camera on the server"); + + await socket.emit('get_camera_status', (response) => { + if (response){ + console.log("Server has a camera !"); + } else { + console.log("The server doesn't seem to have a camera"); + } + }); + +} + function setup_events(){ // When the video is ready, we start streaming it to the canvas. @@ -206,7 +240,7 @@ async function get_webcam(options){ } } -function get_any_webcam(){ +async function get_any_webcam(){ var camera_options = { video: { facingMode: 'environment', // Or 'environment' if we want a camera facing away @@ -214,7 +248,7 @@ function get_any_webcam(){ audio: false }; - if(get_webcam(camera_options)){ + if(await get_webcam(camera_options)){ console.log("Got any camera, or environment camera."); current_camera_is = "any"; return true; @@ -223,11 +257,11 @@ function get_any_webcam(){ } } -function get_front_webcam(){ +async function get_front_webcam(){ // We try to start with the front facing camera, // if we have no support, we switch back to a normal camera. try { - const supports = navigator.mediaDevices.getSupportedConstraints(); + const supports = navigator.mediaDevices.getSupportedConstraints(); if (!supports['facingMode']) { throw new Error("This browser does not support facingMode!"); } else { @@ -246,7 +280,7 @@ function get_front_webcam(){ }; } - if(get_webcam(camera_options)){ + if(await get_webcam(camera_options)){ console.log("Got the front camera"); current_camera_is = "front"; return true; diff --git a/src/templates/base.html b/src/templates/base.html index 237fa2a..2bf4de2 100644 --- a/src/templates/base.html +++ b/src/templates/base.html @@ -56,4 +56,6 @@

Little Prynter is built by n07070 because it's fun :) | Source code - AGPLv3

+ + diff --git a/src/templates/webcam.html b/src/templates/webcam.html index 92fa6ac..8361e85 100644 --- a/src/templates/webcam.html +++ b/src/templates/webcam.html @@ -99,6 +99,7 @@ +
diff --git a/src/web.py b/src/web.py index 9157795..87e9dbd 100644 --- a/src/web.py +++ b/src/web.py @@ -1,4 +1,4 @@ -from flask import Flask, request +from flask import Flask, request, flash from werkzeug.utils import secure_filename from printer import Printer import time @@ -15,28 +15,27 @@ class Web(object): def print_sms(self, texte, sign: str): # TODO: verify the texte before printing it here ? self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign)) - if not os.getenv('LIPY_DEBUG'): - time.sleep(1) + try: + self.printer.print_sms(texte, sign) + except Exception as e: + self.app.logger.error(e) + flash("Error while printing the SMS : "+ str(e)) - return self.printer.print_sms(texte, sign) + flash("You message has been printed :)") - def print_image(self, image, sign: str) -> bool: + + def print_image(self, image, sign: str): self.app.logger.debug("Uploading file") try: self.app.logger.debug("Uploading file from " + str(sign)) if self.upload_file(image): self.app.logger.debug("File has been uploaded, printing...") self.printer.print_img(os.path.join(self.app.config['UPLOAD_FOLDER'], secure_filename(image.filename)), sign) - return True - else: - return False except Exception as e: self.app.logger.error(e) - raise Exception + flash("Could not upload file." + str(e)) - else: - flash("Could not upload file.",'error') - return False + flash("Your image has been printed :)") def login(username: str,password: str) -> bool: pass