Compare commits
17 Commits
multi-prin
...
3769e17444
| Author | SHA1 | Date | |
|---|---|---|---|
|
3769e17444 | ||
|
|
98299cc281 | ||
|
|
d261bb0fdc | ||
|
|
46fda975d4 | ||
|
|
670ab495d8 | ||
|
|
34bffe93af | ||
|
|
8ecb93656c | ||
|
|
eac737b889 | ||
|
|
dfeb1be0f0 | ||
|
|
0eefafdeb2 | ||
|
|
ce70d498ca | ||
|
|
991f6794c7 | ||
|
|
2b8a57b14a | ||
|
|
117f069351 | ||
|
|
981eba100d | ||
|
|
cf557c89cd | ||
|
|
2cc0b40b90 |
61
README.md
61
README.md
@@ -19,56 +19,44 @@ To make this project work, you will need :
|
||||
- Some electric wires.
|
||||
- Some knowledge of the command line,
|
||||
- Some knowledge of Python.
|
||||
- You will need to install [Poetry](https://python-poetry.org/) to manage the depedencies of the projet.
|
||||
- 3h of your time, 5h if things need debugging.
|
||||
- You will need to following packages: `fswebcam`, `libjpeg-dev` ,`zlib1g-dev`,`libffi-dev`,`git`, `virtualenv`,`pip` and `python` >= 3.8.6.
|
||||
- `fswebcam`, `libjpeg-dev` ,`zlib1g-dev`,`libffi-dev`,`git`, `virtualenv`,`pip` and `python` >= 3.8.6.
|
||||
- A webcam for the webcam page to work. Will work on a smartphone. Not required.
|
||||
|
||||
## Context
|
||||
|
||||
### Testing your printer
|
||||
For the EPSON TM-T20III, you can get the CUPS driver from [here](https://download.epson-biz.com/modules/pos/index.php?page=soft&scat=32). It's not specific to the printer I've been using, so you can try with other printers.
|
||||
|
||||
With that, you can try out your printer and print normal text, images or pdf documents for example. For the Adafruit printer, start by following the guide [here](https://learn.adafruit.com/networked-thermal-printer-using-cups-and-raspberry-pi) to install the CUPS software needed to print images. If you want, you can install it via the command line, [following this guide](https://help.ubuntu.com/lts/serverguide/cups.html).
|
||||
With that, you can try out your printer and print normal text, images or pdf documents for example. LittlePrynter itself does not require any other software than the ones installed with it, i.e the ones listed in the `requirements.txt` file.
|
||||
|
||||
For the Adafruit printer, start by following the guide [here](https://learn.adafruit.com/networked-thermal-printer-using-cups-and-raspberry-pi) to install the CUPS software needed to print images. If you want, you can install it via the command line, [following this guide](https://help.ubuntu.com/lts/serverguide/cups.html).
|
||||
|
||||
You can also get some information from [here](https://learn.adafruit.com/mini-thermal-receipt-printer) and [here](https://learn.adafruit.com/instant-camera-using-raspberry-pi-and-thermal-printer) if you're stuck.
|
||||
|
||||
### How LittlePrynter works
|
||||
|
||||
LittlePrynter itself does not require any other software than the ones installed with it, i.e the ones listed in the `pyproject.toml` file.
|
||||
|
||||
The version for the EPSON TM-T20III uses a library called `python-escpos`, which doesn't need a CUPS driver or anything else. It's included in the Python dependencies.
|
||||
|
||||
The project only supports the EPSON printer, but you can try to adapt it for other printers. For example, using `adafruit-thermal`, or `brother-ql`.
|
||||
The project only supports the EPSON printer, but you can try to adapt it for other printers using the `adafruit-thermal` branch, but I won't offer support for it.
|
||||
|
||||
### Install & setup the project :
|
||||
|
||||
Theses commands will copy the software on your computer, go into the directory, then activate a virtual environnement and install all of the project's dependecies.
|
||||
|
||||
```
|
||||
$ git clone https://git.n07070.xyz/n07070/LittlePrynter
|
||||
$ virtualenv LittlePrynter
|
||||
$ cd LittlePrynter
|
||||
$ eval "$(poetry env activate)"
|
||||
$ poetry install
|
||||
$ source bin/activate
|
||||
$ pip install -r requirements.txt
|
||||
```
|
||||
|
||||
> tip : when you're done, you can get out of the poetry environnement either by closing your terminal, or by running `deactivate`.
|
||||
|
||||
### Configure LittlePrynter
|
||||
> tip : when you're done, you can get out of the virtualenv either by closing your terminal, or by running `deactivate`.
|
||||
|
||||
You should see a folder named `configuration`. Enter it, and duplicate the file named `config.toml.sample`, and rename the copy to `config.toml`. Now, edit this file by following the comments in the file itself.
|
||||
|
||||
You should also setup the proper `udev` permissions to access the printers via USB, following the configuration found [here](https://python-escpos.readthedocs.io/en/latest/user/installation.html#setup-udev-for-usb-printers).
|
||||
|
||||
You can now start the web server with
|
||||
|
||||
```
|
||||
$ export FLASK_APP=src/main.py
|
||||
$ flask run --cert=adhoc
|
||||
$ flask run
|
||||
```
|
||||
|
||||
The `--cert=adhoc` argument will make LittlePrinter accessible in HTTPS with a [self-signed SSL certificate](https://en.wikipedia.org/wiki/Self-signed_certificate). This provides a free and easy way to use HTTPS, but does not provide any trust value. It is, on the other hand, a good way to access the webcam in developpement mode or on a local network, because a HTTPS connexion is required by browsers to access the needed APIs.
|
||||
|
||||
This command should start a web server with which you can test your configuration. If you plan on exposing your printer to the Internet, and give it an IP / URL, _please, please, please_, don't run it this way. Look at Flask's documentation and read about running a production server. It's a little more work, but it will prevent your computer/server being hacked in too easily.
|
||||
|
||||
Voilà !
|
||||
@@ -79,34 +67,6 @@ Your contributions are very much welcome ! You can either request an account on
|
||||
|
||||
Please also say if you had a printer to test your code, and which printer you've been using.
|
||||
|
||||
### Code structure
|
||||
|
||||
The app is written about the Flask framework. You can start by looking at the code in the `src/` folder, in the `main.py` file. There, you will see that a few classes are initialized. In general, they are two parts to the program :
|
||||
|
||||
The Web pages and the API, which are the user-facing parts. This is with what the users will interact, and define how the program is going to be used. The web pages are renderer from the `include/` folder where Jinj2 templates are defined.
|
||||
|
||||
The Worker and Printer Queue are the internal parts. When a new thing needs to be printed, usually sent from the Web or API interfaces, a new Task in the type of the document is created, and added to a print queue. Then, a Worker thread looks up the state of the queue every so often and picks jobs to execute on the printers connected to the system.
|
||||
|
||||
The last part of the program is the Raspberry Pi class, that handles to Photomaton mode, which handles button presses, and LED indicator and a flash.
|
||||
|
||||
### Linting
|
||||
|
||||
If you want to contribute code, please make sure to lint the project before commiting. This helps the code keep a general structure, and avoids some commons erros and mistakes.
|
||||
|
||||
To do so, you can run the following command, which will modify your files to be in a certain coding style ;
|
||||
|
||||
```
|
||||
black src/
|
||||
```
|
||||
|
||||
Beware that this command *will* re-write files, so doing `git add <file>` and then `black src/` and then `git diff` to see what the linter has done is a good idea.
|
||||
|
||||
Then, you can run another command, called Pylint ( [documentation](https://www.pylint.org/) ) which will rate your code. Try to get 10/10 : an excellent code works better and make you a better programmer !
|
||||
|
||||
```
|
||||
pylint src/
|
||||
```
|
||||
|
||||
## Screenshots
|
||||
|
||||
|
||||
@@ -129,7 +89,6 @@ If you liked this project, feel free to support my work !
|
||||
- [Github repo with CUPS drivers for the Adafruit Thermal Printer ( zj-58 )](https://github.com/klirichek/zj-58)
|
||||
- [A link to buy one in Europe](https://rlx.sk/sk/various-boards/1829-mini-thermal-receipt-printer-adafruit-597.html)
|
||||
- [Another link to buy one, direct from factory](https://www.cashinotech.com/csn-a2-58mm-mini-panel-thermal-receipt-printer_p11.html)
|
||||
- [Diagnostickoeur](https://framagit.org/stickoeur/diagnostickoeur), another printer software mainly around Brother QL printers.
|
||||
|
||||
|
||||
## Licence
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
# Configuration file the LittlePrynter
|
||||
|
||||
[defaults]
|
||||
signature = "Anonymous"
|
||||
|
||||
# Printer settings
|
||||
[printer]
|
||||
vendor_id = 0x04b8
|
||||
device_id = 0x0e28
|
||||
upload_folder = "src/static/uploads"
|
||||
|
||||
# Raspberry Pi Configuration
|
||||
|
||||
@@ -1,8 +1,5 @@
|
||||
# Configuration file the LittlePrynter
|
||||
|
||||
[defaults]
|
||||
signature = "Anonymous"
|
||||
|
||||
# Printer settings
|
||||
[printer]
|
||||
vendor_id = "0x04b8"
|
||||
|
||||
@@ -6,7 +6,7 @@ After=network.target
|
||||
Type=simple
|
||||
WorkingDirectory=/home/pi/littleprynter/
|
||||
Environment=FLASK_APP=src/main.py
|
||||
ExecStart=/home/pi/littleprynter/bin/flask run --host 0.0.0.0 --no-reload
|
||||
ExecStart=/home/pi/littleprynter/bin/flask run --host 0.0.0.0 --debug --no-reload
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
1
pip-selfcheck.json
Normal file
1
pip-selfcheck.json
Normal file
@@ -0,0 +1 @@
|
||||
{"last_check":"2018-10-10T18:33:22Z","pypi_version":"18.1"}
|
||||
1886
poetry.lock
generated
1886
poetry.lock
generated
File diff suppressed because it is too large
Load Diff
@@ -1,56 +0,0 @@
|
||||
[project]
|
||||
name = "littleprynter"
|
||||
version = "0.1.0"
|
||||
description = "A web interface and API to print on thermal printers"
|
||||
authors = [
|
||||
{name = "n07070",email = "contact@n07070.xyz"}
|
||||
]
|
||||
license = "AGPLv3"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
dependencies = [
|
||||
"flask (>=3.1.3,<4.0.0)",
|
||||
"numpy (>=2.3.4)",
|
||||
"toml (>=0.10.2,<0.11.0)",
|
||||
"flask-socketio (>=5.6.1,<6.0.0)",
|
||||
"flask-limiter (>=4.1.1,<5.0.0)",
|
||||
"gpiozero (>=2.0.1)",
|
||||
"cryptography (>=48.0.0,<49.0.0)",
|
||||
"pyusb (>=1.3.1,<2.0.0)",
|
||||
"pyserial (>=3.5,<4.0)",
|
||||
"qrcode (<9.0)",
|
||||
"python-barcode (>=0.16.1)",
|
||||
"setuptools (==81)",
|
||||
"cffi (>=2.0.0,<3.0.0)",
|
||||
"configobj (>=5.0.9,<6.0.0)",
|
||||
"future (>=1.0.0,<2.0.0)",
|
||||
"h11 (>=0.16.0,<0.17.0)",
|
||||
"idna (>=3.15,<4.0)",
|
||||
"itsdangerous (>=2.2.0,<3.0.0)",
|
||||
"jinja2 (>=3.1.6,<4.0.0)",
|
||||
"jsonpointer (>=3.1.1,<4.0.0)",
|
||||
"jsonschema (>=4.26.0,<5.0.0)",
|
||||
"limits (>=5.8.0,<6.0.0)",
|
||||
"markupsafe (>=3.0.3,<4.0.0)",
|
||||
"pillow (>=12.2.0,<13.0.0)",
|
||||
"pycryptodomex (>=3.23.0,<4.0.0)",
|
||||
"python-dateutil (>=2.9.0.post0,<3.0.0)",
|
||||
"python-engineio (>=4.13.1,<5.0.0)",
|
||||
"python-socketio (>=5.16.1,<6.0.0)",
|
||||
"pyyaml (>=6.0.3,<7.0.0)",
|
||||
"requests (>=2.34.2,<3.0.0)",
|
||||
"python-escpos (>=3.1,<4.0)",
|
||||
"pep8 (>=1.7.1,<2.0.0)",
|
||||
"pylint (>=4.0.5,<5.0.0)",
|
||||
"brother-ql-inventree (>=1.3,<2.0)",
|
||||
]
|
||||
|
||||
[tool.poetry]
|
||||
package-mode = false
|
||||
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python-escpos = {allow-prereleases = true}
|
||||
[build-system]
|
||||
requires = ["poetry-core>=2.0.0,<3.0.0"]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
261
requirements.txt
Normal file
261
requirements.txt
Normal file
@@ -0,0 +1,261 @@
|
||||
appdirs==1.4.4
|
||||
apt-listchanges==4.8
|
||||
argcomplete==3.6.2
|
||||
arrow==1.3.0
|
||||
attrs==25.3.0
|
||||
babel==2.17.0
|
||||
bcrypt==4.2.0
|
||||
bidict==0.23.1
|
||||
blinker==1.9.0
|
||||
certifi==2025.1.31
|
||||
cffi==1.17.1
|
||||
chardet==5.2.0
|
||||
charset-normalizer==3.4.2
|
||||
click==8.3.0
|
||||
cloud-init==25.2
|
||||
colorzero==2.0
|
||||
configobj==5.0.9
|
||||
cryptography==43.0.0
|
||||
Deprecated==1.2.18
|
||||
distlib==0.3.9
|
||||
distro==1.9.0
|
||||
escpos==2.0.0
|
||||
filelock==3.18.0
|
||||
Flask==3.1.2
|
||||
Flask-Limiter==4.0.0
|
||||
Flask-SocketIO==5.5.1
|
||||
fqdn==1.5.1
|
||||
future==1.0.0
|
||||
gpiod==2.2.0
|
||||
gpiozero==2.0.1
|
||||
h11==0.16.0
|
||||
idna==3.10
|
||||
isoduration==20.11.0
|
||||
itsdangerous==2.2.0
|
||||
Jinja2==3.1.6
|
||||
jsonpatch==1.32
|
||||
jsonpointer==2.4
|
||||
jsonschema==4.19.2
|
||||
jsonschema-specifications==2023.12.1
|
||||
lgpio==0.2.2.0
|
||||
limits==5.6.0
|
||||
linkify-it-py==2.0.3
|
||||
markdown-it-py==3.0.0
|
||||
MarkupSafe==2.1.5
|
||||
mdurl==0.1.2
|
||||
numpy==2.3.4
|
||||
oauthlib==3.2.2
|
||||
olefile==0.47
|
||||
ordered-set==4.1.0
|
||||
packaging==25.0
|
||||
pillow==11.1.0
|
||||
platformdirs==4.3.7
|
||||
ply==3.11
|
||||
pycparser==2.22
|
||||
pycryptodomex==3.20.0
|
||||
Pygments==2.18.0
|
||||
PyJWT==2.10.1
|
||||
pyserial==3.5
|
||||
python-apt==3.0.0
|
||||
python-barcode==0.16.1
|
||||
python-dateutil==2.9.0
|
||||
python-engineio==4.12.3
|
||||
python-socketio==5.14.1
|
||||
pyusb==1.3.1
|
||||
PyYAML==6.0.2
|
||||
qrcode==8.2
|
||||
referencing==0.36.2
|
||||
requests==2.32.3
|
||||
rfc3339-validator==0.1.4
|
||||
rfc3986-validator==0.1.1
|
||||
rfc3987==1.3.8
|
||||
rich==13.9.4
|
||||
rpds-py==0.21.0
|
||||
rpi-keyboard-config==1.0
|
||||
rpi-lgpio==0.6
|
||||
setuptools==80.9.0
|
||||
simple-websocket==1.1.0
|
||||
six==1.17.0
|
||||
smbus2==0.4.3
|
||||
spidev==3.6
|
||||
ssh-import-id==5.10
|
||||
toml==0.10.2
|
||||
types-aiofiles==24.1
|
||||
types-antlr4-python3-runtime==4.13
|
||||
types-assertpy==1.1
|
||||
types-atheris==2.3
|
||||
types-aws-xray-sdk==2.14
|
||||
types-beautifulsoup4==4.12
|
||||
types-bleach==6.2
|
||||
types-boltons==24.1
|
||||
types-braintree==4.31
|
||||
types-cachetools==5.5
|
||||
types-caldav==1.3
|
||||
types-capturer==3.0
|
||||
types-cffi==1.16
|
||||
types-chevron==0.14
|
||||
types-click-default-group==1.2
|
||||
types-click-spinner==0.1
|
||||
types-colorama==0.4
|
||||
types-commonmark==0.9
|
||||
types-console-menu==0.8
|
||||
types-corus==0.10
|
||||
types-croniter==5.0.1
|
||||
types-dateparser==1.2
|
||||
types-decorator==5.1
|
||||
types-defusedxml==0.7
|
||||
types-Deprecated==1.2.15
|
||||
types-docker==7.1
|
||||
types-dockerfile-parse==2.0
|
||||
types-docutils==0.21
|
||||
types-editdistance==0.8
|
||||
types-entrypoints==0.4
|
||||
types-ExifRead==3.0
|
||||
types-fanstatic==1.4
|
||||
types-first==2.0
|
||||
types-flake8==7.1
|
||||
types-flake8-bugbear==24.12.12
|
||||
types-flake8-builtins==2.5
|
||||
types-flake8-docstrings==1.7
|
||||
types-flake8-rst-docstrings==0.3
|
||||
types-flake8-simplify==0.21
|
||||
types-flake8-typing-imports==1.16
|
||||
types-Flask-Cors==5.0
|
||||
types-Flask-Migrate==4.0
|
||||
types-Flask-SocketIO==5.4
|
||||
types-fpdf2==2.8.2
|
||||
types-gdb==15.0
|
||||
types-gevent==24.11
|
||||
types-google-cloud-ndb==2.3
|
||||
types-greenlet==3.1
|
||||
types-hdbcli==2.23
|
||||
types-html5lib==1.1
|
||||
types-httplib2==0.22
|
||||
types-humanfriendly==10.0
|
||||
types-hvac==2.3
|
||||
types-ibm-db==3.2.4
|
||||
types-icalendar==6.1
|
||||
types-influxdb-client==1.45
|
||||
types-inifile==0.4
|
||||
types-JACK-Client==0.5
|
||||
types-Jetson.GPIO==2.1
|
||||
types-jmespath==1.0
|
||||
types-jsonschema==4.23
|
||||
types-jwcrypto==1.5
|
||||
types-keyboard==0.13
|
||||
types-ldap3==2.9
|
||||
types-libsass==0.23
|
||||
types-lupa==2.2
|
||||
types-lzstring==1.0
|
||||
types-m3u8==6.0
|
||||
types-Markdown==3.7
|
||||
types-mock==5.1
|
||||
types-mypy-extensions==1.0
|
||||
types-mysqlclient==2.2
|
||||
types-nanoid==2.0.0
|
||||
types-netaddr==1.3
|
||||
types-netifaces==0.11
|
||||
types-networkx==3.4.2
|
||||
types-oauthlib==3.2
|
||||
types-objgraph==3.6
|
||||
types-olefile==0.47
|
||||
types-openpyxl==3.1.5
|
||||
types-opentracing==2.4
|
||||
types-paramiko==3.5
|
||||
types-parsimonious==0.10
|
||||
types-passlib==1.7
|
||||
types-passpy==1.0
|
||||
types-peewee==3.17.8
|
||||
types-pep8-naming==0.14
|
||||
types-pexpect==4.9
|
||||
types-pika-ts==1.3
|
||||
types-polib==1.2
|
||||
types-portpicker==1.6
|
||||
types-protobuf==5.29.1
|
||||
types-psutil==6.1
|
||||
types-psycopg2==2.9.10
|
||||
types-pyasn1==0.6
|
||||
types-pyaudio==0.2
|
||||
types-PyAutoGUI==0.9
|
||||
types-pycocotools==2.0
|
||||
types-pycurl==7.45.4
|
||||
types-pyfarmhash==0.4
|
||||
types-pyflakes==3.2
|
||||
types-pygit2==1.15
|
||||
types-Pygments==2.18
|
||||
types-pyinstaller==6.11
|
||||
types-pyjks==20.0
|
||||
types-PyMySQL==1.1
|
||||
types-pynput==1.7.7
|
||||
types-pyOpenSSL==24.1
|
||||
types-pyRFC3339==2.0.1
|
||||
types-PyScreeze==1.0.1
|
||||
types-pyserial==3.5
|
||||
types-pysftp==0.2
|
||||
types-pytest-lazy-fixture==0.6
|
||||
types-python-crontab==3.2
|
||||
types-python-datemath==3.0.1
|
||||
types-python-dateutil==2.9
|
||||
types-python-http-client==3.3.7
|
||||
types-python-jenkins==1.8
|
||||
types-python-jose==3.3
|
||||
types-python-nmap==0.7
|
||||
types-python-xlib==0.33
|
||||
types-pytz==2024.2
|
||||
types-pywin32==308
|
||||
types-pyxdg==0.28
|
||||
types-PyYAML==6.0
|
||||
types-qrbill==1.1
|
||||
types-qrcode==8.0
|
||||
types-regex==2024.11.6
|
||||
types-reportlab==4.2.5
|
||||
types-requests==2.32
|
||||
types-requests-oauthlib==2.0
|
||||
types-retry==0.9
|
||||
types-RPi.GPIO==0.7
|
||||
types-s2clientprotocol==5
|
||||
types-seaborn==0.13.2
|
||||
types-Send2Trash==1.8
|
||||
types-setuptools==75.6
|
||||
types-shapely==2.0
|
||||
types-simplejson==3.19
|
||||
types-singledispatch==4.1
|
||||
types-six==1.17
|
||||
types-slumber==0.7
|
||||
types-str2bool==1.1
|
||||
types-tabulate==0.9
|
||||
types-tensorflow==2.18.0
|
||||
types-TgCrypto==1.2
|
||||
types-toml==0.10
|
||||
types-toposort==1.10
|
||||
types-tqdm==4.67
|
||||
types-translationstring==1.4
|
||||
types-tree-sitter-languages==1.10
|
||||
types-ttkthemes==3.2
|
||||
types-ujson==5.10
|
||||
types-unidiff==0.7
|
||||
types-untangle==1.2
|
||||
types-usersettings==1.1
|
||||
types-uWSGI==2.0
|
||||
types-vobject==0.9.9
|
||||
types-waitress==3.0.1
|
||||
types-WebOb==1.8
|
||||
types-whatthepatch==1.0
|
||||
types-workalendar==17.0
|
||||
types-WTForms==3.2.1
|
||||
types-wurlitzer==3.1
|
||||
types-xdgenvpy==3.0
|
||||
types-xmltodict==0.14
|
||||
types-zstd==1.5
|
||||
types-zxcvbn==4.4
|
||||
typing_extensions==4.15.0
|
||||
uc-micro-py==1.0.3
|
||||
uritemplate==4.1.1
|
||||
urllib3==2.3.0
|
||||
viivakoodi==0.8.0
|
||||
virtualenv==20.31.2
|
||||
webcolors==1.13
|
||||
Werkzeug==3.1.3
|
||||
wheel==0.46.1
|
||||
wrapt==1.17.3
|
||||
wsproto==1.2.0
|
||||
7
run.sh
Executable file
7
run.sh
Executable file
@@ -0,0 +1,7 @@
|
||||
virtualenv .
|
||||
|
||||
pip install -r requirements.txt
|
||||
|
||||
export FLASK_APP=src/main.py
|
||||
export FLASK_ENV=development
|
||||
flask run --host 192.168.0.42 --debugger --eager-loading
|
||||
272
src/main.py
272
src/main.py
@@ -36,17 +36,15 @@ 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
|
||||
from print_queue import PrintQueue
|
||||
from worker import PrintWorker
|
||||
|
||||
# We create the main Flask object
|
||||
# Variables
|
||||
|
||||
app = Flask(__name__)
|
||||
socketio = SocketIO(app, cors_allowed_origins="*")
|
||||
|
||||
# Global variables
|
||||
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
|
||||
socketio = SocketIO(app)
|
||||
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif"}
|
||||
|
||||
# Load the configuration file
|
||||
try:
|
||||
@@ -72,7 +70,11 @@ except OSError as e:
|
||||
|
||||
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)
|
||||
@@ -80,52 +82,49 @@ except FileExistsError:
|
||||
app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
|
||||
except PermissionError:
|
||||
app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
|
||||
sys.exit(77)
|
||||
|
||||
# Output the config file
|
||||
if not os.getenv("FLASK_DEBUG") is None and os.getenv("FLASK_DEBUG") is True:
|
||||
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 10Mb for a file upload
|
||||
app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000 # Maximum 3Mb for a file upload
|
||||
app.config["TEMPLATES_AUTO_RELOAD"] = True
|
||||
|
||||
# Queue creation
|
||||
print_queue = PrintQueue(app)
|
||||
# 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(
|
||||
print_queue,
|
||||
printer,
|
||||
app,
|
||||
configuration_file
|
||||
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 management
|
||||
web = Web(app, print_queue)
|
||||
# Web & API routes
|
||||
|
||||
# Start worker thread
|
||||
# When created, the worker will try to find printers connected to the system
|
||||
try:
|
||||
worker = PrintWorker(app, print_queue)
|
||||
worker.start()
|
||||
except Exception as e:
|
||||
app.logger.error("Could not start the worker because %s ", str(e))
|
||||
sys.exit(-1)
|
||||
web = Web(app, printer)
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
|
||||
|
||||
# The rate limit
|
||||
limiter = Limiter(
|
||||
get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
|
||||
)
|
||||
|
||||
|
||||
app.logger.info("🖶 Welcome to LittlePrynter !")
|
||||
|
||||
# General routes
|
||||
@app.route("/")
|
||||
@limiter.limit("1/second", override_defaults=False)
|
||||
def index():
|
||||
@@ -142,86 +141,6 @@ def webcam():
|
||||
return render_template("webcam.html")
|
||||
|
||||
|
||||
# Form treatement
|
||||
@app.route("/web/print/sms", methods=["POST"])
|
||||
@limiter.limit("6/minute", override_defaults=False)
|
||||
def web_print_sms():
|
||||
"""Prints a short message on a printer"""
|
||||
app.logger.debug("Printing an sms via web method")
|
||||
|
||||
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 : " + str(e), "error")
|
||||
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 : %s ", str(e)
|
||||
)
|
||||
sign = configuration_file["defaults"]["signature"]
|
||||
|
||||
try:
|
||||
web.print_sms(txt, sign)
|
||||
except RuntimeError as e:
|
||||
app.logger.error("Whoops, we could not print an SMS because : %s ", str(e))
|
||||
flash("Whoops, we could not print an SMS because :" + str(e), "error")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
# end try
|
||||
flash("The SMS has been added to the print queue !", "info")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
@app.route("/web/print/img", methods=["POST"])
|
||||
@limiter.limit("1/second", override_defaults=False)
|
||||
def web_print_img():
|
||||
"""Prints an image on a printer"""
|
||||
app.logger.debug("Printing an image with web method")
|
||||
|
||||
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 : %s", str(e)
|
||||
)
|
||||
sign = configuration_file["defaults"]["signature"]
|
||||
except werkzeug.exceptions.RequestEntityTooLarge as e:
|
||||
flash("Whoops, image is too big: " + str(e), "error")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
# 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 : " + str(e), "error")
|
||||
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 !", "error")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
try:
|
||||
app.logger.debug("Sending the image to the printer.")
|
||||
web.print_image(file, sign)
|
||||
except RuntimeError as e:
|
||||
app.logger.error("The image could not be printed because : %s ", str(e))
|
||||
flash("The image could not be printed because : " + str(e), "error")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
flash("Picture added to the print queue !", "info")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
# API routes
|
||||
# The api has the following methods
|
||||
# api/print/{sms,img,letter,qr,barcode}
|
||||
@@ -243,122 +162,85 @@ def api_index():
|
||||
@limiter.limit("6/minute", override_defaults=False)
|
||||
def api_print_sms():
|
||||
"""Prints a short message on a printer"""
|
||||
app.logger.debug("Printing an sms via API")
|
||||
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))
|
||||
return str(e), 400
|
||||
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. : %s", str(e)
|
||||
app.logger.error(
|
||||
"Whoops, no forms submitted or missing signature : %s ", str(e)
|
||||
)
|
||||
sign = configuration_file["defaults"]["signature"]
|
||||
try:
|
||||
# comment: We try to print the SMS
|
||||
web.print_sms(txt, sign)
|
||||
except RuntimeError as e:
|
||||
return str(e), 500
|
||||
# end try
|
||||
return "OK", 200
|
||||
flash("Whoops, no forms submitted or missing signature : %s", str(e))
|
||||
return redirect(url_for("index"))
|
||||
|
||||
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 via API")
|
||||
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. %s", str(e)
|
||||
)
|
||||
sign = configuration_file["defaults"]["signature"]
|
||||
except Exception as e:
|
||||
app.logger.error("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"))
|
||||
|
||||
# check if the post request has the file part
|
||||
if "img" not in request.files:
|
||||
app.logger.error("Whoops, no images submitted.")
|
||||
return "No image submitted", 400
|
||||
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 submits an empty file without a filename.
|
||||
if file.filename == "":
|
||||
app.logger.error("Submitted file has no filename !")
|
||||
return "Submitted file has no filename !", 400
|
||||
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 RuntimeError as e:
|
||||
return str(e), 500
|
||||
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"))
|
||||
|
||||
return "OK", 200
|
||||
else:
|
||||
app.logger.error("Method not allowed")
|
||||
flash("Method not allowed")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
flash("Picture printed ! ")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
# TODO: This might not depend on the Raspberry Pi
|
||||
@app.route("/api/camera/picture", methods=["GET"])
|
||||
@app.route("/api/camera/picture")
|
||||
def camera_picture():
|
||||
"""Returns a picture taken by the camera on a raspberry pi"""
|
||||
"""Returns a picture taken by the camera"""
|
||||
if RASPBERRY_PI_CONNECTED:
|
||||
try:
|
||||
return rpi.camera_picture()
|
||||
except RuntimeError as e:
|
||||
except Exception as e:
|
||||
return jsonify({"message": "Error getting the stream : " + e}), 500
|
||||
else:
|
||||
return jsonify({"message": "No camera present"}), 500
|
||||
|
||||
|
||||
@app.route("/api/queue", methods=["GET"])
|
||||
def api_queue_status():
|
||||
"""API endpoint for entire queue"""
|
||||
return jsonify(web.get_queue_state())
|
||||
|
||||
|
||||
@app.route("/api/queue/completed", methods=["GET"])
|
||||
def api_queue_completed():
|
||||
"""API endpoint that returns the finished tasks"""
|
||||
return jsonify(web.get_queue_completed())
|
||||
|
||||
|
||||
@app.route("/api/worker", methods=["GET"])
|
||||
def api_worker_state():
|
||||
"""API endpoint to get the worker state"""
|
||||
return jsonify(worker.current_state())
|
||||
|
||||
|
||||
@app.route("/api/worker/start")
|
||||
def api_worker_start():
|
||||
"""
|
||||
Enable to worker. This starts to process the print queue.
|
||||
"""
|
||||
worker.start_worker()
|
||||
return jsonify(worker.current_state())
|
||||
|
||||
|
||||
@app.route("/api/worker/stop")
|
||||
def api_worker_stop():
|
||||
"""
|
||||
Stops the print queue. This stops the processing of the print queue.
|
||||
"""
|
||||
worker.stop_worker()
|
||||
return jsonify(worker.current_state())
|
||||
|
||||
|
||||
## Authentification
|
||||
|
||||
|
||||
@app.route("/login")
|
||||
@limiter.limit("1/second", override_defaults=False)
|
||||
def login_page():
|
||||
"""Unsued, logins"""
|
||||
# web.login(username,password)
|
||||
return redirect(url_for("index")), 501
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
@@ -366,7 +248,7 @@ def login_page():
|
||||
def logout_page():
|
||||
"""Unused, logout"""
|
||||
# web.logout(username, password)
|
||||
return redirect(url_for("index")), 501
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
@app.errorhandler(429)
|
||||
@@ -376,7 +258,7 @@ def ratelimit_handler(e):
|
||||
"Rate limit reached, please slow down :) ( Currently at " + e.description + ")",
|
||||
"error",
|
||||
)
|
||||
app.logger.debug("Rate limit reached %s ", str(e.description))
|
||||
app.logger.debug("Rate limit reached %s " , str(e.description))
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
@@ -392,7 +274,7 @@ def ping():
|
||||
@socketio.on("ping")
|
||||
def handle_message(data):
|
||||
"""Handle sockets pings"""
|
||||
app.logger.debug("Received : %s ", str(data))
|
||||
app.logger.debug("Received : %s " , str(data))
|
||||
socketio.emit("pong", "Pong !")
|
||||
|
||||
|
||||
@@ -404,7 +286,3 @@ def camera_status():
|
||||
socketio.emit("camera_status", True)
|
||||
else:
|
||||
socketio.emit("camera_status", False)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
|
||||
|
||||
@@ -1,144 +0,0 @@
|
||||
"""
|
||||
This class has the method by which we manage the Tasks
|
||||
It's a printing queue, so we need to add, remove and get information on where
|
||||
the queue is
|
||||
"""
|
||||
|
||||
from collections import deque
|
||||
|
||||
# Because actually printing and adding new print job requests happen at
|
||||
# diffrent times, the print queue is managed by it's own thread.
|
||||
import threading
|
||||
|
||||
from datetime import datetime
|
||||
from task import TaskType
|
||||
|
||||
|
||||
class PrintQueue:
|
||||
"""
|
||||
A Double-ended Queue to manage the printing Tasks
|
||||
"""
|
||||
|
||||
def __init__(self, app):
|
||||
self.app = app
|
||||
self._queue = deque()
|
||||
self._lock = threading.Lock()
|
||||
self._completed_tasks = {} # Store completed task info
|
||||
self._task_counter = 0
|
||||
self.app.logger.debug("Created a new PrintQueue")
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._queue)
|
||||
|
||||
def enqueue(self, task):
|
||||
"""Add task to right of the queue and return position"""
|
||||
with self._lock:
|
||||
try:
|
||||
self.app.logger.info("Add task %s to queue ", task.task_id)
|
||||
self._queue.append(task)
|
||||
position = self._queue.index(task)
|
||||
# We return the current position of the task if it was added
|
||||
self.app.logger.debug(
|
||||
"Added a new task %s to the queue at position %s",
|
||||
task.task_id,
|
||||
position,
|
||||
)
|
||||
return position
|
||||
except Exception as e:
|
||||
self.app.logger.error("Could not add a task to the queue : %s ", e)
|
||||
raise e
|
||||
|
||||
def dequeue(self):
|
||||
"""Remove and return next task ( from the left of the queue ) (thread-safe)"""
|
||||
with self._lock:
|
||||
return self._queue.popleft() if len(self._queue) > 0 else None
|
||||
|
||||
def get_position(self, task):
|
||||
"""Get current position of task in queue (1-indexed)"""
|
||||
with self._lock:
|
||||
if task.task_id in self._completed_tasks:
|
||||
return None # Task already completed
|
||||
|
||||
try:
|
||||
# Try to get the position of a Task
|
||||
return self._queue.index(task)
|
||||
except ValueError as e:
|
||||
raise e
|
||||
# end try
|
||||
|
||||
def is_empty(self):
|
||||
"""Check if queue is empty"""
|
||||
with self._lock:
|
||||
self.app.logger.debug("Checking if queue is empty")
|
||||
return len(self._queue) == 0
|
||||
|
||||
def get_queue_state(self):
|
||||
"""Return current queue state"""
|
||||
with self._lock:
|
||||
self.app.logger.debug("Return current queue state")
|
||||
return [
|
||||
{"task_id": t.task_id, "status": t.status, "type": str(t.task_type), "content": str(t.get_print_data())}
|
||||
for t in self._queue
|
||||
]
|
||||
|
||||
def get_queue_completed(self):
|
||||
"""Return completed queue elements"""
|
||||
with self._lock:
|
||||
self.app.logger.debug("Return completed queue elements")
|
||||
return self._completed_tasks
|
||||
|
||||
def get_status(self, task_id):
|
||||
"""Get full status info for a task"""
|
||||
with self._lock:
|
||||
|
||||
if task_id in self._completed_tasks:
|
||||
return self._completed_tasks[task_id]
|
||||
|
||||
# Check in queue if it exists
|
||||
for index, task in enumerate(self._queue):
|
||||
if task.task_id == task_id:
|
||||
# Depending on it's type, we return more info
|
||||
if task.task_type == TaskType.IMAGE:
|
||||
return {
|
||||
"task_id": task_id,
|
||||
"status": task.status,
|
||||
"type": task.task_type,
|
||||
"position": index,
|
||||
"in_queue": True,
|
||||
"content": task.content,
|
||||
"signature": task.signature,
|
||||
}
|
||||
|
||||
if task.task_type == TaskType.TEXT:
|
||||
return {
|
||||
"task_id": task_id,
|
||||
"status": task.status,
|
||||
"type": task.task_type,
|
||||
"position": index,
|
||||
"in_queue": True,
|
||||
"image_path": str(task.image_path),
|
||||
"signature": task.signature,
|
||||
"process": str(task.process),
|
||||
}
|
||||
|
||||
if task.task_type == TaskType.CUT:
|
||||
return {
|
||||
"task_id": task_id,
|
||||
"status": task.status,
|
||||
"type": task.task_type,
|
||||
"position": index,
|
||||
"in_queue": True,
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
def mark_completed(self, task_id, task_status):
|
||||
"""Mark task as completed and remove from queue"""
|
||||
with self._lock:
|
||||
self._completed_tasks[task_id] = {
|
||||
"task_id": task_id,
|
||||
"status": task_status,
|
||||
"position": None,
|
||||
"in_queue": False,
|
||||
"completed_at": datetime.now().isoformat(),
|
||||
}
|
||||
627
src/printer.py
627
src/printer.py
@@ -1,315 +1,183 @@
|
||||
"""
|
||||
This class manages connexion to a Printer
|
||||
"""
|
||||
|
||||
# Importing the module to mage the connection to the printer.
|
||||
from escpos.printer import Usb, USBNotFoundError
|
||||
from time import sleep, gmtime, strftime
|
||||
import os.path
|
||||
import os
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass
|
||||
import time
|
||||
from enum import Enum
|
||||
import uuid
|
||||
import threading
|
||||
import usb.core
|
||||
from PIL import Image, ImageEnhance
|
||||
from PIL import Image, ImageEnhance, ImageOps
|
||||
import numpy as np
|
||||
|
||||
# Importing the modules needed for each supported printer Type
|
||||
import escpos.printer
|
||||
from brother_ql.models import ModelsManager
|
||||
from brother_ql.backends import backend_factory
|
||||
from brother_ql.raster import BrotherQLRaster
|
||||
from brother_ql.conversion import convert
|
||||
from brother_ql.backends.helpers import send
|
||||
|
||||
|
||||
class PrinterType(Enum):
|
||||
class Printer(object):
|
||||
"""
|
||||
What are the capacities of a Printer ?
|
||||
# The connection is based on the ESC/POS library
|
||||
|
||||
## Connection to the USB printer
|
||||
|
||||
## Making sure the printer is alive
|
||||
|
||||
## Making sure it has paper
|
||||
|
||||
## Define default print settings
|
||||
|
||||
## Print starting message, log time of first print, cut.
|
||||
|
||||
## Annonce readyness : return a positive pong message.
|
||||
"""
|
||||
|
||||
EPSON = "epson"
|
||||
BROTHER = "brother"
|
||||
# Is the printer ready to accept a new print ?
|
||||
ready = False
|
||||
|
||||
|
||||
# For Brother-QL Printers
|
||||
@dataclass
|
||||
class PrinterInfo:
|
||||
"""
|
||||
Brother-QL printer information
|
||||
"""
|
||||
identifier: str
|
||||
backend: str
|
||||
protocol: str
|
||||
vendor_id: str
|
||||
product_id: str
|
||||
serial_number: str
|
||||
name: str = "Brother QL Printer"
|
||||
model: str = "QL-570"
|
||||
status: str = "unknown"
|
||||
label_type: str = "unknown"
|
||||
label_size: str = "unknown"
|
||||
label_width: int = 0
|
||||
label_height: int = 0
|
||||
|
||||
def __getitem__(self, item):
|
||||
return getattr(self, item)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
setattr(self, key, value)
|
||||
|
||||
|
||||
class Printer(ABC):
|
||||
"""
|
||||
If it outputs printed paper and speaks like a printer, then it must be a printer.
|
||||
"""
|
||||
|
||||
def __init__(self, app, vendor_id, device_id, printer_type: PrinterType):
|
||||
"""
|
||||
We initialize a Printer via it's USB connexion, and generate a unique ID
|
||||
"""
|
||||
self.id = uuid.uuid4()
|
||||
def __init__(self, app, device_id, vendor_id):
|
||||
super(Printer, self).__init__()
|
||||
self.app = app
|
||||
self.vendor_id = vendor_id
|
||||
self.device_id = device_id
|
||||
self.ready = False
|
||||
self.printer_type = printer_type
|
||||
self._lock = threading.Lock()
|
||||
|
||||
@abstractmethod
|
||||
def _has_paper(self) -> bool:
|
||||
"""Check if the printer has papier"""
|
||||
|
||||
@abstractmethod
|
||||
def _state(self) -> bool:
|
||||
"""Reports the state of the Printer"""
|
||||
|
||||
@abstractmethod
|
||||
def print_task(self, task_type, data) -> None:
|
||||
"""Takes a PrintTask and executes it"""
|
||||
|
||||
|
||||
class EscPosPrinter(Printer):
|
||||
"""
|
||||
Create a new ESC/POS based printer.
|
||||
"""
|
||||
|
||||
def __init__(self, app, vendor_id, device_id):
|
||||
"""
|
||||
Create a connexion to a ESC/POS Printer via USB,
|
||||
Making sure the printer is alive,
|
||||
Making sure it has paper,
|
||||
Define default print settings
|
||||
"""
|
||||
super().__init__(app, vendor_id, device_id, printer_type=PrinterType.EPSON)
|
||||
self.printer = None
|
||||
self.device_id = device_id
|
||||
self.vendor_id = vendor_id
|
||||
self.usb_args = {}
|
||||
self.usb_args["idVendor"] = self.vendor_id
|
||||
self.usb_args["idProduct"] = self.device_id
|
||||
self.usb_args['idVendor'] = self.device_id
|
||||
self.usb_args['idProduct'] = self.vendor_id
|
||||
|
||||
try:
|
||||
# This also calls open(), which we need to close()
|
||||
# or else the device will appear as busy.
|
||||
p = escpos.printer.Usb(self.vendor_id, self.device_id, 0, profile="TM-P80")
|
||||
except escpos.exceptions.DeviceNotFoundError as e:
|
||||
self.app.logger.error(
|
||||
"The USB device is not plugged in : %s",
|
||||
str(e),
|
||||
)
|
||||
except Exception as e:
|
||||
self.app.logger.error("Printer could not be connected : %s ", str(e))
|
||||
|
||||
try:
|
||||
if p.is_online():
|
||||
self.app.logger.debug("Printer online !")
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
# Setting up the printing options.
|
||||
p.set(
|
||||
align="center",
|
||||
font="a",
|
||||
bold=False,
|
||||
underline=0,
|
||||
width=1,
|
||||
height=1,
|
||||
density=9,
|
||||
invert=False,
|
||||
smooth=False,
|
||||
flip=False,
|
||||
double_width=False,
|
||||
double_height=False,
|
||||
custom_size=False,
|
||||
)
|
||||
# Beware : if we print every time the printer becomes ready, it means
|
||||
# we are printing before and after every print !
|
||||
self.printer = p
|
||||
self.printer.close() # We close the connexion to the Printer
|
||||
|
||||
try:
|
||||
self._has_paper()
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
self.ready = True
|
||||
|
||||
def _has_paper(self):
|
||||
"""Check if the printer has paper left"""
|
||||
self.app.logger.debug("Checking paper status...")
|
||||
def check_paper(self) -> bool:
|
||||
# Let's check paper status
|
||||
self.app.logger.debug('Checking paper status...')
|
||||
self.printer.open(self.usb_args)
|
||||
status = self.printer.paper_status()
|
||||
match status:
|
||||
case 0:
|
||||
self.app.logger.error("Printer has no more paper, aborting...")
|
||||
self.app.logger.error('Printer has no more paper, aborting...')
|
||||
self.printer.close()
|
||||
raise RuntimeError("No more paper in the printer")
|
||||
raise Exception("No more paper in the printer")
|
||||
case 1:
|
||||
self.app.logger.warning(
|
||||
"Printer needs paper to be changed very soon ! "
|
||||
)
|
||||
self.app.logger.warning('Printer needs paper to be changed very soon ! ')
|
||||
self.printer.close()
|
||||
return True
|
||||
case 2:
|
||||
self.app.logger.debug("Printer has paper, good to go")
|
||||
self.app.logger.debug('Printer has paper, good to go')
|
||||
self.printer.close()
|
||||
return True
|
||||
|
||||
def _print_txt(self, msg, signature="", bold=False):
|
||||
self.ready = False
|
||||
if not isinstance(msg, str):
|
||||
self.app.logger.error(
|
||||
"It is not possible to print a " + str(type(msg)) + ", only strings."
|
||||
)
|
||||
raise ValueError
|
||||
def init_printer(self):
|
||||
|
||||
# We make sure that the signature is not something too goofy
|
||||
# Is the printer online ? Is the communication with the printer successfull ?
|
||||
waiting_elapsed = 30
|
||||
self.app.logger.debug('Waiting for printer to get online...')
|
||||
|
||||
while not self.ready:
|
||||
try:
|
||||
# This also calls open(), which we need to close()
|
||||
# or else the device will appear as busy.
|
||||
p = Usb(self.device_id, self.vendor_id, 0, profile="TM-P80")
|
||||
except USBNotFoundError as e:
|
||||
self.app.logger.error("The USB device is not plugged in, trying again : " + str(e))
|
||||
pass
|
||||
|
||||
try:
|
||||
if p.is_online():
|
||||
self.ready = True
|
||||
self.app.logger.debug('Printer online !')
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
sleep(1)
|
||||
waiting_elapsed -= 1
|
||||
if waiting_elapsed < 1:
|
||||
self.app.logger.error('Printer took more than 30 seconds to get online, aborting...')
|
||||
waiting_elapsed = 30 # Reset the waiting time for the next print.
|
||||
return False
|
||||
|
||||
|
||||
|
||||
|
||||
# Setting up the printing options.
|
||||
p.set(align='center', font='a', bold=False, underline=0, width=1, height=1, density=9, invert=False, smooth=False, flip=False, double_width=False, double_height=False, custom_size=False)
|
||||
# Beware : if we print every time the printer becomes ready, it means
|
||||
# we are printing before and after every print !
|
||||
self.printer = p
|
||||
self.ready = True;
|
||||
self.printer.close();
|
||||
|
||||
self.check_paper()
|
||||
|
||||
return True
|
||||
|
||||
def print_sms(self, msg, signature="",bold=False):
|
||||
clean_msg = str(msg) + "\n"
|
||||
clean_signature = str(signature)
|
||||
|
||||
# Make checks on the size of the message being printed
|
||||
if len(clean_msg) > 4096:
|
||||
self.app.logger.warning(
|
||||
"Could not print message of this length: " + str(len(clean_msg))
|
||||
)
|
||||
raise RuntimeError(
|
||||
"Could not print message of this length :"
|
||||
+ str(len(clean_msg))
|
||||
+ ", needs to be below 4096 caracters long."
|
||||
)
|
||||
self.app.logger.warning("Could not print message of this length: " + str(len(clean_msg)))
|
||||
raise Exception("Could not print message of this length :" + str(len(clean_msg)) + ", needs to be below 4096 caracters long.")
|
||||
|
||||
if len(signature) > 256:
|
||||
self.app.logger.warning(
|
||||
"Could not print signature of this length: " + str(len(clean_signature))
|
||||
)
|
||||
raise RuntimeError(
|
||||
"Could not print signature of this length :"
|
||||
+ str(len(clean_signature))
|
||||
+ ", needs to be below 256 caracters long."
|
||||
)
|
||||
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 be below 256 caracters long.")
|
||||
|
||||
# Do the actual printing
|
||||
# We would pop the next element in the queue here, if it's a sms type
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.set(align="center", font="a", bold=bold)
|
||||
self.printer.textln(clean_msg)
|
||||
self.printer.open(self.usb_args);
|
||||
self.printer.set(align='center', font='a', bold=bold)
|
||||
self.printer.textln(clean_msg )
|
||||
if clean_signature:
|
||||
self.printer.textln(clean_signature)
|
||||
self.printer.textln()
|
||||
self.printer.textln(clean_signature )
|
||||
self.printer.close()
|
||||
except Exception as e:
|
||||
self.app.logger.error("Unable to print because : " + str(e))
|
||||
raise RuntimeError(
|
||||
"Unable to print a SMS, the printer couldn't do it."
|
||||
) from e
|
||||
|
||||
self.app.logger.info("Printed text")
|
||||
self.ready = True
|
||||
return True
|
||||
|
||||
def _print_img(self, path, signature="", center=True, process=False):
|
||||
self.ready = False
|
||||
clean_signature = str(signature)
|
||||
def print_img(self, path, sign="",center=True,process=False):
|
||||
clean_signature = str(sign)
|
||||
|
||||
if len(signature) > 256:
|
||||
self.app.logger.warning(
|
||||
"Could not print signature of this length: " + str(len(clean_signature))
|
||||
)
|
||||
raise ValueError(
|
||||
"Could not print signature of this length :"
|
||||
+ str(len(clean_signature))
|
||||
+ ", needs to be below 256 caracters long."
|
||||
)
|
||||
if len(sign) > 256:
|
||||
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 be below 256 caracters long.")
|
||||
|
||||
if not os.path.isfile(str(path)):
|
||||
self.app.logger.warning("File does not exist : " + str(path))
|
||||
raise OSError(
|
||||
"The file path for this image :"
|
||||
+ str(path)
|
||||
+ " wasn't found. Please try again."
|
||||
)
|
||||
|
||||
self.app.logger.debug("Printing file from " + str(path))
|
||||
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))
|
||||
|
||||
if process:
|
||||
try:
|
||||
self.app.logger.debug("Proccessing the image")
|
||||
processed_path = _process_image(self, path)
|
||||
except RuntimeError as e:
|
||||
self.app.logger.error(
|
||||
"Error while processing the image, aborting print : %s", str(e)
|
||||
)
|
||||
raise e
|
||||
path = process_image(self, path)
|
||||
except Exception as e:
|
||||
self.app.logger.error(str(e))
|
||||
return False
|
||||
else:
|
||||
processed_path = path
|
||||
self.app.logger.warning("Not proccessing the image")
|
||||
|
||||
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.image(processed_path, center=center)
|
||||
self.printer.textln(signature)
|
||||
self.printer.image(path,center=center)
|
||||
self.printer.close()
|
||||
self.app.logger.debug("Printed an image : " + str(processed_path))
|
||||
except Exception as e:
|
||||
self.app.logger.error(str(e))
|
||||
raise RuntimeError("Could not print the picture") from e
|
||||
finally:
|
||||
try:
|
||||
os.remove(path)
|
||||
os.remove(processed_path)
|
||||
except OSError as e:
|
||||
raise e
|
||||
|
||||
self.app.logger.debug("Removed image : " + str(processed_path))
|
||||
self.app.logger.debug("Printed an image : " + str(path))
|
||||
os.remove(path)
|
||||
self.app.logger.debug("Removed image : " + str(path))
|
||||
|
||||
try:
|
||||
self.printer.close()
|
||||
except Exception as e:
|
||||
self.app.logger.error(
|
||||
"Could not close the printer connexion %s", str(e)
|
||||
)
|
||||
raise RuntimeError("Could not close the printer connexion. ") from e
|
||||
except Exception as e:
|
||||
self.printer.close()
|
||||
self.app.logger.error(str(e))
|
||||
return False
|
||||
|
||||
self.app.logger.info("Printed a picture")
|
||||
self.ready = True
|
||||
return True
|
||||
|
||||
def _qr(self, content):
|
||||
self.ready = False
|
||||
def qr(self, content):
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.qr(content, center=True)
|
||||
self.printer.textln(content)
|
||||
self.printer.close()
|
||||
except RuntimeError as e:
|
||||
except Exception as e:
|
||||
self.printer.close()
|
||||
self.app.logger.error(str(e))
|
||||
raise e
|
||||
return False
|
||||
|
||||
self.app.logger.info("Printed a QR")
|
||||
self.ready = True
|
||||
return True
|
||||
|
||||
def _cut(self):
|
||||
self.ready = False
|
||||
|
||||
|
||||
def cut(self):
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.cut()
|
||||
@@ -320,278 +188,55 @@ class EscPosPrinter(Printer):
|
||||
raise e
|
||||
|
||||
self.app.logger.info("Did a cut")
|
||||
self.ready = True
|
||||
|
||||
def _state(self) -> bool:
|
||||
has_paper = self._has_paper()
|
||||
is_ready = self.ready
|
||||
|
||||
self.app.logger.debug("Has paper : %s " , has_paper )
|
||||
self.app.logger.debug("Ready : %s " , is_ready )
|
||||
|
||||
return is_ready and has_paper # and is_online
|
||||
|
||||
def print_task(self, task_type, data):
|
||||
"""Execute actual print based on task type"""
|
||||
return True
|
||||
|
||||
|
||||
with self._lock:
|
||||
self.app.logger.debug("Acquired lock to start print")
|
||||
|
||||
i_m_ready = self._state()
|
||||
while not i_m_ready:
|
||||
self.app.logger.debug("Waiting for the printer to become ready, current state %s ", str(i_m_ready))
|
||||
i_m_ready = self._state()
|
||||
time.sleep(0.3)
|
||||
|
||||
self.app.logger.debug("Checked state to start printing : %s", self._state())
|
||||
self.ready = False
|
||||
try:
|
||||
self.app.logger.debug("Checking task type")
|
||||
match (task_type.value):
|
||||
case "text":
|
||||
|
||||
self._print_txt(data["txt"], signature=data["sign"])
|
||||
self.ready = True
|
||||
case "image":
|
||||
self._print_img(
|
||||
data["img"], signature=data["sign"], process=data["process"]
|
||||
)
|
||||
self.ready = True
|
||||
case "cut":
|
||||
self._cut()
|
||||
self.ready = True
|
||||
case "qr":
|
||||
self._qr(data["txt"])
|
||||
self.ready = True
|
||||
case _:
|
||||
raise RuntimeError("This task type is not supported")
|
||||
except Exception as e:
|
||||
self.app.logger.debug("Exception occured while printing %s", str(e))
|
||||
self.ready = True
|
||||
raise RuntimeError from e
|
||||
|
||||
|
||||
class BrotherPrinter(Printer):
|
||||
"""
|
||||
Manages connexion and capabilities of a BrotherQL Printer
|
||||
"""
|
||||
|
||||
def __init__(self, app, vendor_id, device_id):
|
||||
super().__init__(
|
||||
app, vendor_id="", device_id="", printer_type=PrinterType.BROTHER
|
||||
)
|
||||
self.printer = None
|
||||
self.usb_args = {}
|
||||
self.usb_args["idVendor"] = self.device_id
|
||||
self.usb_args["idProduct"] = self.vendor_id
|
||||
self.model_manager = ModelsManager()
|
||||
|
||||
# Code taken from https://github.com/5shekel/printit/blob/master/printer_utils.py
|
||||
|
||||
backend = backend_factory("pyusb")
|
||||
available_devices = backend["list_available_devices"]()
|
||||
|
||||
for printer in available_devices:
|
||||
self.app.logger.debug(f"Found device: {printer}")
|
||||
identifier = printer["identifier"]
|
||||
parts = identifier.split("/")
|
||||
|
||||
if len(parts) < 4:
|
||||
self.app.logger.warning(
|
||||
f"Skipping device with invalid identifier format: {identifier}"
|
||||
)
|
||||
continue
|
||||
|
||||
protocol = parts[0]
|
||||
# device_info = parts[2]
|
||||
serial_number = parts[3]
|
||||
|
||||
try:
|
||||
product_id_int = int(self.device_id, 16)
|
||||
for m in self.model_manager.iter_elements():
|
||||
if m.product_id == product_id_int:
|
||||
model = m.identifier
|
||||
break
|
||||
self.app.logger.debug(f"Matched printer model: {model}")
|
||||
except ValueError:
|
||||
self.app.logger.warning(f"Invalid product ID format: {m.product_id}")
|
||||
|
||||
self.printer_info = PrinterInfo(
|
||||
identifier=identifier,
|
||||
backend="pyusb",
|
||||
model=model,
|
||||
protocol=protocol,
|
||||
vendor_id=vendor_id,
|
||||
product_id=self.device_id,
|
||||
serial_number=serial_number,
|
||||
)
|
||||
|
||||
self.ready = True
|
||||
|
||||
def _has_paper(self):
|
||||
raise NotImplementedError("This printer model does not support this.")
|
||||
|
||||
def _state(self):
|
||||
return self.ready
|
||||
|
||||
def _print_img(self, data):
|
||||
"""
|
||||
Print a raster image via a Brother QL printer
|
||||
"""
|
||||
self.ready = False
|
||||
label_type = "102"
|
||||
rotate = 0
|
||||
dither = False
|
||||
try:
|
||||
# Prepare the image for printing
|
||||
qlr = BrotherQLRaster(self.printer_info["model"])
|
||||
|
||||
instructions = convert(
|
||||
qlr=qlr,
|
||||
images=[data["img"]],
|
||||
label=label_type,
|
||||
rotate=rotate,
|
||||
threshold=70,
|
||||
dither=dither,
|
||||
compress=True,
|
||||
red=False,
|
||||
dpi_600=False,
|
||||
hq=False,
|
||||
cut=True,
|
||||
)
|
||||
|
||||
# Debug logging
|
||||
if os.getenv("FLASK_DEBUG"):
|
||||
self.app.logger.debug(f"""
|
||||
Print parameters:
|
||||
- Label type: {label_type}
|
||||
- Rotate: {rotate}
|
||||
- Dither: {dither}
|
||||
- Model: {self.printer_info['model']}
|
||||
- Backend: {self.printer_info['backend']}
|
||||
- Identifier: {self.printer_info['identifier']}
|
||||
""")
|
||||
|
||||
# Try to print using Python API
|
||||
# send() = status = {
|
||||
# 'instructions_sent': True, # The instructions were sent to the printer.
|
||||
# 'outcome': 'unknown', # String description of the outcome of the sending operation like: 'unknown', 'sent', 'printed', 'error'
|
||||
# 'printer_state': None, # If the selected backend supports reading back the printer state, this key will contain it.
|
||||
# 'did_print': False, # If True, a print was produced. It defaults to False if the outcome is uncertain (due to a backend without read-back capability).
|
||||
# 'ready_for_next_job': False, # If True, the printer is ready to receive the next instructions. It defaults to False if the state is unknown.
|
||||
# }
|
||||
status = send(
|
||||
instructions=instructions,
|
||||
printer_identifier=self.printer_info["identifier"],
|
||||
backend_identifier="pyusb",
|
||||
)
|
||||
|
||||
if (
|
||||
not status["did_print"]
|
||||
or status["outcome"] == "error"
|
||||
or status["outcome"] == "unknown"
|
||||
):
|
||||
raise RuntimeError("Failed to print using Python API")
|
||||
|
||||
if status["printer_state"]:
|
||||
self.ready = bool(status["printer_state"])
|
||||
else:
|
||||
self.ready = True
|
||||
|
||||
except usb.core.USBError as e:
|
||||
# Treat timeout errors as successful since they often occur after print completion
|
||||
if e.errno == 110: # Operation timed out
|
||||
self.app.logger.debug(
|
||||
"USB timeout occurred - this is normal and the print likely completed"
|
||||
)
|
||||
self.app.logger.debug("Print completed (timeout is normal)")
|
||||
self.ready = True
|
||||
|
||||
error_msg = f"USBError encountered: {e}"
|
||||
self.app.logger.debug(error_msg)
|
||||
raise RuntimeError from e
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Unexpected error during printing: {str(e)}"
|
||||
self.app.logger.debug(error_msg)
|
||||
raise RuntimeError from e
|
||||
|
||||
def print_task(self, task_type, data):
|
||||
"""Execute actual print based on task type"""
|
||||
with self._lock:
|
||||
if self._state:
|
||||
self._state = False
|
||||
match (task_type.value):
|
||||
case "image":
|
||||
self._print_img(data["img"])
|
||||
self._state = True
|
||||
case "cut":
|
||||
# The cut happens by default on Brother QL printers.
|
||||
self._state = True
|
||||
case _:
|
||||
raise RuntimeError("This task type is not supported")
|
||||
else:
|
||||
raise RuntimeError("The printer is not ready to print yet !")
|
||||
|
||||
# These values are by default for now
|
||||
|
||||
# raise NotImplementedError("This printer type is not implemented yet")
|
||||
|
||||
|
||||
def _process_image(self, path):
|
||||
def process_image(self, path):
|
||||
brightness_factor = 1.5 # Used only if image is too dark
|
||||
brightness_threshold = 100 # Brightness threshold (0–255)
|
||||
contrast_factor = 2 # Less than 1.0 = lower contrast
|
||||
contrast_factor = 0.6 # Less than 1.0 = lower contrast
|
||||
max_width = 575
|
||||
max_height = 1000
|
||||
|
||||
with Image.open(path) as original_img:
|
||||
# Convert to RGB if needed (JPEG doesn't support alpha)
|
||||
if original_img.mode in ("RGBA", "P"):
|
||||
self.app.logger.debug("Converting the image from RGBA to RGBA")
|
||||
self.app.logger.debug("Converting the image to RGB from RGBA")
|
||||
original_img = original_img.convert("RGB")
|
||||
|
||||
# Resize while maintaining aspect ratio
|
||||
original_img.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
|
||||
original_img.thumbnail((max_width, max_height), Image.LANCZOS)
|
||||
self.app.logger.debug("Resized the image")
|
||||
|
||||
# # Convert to grayscale for dithering
|
||||
# dithered_img = original_img.convert("L").convert("1")
|
||||
# Dithering using default method (Floyd–Steinberg)
|
||||
# dithered_img = original_img.convert("L").convert("1") # Dithering using default method (Floyd–Steinberg)
|
||||
# self.app.logger.debug("Dithered the image")
|
||||
|
||||
# Compute brightness of original image (grayscale average)
|
||||
grayscale = original_img.convert("L")
|
||||
avg_brightness = np.array(grayscale).mean()
|
||||
self.app.logger.debug(
|
||||
"Average brightness of the image : " + str(avg_brightness)
|
||||
)
|
||||
self.app.logger.debug("Average brightness of the image : " + str(avg_brightness) )
|
||||
|
||||
# Dynamically compute brightness factor if too dark
|
||||
if avg_brightness < brightness_threshold:
|
||||
brightness_factor = (
|
||||
1 + (brightness_threshold - avg_brightness) / brightness_threshold
|
||||
)
|
||||
brightness_factor = min(
|
||||
max(brightness_factor, 1.1), 2.5
|
||||
) # Clamp between 1.1 and 2.5
|
||||
self.app.logger.debug(
|
||||
f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}"
|
||||
)
|
||||
enhancer = ImageEnhance.Brightness(grayscale)
|
||||
grayscale = enhancer.enhance(brightness_factor)
|
||||
brightness_factor = 1 + (brightness_threshold - avg_brightness) / brightness_threshold
|
||||
brightness_factor = min(max(brightness_factor, 1.1), 2.5) # Clamp between 1.1 and 2.5
|
||||
self.app.logger.debug(f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}")
|
||||
enhancer = ImageEnhance.Brightness(original_img)
|
||||
original_img = enhancer.enhance(brightness_factor)
|
||||
|
||||
# Computer current contrast of grayscale image
|
||||
contrast = np.clip(np.std(np.array(grayscale)), 0, 255)
|
||||
self.app.logger.debug("Standard deviation of the contrast : %s", contrast)
|
||||
# # Enhance contrast
|
||||
contrast_enhancer = ImageEnhance.Contrast(grayscale)
|
||||
original_img = contrast_enhancer.enhance(contrast_factor)
|
||||
# # Reduce contrast
|
||||
# contrast_enhancer = ImageEnhance.Contrast(original_img)
|
||||
# original_img = contrast_enhancer.enhance(contrast_factor)
|
||||
|
||||
# Final resize check
|
||||
if original_img.height > max_height:
|
||||
raise ValueError("Image is too long, sorry! Keep it below 575×1000 pixels.")
|
||||
self.app.logger.error("Image is too long, sorry! Keep it below 575×1000 pixels.")
|
||||
return False
|
||||
|
||||
# Convert to JPEG and save
|
||||
jpeg_path = os.path.splitext(path)[0] + "_processed.jpg"
|
||||
grayscale.save(jpeg_path, format="JPEG", quality=95, optimize=True)
|
||||
self.app.logger.debug("Processed and saved image.")
|
||||
original_img.save(jpeg_path, format='JPEG', quality=95, optimize=True)
|
||||
|
||||
return jpeg_path
|
||||
|
||||
160
src/printers.py
160
src/printers.py
@@ -1,160 +0,0 @@
|
||||
"""
|
||||
A collection of Printers.
|
||||
|
||||
It has methods to discover printers, and provides an interface for
|
||||
the methods expected from printers.
|
||||
"""
|
||||
|
||||
from collections.abc import Set
|
||||
import usb.core
|
||||
import usb.util
|
||||
from printer import Printer, EscPosPrinter, BrotherPrinter
|
||||
|
||||
|
||||
class Printers:
|
||||
"""
|
||||
Finds and creates a set of Printer that can be used by the Workers to print.
|
||||
"""
|
||||
|
||||
def __init__(self, app):
|
||||
"""
|
||||
Discover printers connected to the computer and return a Collection of Printer()
|
||||
"""
|
||||
self.app = app
|
||||
self.printers = self._discover_printers()
|
||||
|
||||
def _discover_printers(self) -> Set[Printer]:
|
||||
"""
|
||||
Gets connected USB printer devices using the pyusb library.
|
||||
|
||||
We analyse the USB devices, get the ones that match
|
||||
the printer class ( 7 ) and for Brother and EPSON printers,
|
||||
try to create a Printer object that can be used by the Worker class
|
||||
to execute prints.
|
||||
|
||||
Returns a set of Printer
|
||||
"""
|
||||
|
||||
self.app.logger.debug("Discovering USB Devices connected to this system")
|
||||
|
||||
printers = set()
|
||||
|
||||
# Find all connected USB devices
|
||||
devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
|
||||
if not devices:
|
||||
self.app.logger.warning(
|
||||
"No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
|
||||
)
|
||||
raise RuntimeError(
|
||||
"No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
|
||||
)
|
||||
|
||||
for dev in devices:
|
||||
# Attempt to get the manufacturer and product strings
|
||||
try:
|
||||
manufacturer = usb.util.get_string(dev, dev.iManufacturer)
|
||||
except Exception:
|
||||
manufacturer = "Unknown"
|
||||
|
||||
try:
|
||||
product = usb.util.get_string(dev, dev.iProduct)
|
||||
except Exception:
|
||||
product = "Unknown"
|
||||
self.app.logger.debug(
|
||||
"Looking at %s %s (%s:%s)",
|
||||
manufacturer,
|
||||
product,
|
||||
hex(dev.idVendor),
|
||||
hex(dev.idProduct),
|
||||
)
|
||||
|
||||
if manufacturer == "EPSON":
|
||||
try:
|
||||
# We create a new EscPosPrinter()
|
||||
self.app.logger.debug("Trying to creat a new EPSON printer")
|
||||
prid = dev.idProduct
|
||||
vendir = dev.idVendor
|
||||
escpos_printer = EscPosPrinter(
|
||||
self.app, vendor_id=vendir, device_id=prid
|
||||
)
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
# If the object creation is successfull, we add it to the list of Printers
|
||||
printers.add(escpos_printer)
|
||||
self.app.logger.debug("Found a %s printer", manufacturer)
|
||||
|
||||
# We already found the type of printer,
|
||||
# we don't need an extra comparaison.
|
||||
continue
|
||||
|
||||
# or a Brother Printer
|
||||
if manufacturer == "Brother":
|
||||
try:
|
||||
# We create a new BrotherPrinter()
|
||||
self.app.logger.debug("Trying to creat a new BROTHER printer")
|
||||
prid = dev.idProduct
|
||||
vendir = dev.idVendor
|
||||
brother_printer = BrotherPrinter(
|
||||
self.app, vendor_id=vendir, device_id=prid
|
||||
)
|
||||
except Exception as e:
|
||||
self.app.logger.error(
|
||||
"Could not create a %s printer class with %s:%s" , product,
|
||||
dev.idVendor,
|
||||
dev.idProduct,
|
||||
)
|
||||
raise e
|
||||
|
||||
# If the object creation is successfull, we add it to the list of Printers
|
||||
printers.add(brother_printer)
|
||||
self.app.logger.debug("Found a %s printer" , manufacturer)
|
||||
|
||||
self.app.logger.debug("Found %s printers" , len(printers))
|
||||
|
||||
if len(printers) < 1:
|
||||
self.app.logger.warning("Not printers found ! Please plug in a Printer and restart the program.")
|
||||
raise RuntimeError("No printers found")
|
||||
|
||||
return printers
|
||||
|
||||
def any(self) -> Printer:
|
||||
"""
|
||||
Return a dict key: UUID, value: Printer, with any connected printer.
|
||||
"""
|
||||
if len(self.printers) > 0:
|
||||
for i in self.printers:
|
||||
return i
|
||||
else:
|
||||
raise RuntimeError("No printers available")
|
||||
|
||||
# def get_printer(self, printer_type):
|
||||
# """
|
||||
# Return a specific printer
|
||||
|
||||
# printer_type -- a printer type
|
||||
# """
|
||||
# return NotImplementedError()
|
||||
|
||||
|
||||
class _FindClass:
|
||||
"""
|
||||
Use by usb.core to modify the way USB devices are found
|
||||
Taken from pyUSB documentation on Github
|
||||
"""
|
||||
def __init__(self, class_):
|
||||
self._class = class_
|
||||
|
||||
def __call__(self, device):
|
||||
# first, let's check the device
|
||||
if device.bDeviceClass == self._class:
|
||||
return True
|
||||
# ok, transverse all devices to find an
|
||||
# interface that matches our class
|
||||
for cfg in device:
|
||||
# find_descriptor: what's it?
|
||||
intf = usb.util.find_descriptor(cfg, bInterfaceClass=self._class)
|
||||
if intf is not None:
|
||||
return True
|
||||
|
||||
return False
|
||||
218
src/raspberry.py
218
src/raspberry.py
@@ -1,21 +1,12 @@
|
||||
"""
|
||||
This class executes when we are on a raspberry Pi.
|
||||
|
||||
It handles the press of a button via GPIO,
|
||||
activates a flash and prints out the picture
|
||||
that was taken via a USB Webcam.
|
||||
"""
|
||||
|
||||
import io # To check if we are on a Raspberry Pi
|
||||
from flask_socketio import SocketIO
|
||||
from gpiozero import Button, LED, DigitalOutputDevice
|
||||
from time import sleep, gmtime, strftime
|
||||
from PIL import Image
|
||||
import io # To check if we are on a Raspberry Pi
|
||||
import subprocess
|
||||
import os
|
||||
from time import sleep, gmtime, strftime
|
||||
from gpiozero import Button, LED, DigitalOutputDevice
|
||||
from PIL import Image
|
||||
from task import TextTask, ImageTask, CutTask
|
||||
|
||||
|
||||
class Raspberry:
|
||||
class Raspberry(object):
|
||||
"""
|
||||
This class will manage three things :
|
||||
- Connecting to a USB webcam
|
||||
@@ -23,103 +14,78 @@ class Raspberry:
|
||||
- Activating a flash ( or light )
|
||||
- Flash an indicator light
|
||||
|
||||
# pylint: disable=too-many-instance-attributes
|
||||
# dede
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
print_queue,
|
||||
app,
|
||||
configuration_file
|
||||
):
|
||||
self.print_queue = print_queue
|
||||
def __init__(self, printer, app, socketio, button_gpio_port_number, indicator_gpio_port_number, flash_gpio_port_number, is_flash_present,):
|
||||
self.printer = printer
|
||||
self.socketio = socketio
|
||||
self.app = app
|
||||
self.configuration_file = configuration_file
|
||||
self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
|
||||
|
||||
def is_raspberry_pi(self):
|
||||
"""
|
||||
Checking if we are on a Raspberry Pi by checking
|
||||
information on the /proc/cpuinfo file
|
||||
"""
|
||||
self.flash_gpio = flash_gpio_port_number
|
||||
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 is_raspberry_pi(self, raise_on_errors=False):
|
||||
# Check if we are running on a raspberry pi
|
||||
try:
|
||||
with io.open("/proc/cpuinfo", "r", encoding="utf-8") as cpuinfo:
|
||||
with io.open('/proc/cpuinfo', 'r') as cpuinfo:
|
||||
found = False
|
||||
for line in cpuinfo:
|
||||
if line.startswith("Hardware"):
|
||||
if line.startswith('Hardware'):
|
||||
found = True
|
||||
label, value = line.strip().split(":", 1)
|
||||
label, value = line.strip().split(':', 1)
|
||||
value = value.strip()
|
||||
if value not in (
|
||||
"BCM2708",
|
||||
"BCM2709",
|
||||
"BCM2711",
|
||||
"BCM2835",
|
||||
"BCM2836",
|
||||
'BCM2708',
|
||||
'BCM2709',
|
||||
'BCM2711',
|
||||
'BCM2835',
|
||||
'BCM2836'
|
||||
):
|
||||
self.app.logger.debug(
|
||||
"This system does not appear to be a Raspberry Pi."
|
||||
)
|
||||
self.app.logger.debug('This system does not appear to be a Raspberry Pi.')
|
||||
return False
|
||||
|
||||
if not found:
|
||||
self.app.logger.warning(
|
||||
"Couldn't get sufficient hardware information from /proc/cpuinfo"
|
||||
)
|
||||
self.app.logger.warning("Unable to determine if we are on a Raspberry Pi.")
|
||||
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`.")
|
||||
self.app.logger.error('Unable to open `/proc/cpuinfo`.')
|
||||
return False
|
||||
|
||||
self.app.logger.debug("It seems we are on a Raspberry Pi")
|
||||
self.app.logger.debug('It seems we are on a Raspberry Pi')
|
||||
|
||||
try:
|
||||
self._initialise_gpio()
|
||||
self.initialise_gpio()
|
||||
except Exception as e:
|
||||
self.app.logger.debug("Could not init GPIO : " + str(e))
|
||||
self.app.logger.debug('Could not init GPIO : ' + str(e))
|
||||
raise e
|
||||
return True
|
||||
|
||||
def _initialise_gpio(self):
|
||||
"""
|
||||
Set GPIO ports from configuration and activate them
|
||||
to show the user it's working.
|
||||
"""
|
||||
self.app.logger.debug("Initializing GPIO")
|
||||
def initialise_gpio(self):
|
||||
self.app.logger.debug('Initializing GPIO')
|
||||
|
||||
self.led = LED(self.configuration_file["rpi"]["indicator_gpio_port_number"])
|
||||
self.app.logger.debug("Activated indicator LED")
|
||||
self.led = LED(self.led_gpio)
|
||||
self.app.logger.debug('Activated indicator LED')
|
||||
self.indicator_countdown(iters=3)
|
||||
self.button = Button(
|
||||
self.configuration_file["rpi"]["button_gpio_port_number"],
|
||||
pull_up=True, bounce_time=0.1
|
||||
)
|
||||
self.button = Button(self.button_gpio, pull_up=True, bounce_time=0.1)
|
||||
self.button.when_pressed = self.on_button_pressed
|
||||
self.app.logger.debug("Activated button")
|
||||
self.app.logger.debug('Activated button')
|
||||
|
||||
# The "flash" is a relay-controlled device ( light bulb for example )
|
||||
self.flash = DigitalOutputDevice(self.configuration_file["rpi"]["flash_gpio_port_number"])
|
||||
self.flash = DigitalOutputDevice(self.flash_gpio)
|
||||
self.flash_toggle()
|
||||
self.app.logger.debug("Activated flash")
|
||||
self.app.logger.debug('Activated flash')
|
||||
|
||||
def indicator_countdown(self, iters=10, multi=10):
|
||||
"""
|
||||
Activates the LED faster and faster to show a countdown
|
||||
"""
|
||||
for i in range(iters, 0, -1):
|
||||
def indicator_countdown(self,iters=10,multi=10):
|
||||
for i in range(iters,0,-1):
|
||||
self.led.on()
|
||||
sleep(i / multi)
|
||||
sleep(i/multi)
|
||||
self.led.off()
|
||||
sleep(i / multi)
|
||||
sleep(i/multi)
|
||||
|
||||
def indicator_led(self, timing=0.2, l=5):
|
||||
"""
|
||||
Turns on the indicator LED for a certain period of time
|
||||
"""
|
||||
for _ in range(l):
|
||||
def indicator_led(self,timing=0.2,l=5):
|
||||
for i in range(l):
|
||||
self.app.logger.debug("LED turned on")
|
||||
self.led.on()
|
||||
sleep(timing)
|
||||
@@ -128,9 +94,6 @@ class Raspberry:
|
||||
sleep(timing)
|
||||
|
||||
def flash_toggle(self):
|
||||
"""
|
||||
Flashes the flash
|
||||
"""
|
||||
self.app.logger.debug("Flash turned on")
|
||||
self.flash.on()
|
||||
sleep(0.3)
|
||||
@@ -138,35 +101,26 @@ class Raspberry:
|
||||
self.app.logger.debug("Flash turned off")
|
||||
|
||||
def take_picture(self):
|
||||
"""
|
||||
Takes a picture via the USB webcam
|
||||
"""
|
||||
# Validate if the image path is valid
|
||||
if not os.path.isdir(os.path.dirname(self.image_path)):
|
||||
self.app.logger.error(
|
||||
f"Invalid directory for image path: {self.image_path}"
|
||||
)
|
||||
self.app.logger.error(f"Invalid directory for image path: {self.image_path}")
|
||||
return False
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["fswebcam", "--no-banner", "-r", "1920x1080", self.image_path],
|
||||
['fswebcam', '--no-banner', '-r', '1920x1080', self.image_path],
|
||||
check=True, # Will raise CalledProcessError if the command fails
|
||||
stdout=subprocess.PIPE, # Capture standard output
|
||||
stderr=subprocess.PIPE, # Capture error output
|
||||
stderr=subprocess.PIPE # Capture error output
|
||||
)
|
||||
|
||||
# Optionally log the command output
|
||||
self.app.logger.debug(
|
||||
f"Image captured successfully: {result.stdout.decode()}"
|
||||
)
|
||||
self.app.logger.debug(f"Image captured successfully: {result.stdout.decode()}")
|
||||
except subprocess.CalledProcessError as e:
|
||||
# Log error output if available
|
||||
self.app.logger.error(
|
||||
f"Unable to take a picture. Error: {e.stderr.decode()}"
|
||||
)
|
||||
self.app.logger.error(f"Unable to take a picture. Error: {e.stderr.decode()}")
|
||||
return False
|
||||
except RuntimeError as e:
|
||||
except Exception as e:
|
||||
# Catch any unexpected errors
|
||||
self.app.logger.error(f"Unexpected error while taking picture: {str(e)}")
|
||||
return False
|
||||
@@ -178,47 +132,32 @@ class Raspberry:
|
||||
|
||||
return True
|
||||
|
||||
def overlay_logo(
|
||||
self,
|
||||
image_path,
|
||||
logo_path,
|
||||
output_path=None,
|
||||
position="bottom_right",
|
||||
margin=10,
|
||||
):
|
||||
"""
|
||||
Takes an image and overlays it with a another picture.
|
||||
"""
|
||||
|
||||
def overlay_logo(self, image_path, logo_path, output_path=None, position='bottom_right', margin=10):
|
||||
try:
|
||||
image = Image.open(image_path).convert("RGBA")
|
||||
logo = Image.open(logo_path).convert("RGBA")
|
||||
|
||||
# Resize logo if it's too big (logo will be 30% the width of the image)
|
||||
logo_ratio = (
|
||||
0.30 # You can change the ratio if you want the logo bigger or smaller
|
||||
)
|
||||
logo_ratio = 0.30 # You can change the ratio if you want the logo bigger or smaller
|
||||
logo_width = int(image.width * logo_ratio)
|
||||
logo_height = int(logo.height * (logo_width / logo.width))
|
||||
logo = logo.resize((logo_width, logo_height), Image.Resampling.LANCZOS)
|
||||
|
||||
# Calculate position based on the chosen location
|
||||
if position == "bottom_right":
|
||||
if position == 'bottom_right':
|
||||
x = image.width - logo.width - margin
|
||||
y = image.height - logo.height - margin
|
||||
elif position == "top_left":
|
||||
elif position == 'top_left':
|
||||
x, y = margin, margin
|
||||
elif position == "top_right":
|
||||
elif position == 'top_right':
|
||||
x = image.width - logo.width - margin
|
||||
y = margin
|
||||
elif position == "bottom_left":
|
||||
elif position == 'bottom_left':
|
||||
x = margin
|
||||
y = image.height - logo.height - margin
|
||||
else:
|
||||
raise ValueError(
|
||||
"Invalid position." +
|
||||
"Choose from 'bottom_right', 'top_left', " +
|
||||
" 'top_right', or 'bottom_left'."
|
||||
)
|
||||
raise ValueError("Invalid position. Choose from 'bottom_right', 'top_left', 'top_right', or 'bottom_left'.")
|
||||
|
||||
# Composite the logo onto the image
|
||||
image.paste(logo, (x, y), logo) # Use logo as its own alpha mask
|
||||
@@ -228,16 +167,13 @@ class Raspberry:
|
||||
output_path = image_path # Overwrite the original image if no output path is given
|
||||
image.save(output_path)
|
||||
|
||||
except RuntimeError as e:
|
||||
except Exception as e:
|
||||
self.app.logger.error(f"Error overlaying logo: {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def crop_to_square(self, image_path, output_path=None):
|
||||
"""
|
||||
Crop an image so that it becomes a square
|
||||
"""
|
||||
try:
|
||||
image = Image.open(image_path)
|
||||
width, height = image.size
|
||||
@@ -258,43 +194,37 @@ class Raspberry:
|
||||
|
||||
image.save(output_path)
|
||||
|
||||
except RuntimeError as e:
|
||||
except Exception as e:
|
||||
self.app.logger.error(f"Error cropping image to square: {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def on_button_pressed(self):
|
||||
"""
|
||||
When a button press is detected, a picture is taken from the webcam
|
||||
and added to the print queue.
|
||||
"""
|
||||
self.app.logger.debug("Button has been pressed")
|
||||
self.led.on()
|
||||
self.app.logger.debug("Counting down")
|
||||
self.indicator_countdown(
|
||||
iters=4, multi=20
|
||||
) # The indicator will flash a countdown LED
|
||||
self.indicator_countdown(iters=4,multi=20) # The indicator will flash a countdown LED
|
||||
self.app.logger.debug("Taking picture")
|
||||
try:
|
||||
self.flash.on()
|
||||
self.take_picture()
|
||||
except RuntimeError as e:
|
||||
self.app.logger.error(
|
||||
"Could not take a picture after the button press : " + str(e)
|
||||
)
|
||||
except Exception as e:
|
||||
self.app.logger.error("Could not take a picture after the button press : " + str(e))
|
||||
finally:
|
||||
self.flash.off()
|
||||
self.app.logger.debug("Printing picture")
|
||||
self.led.on()
|
||||
self.crop_to_square(self.image_path)
|
||||
self.print_queue.enqueue(ImageTask(self.image_path, signature="", process=True))
|
||||
self.print_queue.enqueue(
|
||||
TextTask(content="Imprimé par LittlePrynter", signature="")
|
||||
)
|
||||
time = strftime("%Y-%m-%d %H:%M", gmtime())
|
||||
self.print_queue.enqueue(TextTask(content=time, signature=""))
|
||||
self.print_queue.enqueue(CutTask())
|
||||
self.printer.print_img('src/static/images/extase-club.png',process=True)
|
||||
self.printer.print_img(self.image_path,process=True)
|
||||
self.printer.print_sms("")
|
||||
self.printer.print_sms("With Love From Société.Vide",signature="",bold=True)
|
||||
self.printer.print_sms("Printed by LittlePrynter",signature="")
|
||||
self.printer.print_sms("n07070.xyz",signature="")
|
||||
self.printer.print_sms(strftime("%Y-%m-%d %H:%M", gmtime()),signature="")
|
||||
self.printer.qr("https://n07070.xyz/articles/littleprynter")
|
||||
self.printer.cut()
|
||||
self.led.off()
|
||||
self.app.logger.debug("Added a photomaton picture to the print queue")
|
||||
return True
|
||||
self.app.logger.debug("Done printing picture")
|
||||
return True
|
||||
@@ -3,36 +3,6 @@
|
||||
margin-right: 20%;
|
||||
} */
|
||||
|
||||
html, body {
|
||||
height: 100%;
|
||||
width: 100%;
|
||||
}
|
||||
|
||||
.fullscreen-countdown {
|
||||
position: fixed;
|
||||
top: 0;
|
||||
left: 0;
|
||||
right: 0;
|
||||
bottom: 0;
|
||||
z-index: 100;
|
||||
height: 100%;
|
||||
width: 100%;
|
||||
/* backdrop-filter: blur(10px); */
|
||||
color: black;
|
||||
box-sizing: initial;
|
||||
|
||||
}
|
||||
|
||||
#countdown-number {
|
||||
font-weight: bolder;
|
||||
font-size: 10em;
|
||||
text-shadow: 0px 0px 3em white, 1px 1px 1px rgba(255, 255, 255, 1);
|
||||
position: fixed;
|
||||
top: 50%;
|
||||
left: 50%;
|
||||
transform: translate(-50%, -50%);
|
||||
}
|
||||
|
||||
/*// Small devices (landscape phones, 576px and up)*/
|
||||
@media (min-width: 576px) {
|
||||
|
||||
|
||||
@@ -186,18 +186,18 @@ function setup_events(){
|
||||
}, false);
|
||||
|
||||
switch_cameras.addEventListener('click', function(ev) {
|
||||
flip_cameras();
|
||||
flip_cameras(camera_options);
|
||||
}, false );
|
||||
|
||||
printButton.addEventListener('click', async function(ev) {
|
||||
ev.preventDefault();
|
||||
printButton.addEventListener('click', function(ev){
|
||||
data = take_picture();
|
||||
try {
|
||||
await countDownToPicture(); // wait for countdown to finish
|
||||
var data = await take_picture(); // take_picture can be async
|
||||
print_picture(data);
|
||||
} catch (e) {
|
||||
alert("Failed to print a picture because : " + e);
|
||||
}
|
||||
|
||||
ev.preventDefault();
|
||||
}, false);
|
||||
|
||||
}
|
||||
@@ -212,10 +212,6 @@ function clear_canvas() {
|
||||
}
|
||||
|
||||
function dataURLtoFile(dataurl, filename) {
|
||||
if(dataurl === "") {
|
||||
console.log("We didn't receive data");
|
||||
}
|
||||
|
||||
var arr = dataurl.split(','), mime = arr[0].match(/:(.*?);/)[1],
|
||||
bstr = atob(arr[1]), n = bstr.length, u8arr = new Uint8Array(n);
|
||||
while(n--){
|
||||
@@ -224,18 +220,19 @@ function dataURLtoFile(dataurl, filename) {
|
||||
return new File([u8arr], filename, {type:mime});
|
||||
}
|
||||
|
||||
async function take_picture() {
|
||||
const context = canvas.getContext('2d');
|
||||
let data = null;
|
||||
function take_picture(){
|
||||
var context = canvas.getContext('2d');
|
||||
if (width && height) {
|
||||
canvas.width = width;
|
||||
canvas.height = height;
|
||||
context.drawImage(video, 0, 0, width, height);
|
||||
data = canvas.toDataURL('image/png');
|
||||
photo.setAttribute('src', data);
|
||||
canvas.width = width;
|
||||
canvas.height = height;
|
||||
context.drawImage(video, 0, 0, width, height);
|
||||
|
||||
var data = canvas.toDataURL('image/png');
|
||||
photo.setAttribute('src', data);
|
||||
} else {
|
||||
clear_canvas();
|
||||
clear_canvas();
|
||||
}
|
||||
|
||||
return data;
|
||||
}
|
||||
|
||||
@@ -255,28 +252,15 @@ function print_picture(data){
|
||||
// headers:{
|
||||
// 'Content-Type': 'multipart/form-data'
|
||||
// }
|
||||
}).then(function(response) {
|
||||
console.log('Reponse:', response);
|
||||
if(response.status != 200 ){
|
||||
alert("The picture could not be printed be : " + response.statusText)
|
||||
}
|
||||
} , true)
|
||||
}).then(function(response) { console.log('Success:', response); } , true)
|
||||
.catch(error => console.error('Error:', error), false);
|
||||
|
||||
}
|
||||
|
||||
function flip_cameras(){
|
||||
console.log("Current facing mode : " + camera_options.video.facingMode)
|
||||
function flip_cameras(camera_options){
|
||||
if ( supports_facing_mode ) {
|
||||
console.log("Support facing mode, trying to switch !");
|
||||
if (camera_options.video.facingMode == 'user'){
|
||||
camera_options = { audio: false, video: { facingMode: "environment" },};
|
||||
} else {
|
||||
camera_options = { audio: false, video: { facingMode: "user" },};
|
||||
}
|
||||
get_webcam((camera_options.video.facingMode === 'user' ? 'environment' : 'user'));
|
||||
}
|
||||
console.log("Switching to " + camera_options.video.facingMode );
|
||||
get_webcam(camera_options);
|
||||
}
|
||||
|
||||
function stop_video_streams(){
|
||||
@@ -316,33 +300,4 @@ function no_webcam_error(){
|
||||
throw new Error("Unable to get a video device, stopping the photobooth.");
|
||||
}
|
||||
|
||||
async function countDownToPicture(){
|
||||
return new Promise((resolve) => {
|
||||
console.log("Starting countdown");
|
||||
// We create full-page overlay that displays a number to countdown until the picture is taken.
|
||||
const countdown_div = document.getElementById("countdown");
|
||||
countdown_div.style.display = "block";
|
||||
|
||||
// Set the different styling attributes of the element
|
||||
countdown_div.setAttribute("class", "fullscreen-countdown");
|
||||
|
||||
// The loop must run for 3 seconds
|
||||
interval = 1000;
|
||||
var loops = 3;
|
||||
|
||||
var x = setInterval(() => {
|
||||
console.log(loops);
|
||||
// Update the content of the div
|
||||
document.getElementById("countdown-number").innerHTML = loops;
|
||||
if (loops == 0) {
|
||||
countdown_div.style.display = "none";
|
||||
// The timer has finished
|
||||
clearInterval(x);
|
||||
resolve();
|
||||
}
|
||||
loops--;
|
||||
}, interval);
|
||||
});
|
||||
}
|
||||
|
||||
window.addEventListener('load', startup, false);
|
||||
|
||||
110
src/task.py
110
src/task.py
@@ -1,110 +0,0 @@
|
||||
"""
|
||||
Here we define the types of tasks
|
||||
We are using Abstract Base Classes,
|
||||
like this we can define types of tasks ( text, images, ... )
|
||||
that all work with the same basic options
|
||||
|
||||
The tasks are going to be injected into a Queue.
|
||||
It's a usefull way of storing information in our
|
||||
program, while making sure that things are indeed printed.
|
||||
It's also a way to prevent two concurrent connexions creating
|
||||
a access conflict on a single printer, like two people wanting
|
||||
to print at the same time.
|
||||
|
||||
We can also delay and store printing tasks until a printer becomes
|
||||
available if none is online.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
## See https://docs.python.org/3/library/abc.html to learn more about this
|
||||
|
||||
from enum import Enum
|
||||
import uuid
|
||||
|
||||
|
||||
## You can expand this if you want to take other types of tasks into account
|
||||
class TaskType(Enum):
|
||||
"""
|
||||
The different tasks supported by the printers
|
||||
"""
|
||||
|
||||
TEXT = "text"
|
||||
IMAGE = "image"
|
||||
CUT = "cut"
|
||||
QR = "qr"
|
||||
|
||||
|
||||
class PrintTask(ABC):
|
||||
"""
|
||||
A print task holds information about what we are looking to print.
|
||||
"""
|
||||
|
||||
def __init__(self, task_type: TaskType):
|
||||
self.task_id = self._generate_id()
|
||||
self.task_type = task_type
|
||||
self.status = "pending" # pending, processing, completed, failed
|
||||
|
||||
@abstractmethod
|
||||
def get_print_data(self):
|
||||
"""Return data formatted for printer"""
|
||||
|
||||
def _generate_id(self):
|
||||
# Generate unique task ID
|
||||
return str(uuid.uuid4())
|
||||
|
||||
|
||||
class TextTask(PrintTask):
|
||||
"""
|
||||
This tasks represents a texte content, and it's signature.
|
||||
"""
|
||||
|
||||
def __init__(self, content, signature):
|
||||
super().__init__(TaskType.TEXT)
|
||||
self.content = content
|
||||
self.signature = signature
|
||||
|
||||
def get_print_data(self):
|
||||
return {"txt": self.content, "sign": self.signature}
|
||||
|
||||
|
||||
class QRTask(TextTask):
|
||||
"""This task prints a QR-Code, the signature is ignore and is always the content itself"""
|
||||
|
||||
def __init__(self, content):
|
||||
super().__init__(content, signature="")
|
||||
self.content = content
|
||||
self.signature = content
|
||||
|
||||
def get_print_data(self):
|
||||
return {"txt": self.content, "sign": self.signature}
|
||||
|
||||
|
||||
class ImageTask(PrintTask):
|
||||
"""
|
||||
This tasks represents a image content ( in the form of it's path ), and it's signature.
|
||||
"""
|
||||
|
||||
def __init__(self, image_path, signature, process):
|
||||
super().__init__(TaskType.IMAGE)
|
||||
self.image_path = image_path
|
||||
self.signature = signature
|
||||
self.process = process
|
||||
|
||||
def get_print_data(self):
|
||||
# Return image data in printer-compatible format
|
||||
return {"img": self.image_path, "sign": self.signature, "process": self.process}
|
||||
|
||||
|
||||
class CutTask(PrintTask):
|
||||
"""
|
||||
This class activates the cutter on the printer if it exists
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(TaskType.CUT)
|
||||
|
||||
# There is no print data,
|
||||
# the task existence in itself is indication of what to do
|
||||
def get_print_data(self):
|
||||
return None
|
||||
@@ -5,10 +5,9 @@
|
||||
<div class="card">
|
||||
<h3 class="card-header">Print a short message</h3>
|
||||
<div class="card-body">
|
||||
<form class="form-group" action="/web/print/sms" method="post">
|
||||
<textarea class="form-control" type="text" name="txt" placeholder="4096 chars or less " maxlength="4096"></textarea>
|
||||
<br>
|
||||
<input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200"><br>
|
||||
<form class="form-group" action="/api/print/sms" method="post">
|
||||
<input class="form-control" type="text" name="txt" placeholder="200 chars or less " maxlength="200" required><br>
|
||||
<input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200" required><br>
|
||||
<input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
|
||||
</form>
|
||||
</div>
|
||||
@@ -19,9 +18,9 @@
|
||||
<div class="card">
|
||||
<h3 class="card-header">Print an image</h3>
|
||||
<div class="card-body">
|
||||
<form enctype="multipart/form-data" class="form-group" action="/web/print/img" method="post">
|
||||
<form enctype="multipart/form-data" class="form-group" action="/api/print/img" method="post">
|
||||
<input class="form-control" type="file" name="img" required><br>
|
||||
<input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200"><br>
|
||||
<input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200" required><br>
|
||||
<input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
|
||||
</form>
|
||||
</div>
|
||||
|
||||
@@ -54,10 +54,6 @@
|
||||
|
||||
</div>
|
||||
|
||||
<div id="countdown">
|
||||
<div id="countdown-number"></div>
|
||||
</div>
|
||||
|
||||
<style media="screen">
|
||||
canvas {
|
||||
display: none;
|
||||
|
||||
72
src/user.py
72
src/user.py
@@ -1,41 +1,39 @@
|
||||
# class User(object):
|
||||
# """docstring for User."""
|
||||
class User(object):
|
||||
"""docstring for User."""
|
||||
|
||||
# def __init__(self, arg):
|
||||
# super(User, self).__init__()
|
||||
# self.arg = arg
|
||||
def __init__(self, arg):
|
||||
super(User, self).__init__()
|
||||
self.arg = arg
|
||||
|
||||
# @app.route('/login', methods=['POST','GET'])
|
||||
# @limiter.limit("100 per minute", error_message=error_handler_limiter)
|
||||
def login():
|
||||
if request.method == 'POST':
|
||||
if not session.get('logged_in'):
|
||||
if request.form['username'] and request.form['password']:
|
||||
# Get the json
|
||||
with open('users.json') as f:
|
||||
users_file = json.load(f)
|
||||
for user in users_file["users"]:
|
||||
if users_file["users"][user] == request.form['password']:
|
||||
session['logged_in'] = True
|
||||
session['user'] = request.form['username']
|
||||
|
||||
# # @app.route('/login', methods=['POST','GET'])
|
||||
# # @limiter.limit("100 per minute", error_message=error_handler_limiter)
|
||||
# def login():
|
||||
# if request.method == "POST":
|
||||
# if not session.get("logged_in"):
|
||||
# if request.form["username"] and request.form["password"]:
|
||||
# # Get the json
|
||||
# with open("users.json") as f:
|
||||
# users_file = json.load(f)
|
||||
# for user in users_file["users"]:
|
||||
# if users_file["users"][user] == request.form["password"]:
|
||||
# session["logged_in"] = True
|
||||
# session["user"] = request.form["username"]
|
||||
if not session.get('logged_in'):
|
||||
flash('Mot de passe ou pseudo invalide.','danger')
|
||||
return redirect(url_for('login'))
|
||||
else:
|
||||
return redirect(url_for('display_index_page'))
|
||||
else:
|
||||
flash('Incorrect logins')
|
||||
return render_template('password.html')
|
||||
else:
|
||||
return render_template('password.html')
|
||||
else:
|
||||
return render_template('password.html')
|
||||
|
||||
# if not session.get("logged_in"):
|
||||
# flash("Mot de passe ou pseudo invalide.", "danger")
|
||||
# return redirect(url_for("login"))
|
||||
# else:
|
||||
# return redirect(url_for("display_index_page"))
|
||||
# else:
|
||||
# flash("Incorrect logins")
|
||||
# return render_template("password.html")
|
||||
# else:
|
||||
# return render_template("password.html")
|
||||
# else:
|
||||
# return render_template("password.html")
|
||||
|
||||
|
||||
# @app.route("/logout")
|
||||
# def logout():
|
||||
# session["logged_in"] = False
|
||||
# flash("Tu est déconnecté", "info")
|
||||
# return redirect(url_for("login"))
|
||||
@app.route("/logout")
|
||||
def logout():
|
||||
session['logged_in'] = False
|
||||
flash('Tu est déconnecté', 'info')
|
||||
return redirect(url_for('login'))
|
||||
|
||||
147
src/web.py
147
src/web.py
@@ -1,122 +1,69 @@
|
||||
"""
|
||||
Manage all of the inputs from a web source
|
||||
"""
|
||||
|
||||
import os
|
||||
from flask import Flask, request, flash
|
||||
from werkzeug.utils import secure_filename
|
||||
from task import TextTask, ImageTask, CutTask
|
||||
from printer import Printer
|
||||
import time
|
||||
import os
|
||||
|
||||
class Web(object):
|
||||
"""docstring for web."""
|
||||
|
||||
class Web:
|
||||
"""Web is the class that gets all of the information from web calls
|
||||
( API and Web page ) and provides checks before sending stuff to printing"""
|
||||
|
||||
def __init__(self, app, print_queue):
|
||||
def __init__(self, app, printer ):
|
||||
super(Web).__init__()
|
||||
self.print_queue = print_queue
|
||||
self.printer = printer
|
||||
self.app = app
|
||||
|
||||
def print_sms(self, texte, sign: str) -> bool:
|
||||
"""
|
||||
Get text and a signature, prints the text and cuts after that.
|
||||
"""
|
||||
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))
|
||||
|
||||
try:
|
||||
# We create two new tasks and add them directly to the queue
|
||||
# TODO: this might need to be improved because
|
||||
# !! there is no garantee !! that both the SMS task and the Cut task
|
||||
# are added back to back, another task could be
|
||||
# inserted between the two.
|
||||
sms = self.print_queue.enqueue(TextTask(content=texte, signature=sign))
|
||||
cut = self.print_queue.enqueue(CutTask())
|
||||
self.printer.print_sms(texte, sign)
|
||||
self.printer.cut()
|
||||
except Exception as e:
|
||||
self.app.logger.error(e)
|
||||
raise RuntimeError("Could not add SMS to queue, " + str(e)) from e
|
||||
self.app.logger.info("Added two new tasks at position %s and %s", sms, cut)
|
||||
return True
|
||||
flash("Error while printing the SMS : "+ str(e))
|
||||
|
||||
def print_image(self, image, sign: str) -> bool:
|
||||
"""
|
||||
Get an image and a signature, prints the image and cuts after that.
|
||||
"""
|
||||
flash("You message " + str( texte ) + " has been printed :)")
|
||||
|
||||
|
||||
def print_image(self, image, sign):
|
||||
self.app.logger.debug("Uploading file")
|
||||
try:
|
||||
file_uploaded = self.upload_file(image)
|
||||
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=sign,process=True)
|
||||
self.printer.cut()
|
||||
except Exception as e:
|
||||
self.app.logger.error(e)
|
||||
raise RuntimeError("Could not upload file : " + str(e)) from e
|
||||
flash("Could not upload file." + str(e))
|
||||
|
||||
if file_uploaded:
|
||||
self.app.logger.debug("File has been uploaded, printing...")
|
||||
try:
|
||||
img = self.print_queue.enqueue(
|
||||
ImageTask(
|
||||
os.path.join(
|
||||
self.app.config["UPLOAD_FOLDER"],
|
||||
secure_filename(image.filename),
|
||||
),
|
||||
signature=sign,
|
||||
process=True,
|
||||
)
|
||||
)
|
||||
flash("Your image has been printed :)")
|
||||
|
||||
cut = self.print_queue.enqueue(CutTask())
|
||||
except Exception as e:
|
||||
raise RuntimeError("Could not add IMG to queue" + str(e)) from e
|
||||
def login(username: str,password: str) -> bool:
|
||||
pass
|
||||
|
||||
self.app.logger.info("Added two new tasks at position %s and %s", img, cut)
|
||||
|
||||
return True
|
||||
|
||||
# def login(self, username: str, password: str) -> bool:
|
||||
# """Not implemented"""
|
||||
# return
|
||||
|
||||
# def logout(self, username: str, password: str) -> bool:
|
||||
# """Not implemented"""
|
||||
# return
|
||||
def logout(username: str, password: str) -> bool:
|
||||
pass
|
||||
|
||||
def allowed_file(self, filename) -> bool:
|
||||
"""
|
||||
Check if the file extension is allowed
|
||||
"""
|
||||
self.app.logger.debug("Checking if the file extension is allowed")
|
||||
return (
|
||||
"." in filename
|
||||
and filename.rsplit(".", 1)[1].lower()
|
||||
in self.app.config["ALLOWED_EXTENSIONS"]
|
||||
)
|
||||
self.app.logger.debug("Is the filename allowed ?")
|
||||
return '.' in filename and filename.rsplit('.', 1)[1].lower() in self.app.config['ALLOWED_EXTENSIONS']
|
||||
|
||||
def upload_file(self, image) -> bool:
|
||||
"""
|
||||
Save the file after executing checks on it
|
||||
"""
|
||||
def upload_file(self, image)-> bool:
|
||||
self.app.logger.debug("Validating file")
|
||||
if not image is None or not image == "":
|
||||
if self.allowed_file(image.filename):
|
||||
filename = secure_filename(image.filename)
|
||||
self.app.logger.debug("File valid")
|
||||
try:
|
||||
image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
|
||||
except OSError as e:
|
||||
self.app.logger.error("Could not save file %s", e)
|
||||
raise RuntimeError("An OS error occured while uploading this file : " + str(e)) from e
|
||||
if image and self.allowed_file(image.filename):
|
||||
filename = secure_filename(image.filename)
|
||||
self.app.logger.debug("File valid")
|
||||
try:
|
||||
image.save(os.path.join(self.app.config['UPLOAD_FOLDER'], filename))
|
||||
self.app.logger.debug("File saved")
|
||||
except Exception as e:
|
||||
self.app.logger.error("Could not save file")
|
||||
flash(str(e),'error')
|
||||
return False
|
||||
|
||||
self.app.logger.debug(
|
||||
"File saved to "
|
||||
+ str(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
|
||||
)
|
||||
return True
|
||||
|
||||
self.app.logger.error(
|
||||
"Could not save file because the filename is forbidden"
|
||||
)
|
||||
raise RuntimeError("This file type is forbidden.")
|
||||
|
||||
def get_queue_state(self):
|
||||
"""Return current queue state"""
|
||||
return self.print_queue.get_queue_state()
|
||||
|
||||
def get_queue_completed(self):
|
||||
"""Return completed queue elements"""
|
||||
return self.print_queue.get_queue_completed()
|
||||
self.app.logger.debug("File saved to " + str(os.path.join(self.app.config['UPLOAD_FOLDER'], filename)))
|
||||
return True
|
||||
else:
|
||||
self.app.logger.error("Could not save file " + str(filename))
|
||||
return False
|
||||
|
||||
147
src/worker.py
147
src/worker.py
@@ -1,147 +0,0 @@
|
||||
"""
|
||||
This is the main printing thread. A worker thread consums Tasks from
|
||||
a PrintQueue, while trying to find available printers.
|
||||
"""
|
||||
|
||||
import threading
|
||||
import time
|
||||
from printers import Printers
|
||||
|
||||
|
||||
class PrintWorker(threading.Thread):
|
||||
"""
|
||||
A thread used to consume Tasks added to a Print Queue.
|
||||
|
||||
On initialisation, the worker will try to find Printers,
|
||||
and on each print, choose an available Printer from a list of Printers.
|
||||
|
||||
If a print fails, it's not retried, but will be added to a list of completed
|
||||
tasks.
|
||||
"""
|
||||
def __init__(self, app, print_queue):
|
||||
super().__init__(daemon=True)
|
||||
self.app = app
|
||||
self.print_queue = print_queue
|
||||
self.printer = None
|
||||
self._lock = threading.Lock()
|
||||
self.running = True
|
||||
self.state = "idle" # idle, printing, dead, drinking-a-beer
|
||||
self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
|
||||
|
||||
try:
|
||||
self.printers = Printers(self.app)
|
||||
self.printers_obj = self.printers.printers
|
||||
self.printers = iter(self.printers.printers)
|
||||
except RuntimeError as e:
|
||||
self.app.logger.warning("Could not get any Printers")
|
||||
raise e
|
||||
|
||||
def run(self):
|
||||
"""Background thread that processes queue items"""
|
||||
self.app.logger.debug("Worker %s started working.", threading.get_ident())
|
||||
self.app.logger.debug("Current threads : %s" , threading.active_count())
|
||||
self.app.logger.debug("Threads actives : %s " , threading.enumerate())
|
||||
|
||||
while True:
|
||||
|
||||
# If the printer is dead or asleep, it can't work.
|
||||
if not self.running:
|
||||
time.sleep(0.2)
|
||||
continue
|
||||
|
||||
# If we have no available printer, we look at the list printers
|
||||
# we know about, and try to find one that is available.
|
||||
# When we find a printer, we acquire it
|
||||
# When we are finished with a printer, we release it to the world.
|
||||
while not self.printer or not self.printer.ready:
|
||||
time.sleep(1)
|
||||
try:
|
||||
self.app.logger.debug("Changing printers")
|
||||
self.printer = next(self.printers)
|
||||
self.app.logger.debug(
|
||||
"The worker got a %s printer and it's %s",
|
||||
self.printer.printer_type,
|
||||
"Ready" if self.printer.ready else "Not ready",
|
||||
)
|
||||
except Exception as e:
|
||||
self.app.logger.error("No printer detected" + str(e))
|
||||
self.printer = None
|
||||
|
||||
if self.state != "idle":
|
||||
self.app.logger("We are not idle, waiting...")
|
||||
time.sleep(1)
|
||||
continue
|
||||
|
||||
self.state = "printing"
|
||||
|
||||
with self._lock:
|
||||
try:
|
||||
task = self.print_queue.dequeue()
|
||||
except Exception as e:
|
||||
self.app.logger.error("Could not get a new task ! %s ", str(e))
|
||||
self.state = "idle"
|
||||
raise RuntimeError(
|
||||
"We could not get a new task because " + str(e)
|
||||
) from e
|
||||
|
||||
if task:
|
||||
try:
|
||||
self.app.logger.info("Got a new task")
|
||||
self.app.logger.debug("Got task %s", task.task_id)
|
||||
task.status = "processing"
|
||||
|
||||
print_data = task.get_print_data()
|
||||
|
||||
try:
|
||||
self.printer.print_task(task.task_type, print_data)
|
||||
except RuntimeError as e:
|
||||
self.state = "idle"
|
||||
self.app.logger.error("Could not print : %s", str(e))
|
||||
raise e
|
||||
|
||||
task.status = "completed"
|
||||
self.print_queue.mark_completed(task.task_id, "completed")
|
||||
self.app.logger.debug(
|
||||
"Finished printing task %s " , task.task_id
|
||||
)
|
||||
self.state = "idle"
|
||||
|
||||
except RuntimeError as e:
|
||||
task.status = "failed"
|
||||
self.state = "idle"
|
||||
self.print_queue.mark_completed(task.task_id, "failed")
|
||||
self.app.logger.error(
|
||||
"Could not print task %s because %s " , task.task_id, str(e)
|
||||
)
|
||||
|
||||
else:
|
||||
# When they are no new tasks to handle, we put the thread to sleep.
|
||||
self.state = "idle"
|
||||
time.sleep(0.1)
|
||||
|
||||
def stop_worker(self):
|
||||
"""
|
||||
Give the worker a break
|
||||
"""
|
||||
self.app.logger.debug("Giving the worker a break")
|
||||
self.state = "drinking-a-beer"
|
||||
self.running = False
|
||||
|
||||
def start_worker(self):
|
||||
"""
|
||||
Get the worker back to it
|
||||
"""
|
||||
self.app.logger.debug("Time to work !")
|
||||
self.state = "idle"
|
||||
self.running = True
|
||||
|
||||
def current_state(self):
|
||||
"""
|
||||
Return the worker state
|
||||
"""
|
||||
return {
|
||||
"is_running": self.running,
|
||||
"queue_size": len(self.print_queue),
|
||||
"state": self.state,
|
||||
"printers": len(self.printers_obj),
|
||||
}
|
||||
Reference in New Issue
Block a user