n07070/littleprynter
2
0
Fork 0
Code Issues 15 Pull Requests 1 Projects Releases Activity

Merge pull request 'Restructure the code and implement a printing queue' (#29) from restructure-printing-queue into master

Browse Source 
Reviewed-on: #29
This commit was merged in pull request #29.
This commit is contained in:
n07070
2026-05-27 00:00:56 +02:00
parent bd9888caf7 bbfe1936da
commit 0e3cc46a41
13 changed files with 882 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@@ -79,6 +79,16 @@ 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. 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 ### 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. 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.

180
poetry.lock generated
View File

@@ -75,6 +75,25 @@ files = [
{file = "blinker-1.9.0.tar.gz", hash = "sha256:b4ce2265a7abece45e7cc896e98dbebe6cead56bcf805a3d23136d145f5445bf"}, {file = "blinker-1.9.0.tar.gz", hash = "sha256:b4ce2265a7abece45e7cc896e98dbebe6cead56bcf805a3d23136d145f5445bf"},
] ]
[[package]]
name = "brother-ql-inventree"
version = "1.3"
description = "Python package to talk to Brother QL label printers"
optional = false
python-versions = "*"
groups = ["main"]
files = [
{file = "brother_ql_inventree-1.3-py3-none-any.whl", hash = "sha256:0f7e0d78bae04f44bcfe1010ed0d99f98d5b4db1d6179da242d5bd52bb0c9ea4"},
{file = "brother_ql_inventree-1.3.tar.gz", hash = "sha256:24335ca5f4b3444c692698b599459a7e6c4bd036dd580074c63d39382914fca3"},
]
[package.dependencies]
attrs = "*"
click = "*"
packbits = "*"
pillow = ">=10.0.0"
pyusb = "*"
[[package]] [[package]]
name = "certifi" name = "certifi"
version = "2026.4.22" version = "2026.4.22"
@@ -867,84 +886,84 @@ files = [
[[package]] [[package]]
name = "numpy" name = "numpy"
version = "2.4.5" version = "2.4.6"
description = "Fundamental package for array computing in Python" description = "Fundamental package for array computing in Python"
optional = false optional = false
python-versions = ">=3.11" python-versions = ">=3.11"
groups = ["main"] groups = ["main"]
files = [ files = [
{file = "numpy-2.4.5-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:3176dc8ff71dbb593606f91a69ad0c3cd3303c7eb546af477370ab9edf760288"}, {file = "numpy-2.4.6-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:0280e0356c0829a18d9de1cb7eee50ec22ca639878d7240307ca0943d73cd2c4"},
{file = "numpy-2.4.5-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:1811150e5148f5a01a7cc282cb2f489b4a3050a773e173adb480e507bad3a3d7"}, {file = "numpy-2.4.6-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:110f8b71aacb688ec69062bb7f6938a0f8acb01b7c1c4beb453c65b6d234584d"},
{file = "numpy-2.4.5-cp311-cp311-macosx_14_0_arm64.whl", hash = "sha256:0d63a780070871210853ba01e90b88f9b85cf2abf63a7f143d5127189265ddf6"}, {file = "numpy-2.4.6-cp311-cp311-macosx_14_0_arm64.whl", hash = "sha256:4cfe66903cc32a9921a6733d96b19bb6abf310397581bbad89c228f5abaf0ee8"},
{file = "numpy-2.4.5-cp311-cp311-macosx_14_0_x86_64.whl", hash = "sha256:0c6919cefafb3b76cd46a89dbb203bf1dd95529d2a6d09fef2d325d95d6a79d8"}, {file = "numpy-2.4.6-cp311-cp311-macosx_14_0_x86_64.whl", hash = "sha256:8155154c7c691289fe18f510b5d4657c68c67989f293f0535a91360392ff6538"},
{file = "numpy-2.4.5-cp311-cp311-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d51efede1e58e8b11877536a5518f60e318d8ff69b89ad7b38ee5e431b24d772"}, {file = "numpy-2.4.6-cp311-cp311-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0ab0a9c4ffb1a6d95ef519fe4247dba8eb6b18ad93999f76b7f657039acabd47"},
{file = "numpy-2.4.5-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:07ce7e74da92d7c71b5df157b9758bcdd53d7fea10602154de3afd2b3ddc34dd"}, {file = "numpy-2.4.6-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:89cd468399cfd2504718f0ba50e410dca55a170b61a02ad92bb18c8a65186e93"},
{file = "numpy-2.4.5-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:d7828234a13185effb34979e146f9921f2a65dfbbe215e6dbb57d6478fc8e059"}, {file = "numpy-2.4.6-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:c2d37ab77531417474168eb79d6d80b14f821a966818505d03013d0833edb7a8"},
{file = "numpy-2.4.5-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:f96083adc3dfc1bbf778f2c79654d88115fa07074c97cb724fe9508f12d91c55"}, {file = "numpy-2.4.6-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:f407cb6b8e9d6d8c626bc73c945db1706035af8fd632295547bf1c9e46d092d6"},
{file = "numpy-2.4.5-cp311-cp311-win32.whl", hash = "sha256:4ed78c904a638b6e5d7cd4db90c06fca5fc6ec2f28d258305368f454a50e79cf"}, {file = "numpy-2.4.6-cp311-cp311-win32.whl", hash = "sha256:ddea102b48f9e339f3948bf22040944184627a30fdf7f858667673b9c5f033c8"},
{file = "numpy-2.4.5-cp311-cp311-win_amd64.whl", hash = "sha256:079b0fad6f2899b23c5da89792b5409d2d83fc83e8bd5c2299cc9c397a264864"}, {file = "numpy-2.4.6-cp311-cp311-win_amd64.whl", hash = "sha256:1e254a00cdf42b1e4d5b3d68d33af63268d41340d8885df2ab6470f2e1500147"},
{file = "numpy-2.4.5-cp311-cp311-win_arm64.whl", hash = "sha256:d6c78e260b53affe9b395a9d54fc61f101f9521c4d9452c7e9e3718b19e2215b"}, {file = "numpy-2.4.6-cp311-cp311-win_arm64.whl", hash = "sha256:ed9749eef4cbd126da3dc1d6bcb3a57f5eb7ac6a6484146bdbf743f552dfc577"},
{file = "numpy-2.4.5-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:654fb8674b61b1c4bd568f944d13a908566fdcb0d797303521d4149d16da05ef"}, {file = "numpy-2.4.6-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:001fbb8e08d942dd57599e781f2472269ee7f2755fae407b4f67b2f0b17da3f1"},
{file = "numpy-2.4.5-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:4cd9f6fa7ce10dc4627f2bb81dd9075dab67e94632e04c2b638e12575ddaa862"}, {file = "numpy-2.4.6-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:ebfb099f8dcf083deef3ac1ca4c1503f387cf76296fcb3816b66f5ecb5f54fdb"},
{file = "numpy-2.4.5-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:4f5bc96d35d94e4ceab8b38a92241b4611e95dc44e63b9f1fa2a331858ee3507"}, {file = "numpy-2.4.6-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:3213d622a0283a39a93d188f3cf72b26862df52fbb4ca3697f51705016523d41"},
{file = "numpy-2.4.5-cp312-cp312-macosx_14_0_x86_64.whl", hash = "sha256:4bb33e900ee81730ad77a258965134aa8ceac805124f7e5229347beda4b8d0aa"}, {file = "numpy-2.4.6-cp312-cp312-macosx_14_0_x86_64.whl", hash = "sha256:357cc07a6d7b0b182ff02249616a03742827ebb1277546b5c7cd7f7620a45698"},
{file = "numpy-2.4.5-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:32f8f852273ef32b291201ac2a2c97629c4a1ee8632bb670e3443eaa09fc2e72"}, {file = "numpy-2.4.6-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:5f9fb9157b4ce2971008323afe46053787b526ef624fea915b261468a8421a0f"},
{file = "numpy-2.4.5-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:685681e956fc8dcb75adc6ff26694e1dfd738b24bd8d4696c51ca0110157f912"}, {file = "numpy-2.4.6-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:90f9849678c75fe7afa2d348ac842c168b0a4d3d61919687216dfc547976d853"},
{file = "numpy-2.4.5-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:6f64dd84b277a737eb59513f6b9bb6195bf41ab11941ef15b2562dbab43fa8ef"}, {file = "numpy-2.4.6-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:c1a2af6c6ef86344a6b0db6b97834208bf598db514f2b155042439b62605601a"},
{file = "numpy-2.4.5-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:b42d9496f79e3a728192f05a42d86e36163217b7cdecb3813d0028a0aa6b72d7"}, {file = "numpy-2.4.6-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:e5805d5a22fd19c8ccff10a9561f9df94436b0545619ea579db2d3c35294bce2"},
{file = "numpy-2.4.5-cp312-cp312-win32.whl", hash = "sha256:86d980970f5110595ca14855768073b08585fc1acc36895de303e039e7dee4a5"}, {file = "numpy-2.4.6-cp312-cp312-win32.whl", hash = "sha256:e3eeb0aabd6bd5ce64faae67e9935203a6991b4bc2a485a767fbafb2c5125f45"},
{file = "numpy-2.4.5-cp312-cp312-win_amd64.whl", hash = "sha256:3333dba6a4e611d666f69e177ba8fe4140366ff681a5feb2374d3fd4fff3acb6"}, {file = "numpy-2.4.6-cp312-cp312-win_amd64.whl", hash = "sha256:d8e8286dd7cea7895157318d1b91cdacac64c479f3cbc8dce548331728484751"},
{file = "numpy-2.4.5-cp312-cp312-win_arm64.whl", hash = "sha256:4593d197270b894efeb538dcbe227e4bcf1c77f88c4c6bf933ead812cfaa4453"}, {file = "numpy-2.4.6-cp312-cp312-win_arm64.whl", hash = "sha256:4081eb135ac24158bd51cdfbef16f1c64df7063b1143f24731387137c092bec8"},
{file = "numpy-2.4.5-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:1ef248460b645c102026b82337cc4e88231909c66dd77b59ec6d6cac7e44f277"}, {file = "numpy-2.4.6-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:511dbaf848decaaaf4b4ca48032619fb3138710c4bf7da7617765edad1ef96b0"},
{file = "numpy-2.4.5-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:4603622bdcdbf8dccb1d9d5b21d16a7aa4e473ae6c8e14048d846fd4ca2907a0"}, {file = "numpy-2.4.6-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:bf162abab1c1a736333192707cef898e735a5ca00f38f27eeedf44b39d9e85eb"},
{file = "numpy-2.4.5-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:6c18d49c67689c562854b53fdc433b93e47c12952aa6fa6d59f185e1a5992419"}, {file = "numpy-2.4.6-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:043191bfa8eab18c776647b62723ac9dddece59743b13f49b2016094129c2b3f"},
{file = "numpy-2.4.5-cp313-cp313-macosx_14_0_x86_64.whl", hash = "sha256:b1c663ddc641f4192e90511bec61a09bc231e3bbdb996cdc6edbcaa0e528d685"}, {file = "numpy-2.4.6-cp313-cp313-macosx_14_0_x86_64.whl", hash = "sha256:6180d8b35af935aed8ece3a85e0a43f87393ae0ac87c8d2c8bd2c993f7270ef3"},
{file = "numpy-2.4.5-cp313-cp313-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:93793222b524f692f12b2f8752ce8b1d9d9125b2bfd5dbf0fb69c92c5e1ce86c"}, {file = "numpy-2.4.6-cp313-cp313-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:72fbe16c6fac95aedf5937fa873445cec2110be35d8a4e9433d7501fd98dae6b"},
{file = "numpy-2.4.5-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1616bde34b2bcba2fa9bde06217ce00da4f3d1bdfb264d54525a99e8fe170d83"}, {file = "numpy-2.4.6-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a7830bab239b79cda9c08c2da014761cafb48da6150e1da17ac06283f43b6089"},
{file = "numpy-2.4.5-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:09d7d97da1c2c62f4818b3e150a57572ff8dcf1cf5ac501aac832ffd4ebd9566"}, {file = "numpy-2.4.6-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:ef4aea96ce4d3b074422cb4f2f64e216bf9e213004bb58ecfdf50ea02ea8eb9a"},
{file = "numpy-2.4.5-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:2d68d0b355ab2e39fe0de59001d7151dfdbbb880ef67baeed806661e03df5097"}, {file = "numpy-2.4.6-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:dfa20cc6ca228e6b155b11da03825975ce66aea520985dbbddf0f2a5a495c605"},
{file = "numpy-2.4.5-cp313-cp313-win32.whl", hash = "sha256:fe28b64777ddfa0eca9b5f51474034ebe3dcb8324f48f27b28f479085673ae33"}, {file = "numpy-2.4.6-cp313-cp313-win32.whl", hash = "sha256:56b39e5e0622a09a25bf5baf62f4bcf0cb8a41ae6e2819cf49bbc5a74c083f91"},
{file = "numpy-2.4.5-cp313-cp313-win_amd64.whl", hash = "sha256:fb4a6c9c537d6ccec9cc4aeae4261bd3cc79b070c67ddc0646f5b1c07fddde42"}, {file = "numpy-2.4.6-cp313-cp313-win_amd64.whl", hash = "sha256:c4fc99836233ea196540b17ab0983aff60ed07941751930f5f4d05bc3b3b7359"},
{file = "numpy-2.4.5-cp313-cp313-win_arm64.whl", hash = "sha256:6d7df2da2e7ea0624a43aa368104b3a3ce14aae98ad4bb2c9a93fecef76f1c97"}, {file = "numpy-2.4.6-cp313-cp313-win_arm64.whl", hash = "sha256:a7c711e21628b52034bb5ab8d1bce291f752fcc5e92accc615778acee1ff4778"},
{file = "numpy-2.4.5-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:2a235607a18df941760a695927051af4b1cd5d3ee85840d0e2af816785771feb"}, {file = "numpy-2.4.6-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:112b06a867b235ef466ed3508ddf0238050df9c727cafb5301ac385b899189a1"},
{file = "numpy-2.4.5-cp313-cp313t-macosx_14_0_arm64.whl", hash = "sha256:58dcf64969d870f36bc7fbd557d2617e997db7dc06261b6e3327148ea460d0a4"}, {file = "numpy-2.4.6-cp313-cp313t-macosx_14_0_arm64.whl", hash = "sha256:eaf7fa2de5c0be8ae6ff8e9bea2ccd725e980541244521d8d4b5f3354a27babe"},
{file = "numpy-2.4.5-cp313-cp313t-macosx_14_0_x86_64.whl", hash = "sha256:235f54b0156274d8fa3155db3ed6d2f401c7e8f3367c90db0a12f02a58fde6ed"}, {file = "numpy-2.4.6-cp313-cp313t-macosx_14_0_x86_64.whl", hash = "sha256:7265a2f3d436e54ef9f2b52b5c937e6be778781bd97a590319d7348f1c1ca997"},
{file = "numpy-2.4.5-cp313-cp313t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ef3b5bb65437a3555c648e706475db01c645559ca80dc8b03e4f202ea757e0d6"}, {file = "numpy-2.4.6-cp313-cp313t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:f74a575920ab21fe304421a3fc28793d82e299cae9eccb37084e9fc7f3617c20"},
{file = "numpy-2.4.5-cp313-cp313t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7f09a7e5f017d7098c66522097c96257411c9620c0926212200d66bc8cee3976"}, {file = "numpy-2.4.6-cp313-cp313t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ede83e07a75dd06bc501566c1eca2afc0d61677c1472ac9ad93fdee6e638a48d"},
{file = "numpy-2.4.5-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:993a88d8fdd8554466a8765cd8bacd97ba56b70ca6b0a04bcdca77f5afed4222"}, {file = "numpy-2.4.6-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:68bb27509ac1b9a3443094260f6326150663b06abe40b73a2f81160623da5b67"},
{file = "numpy-2.4.5-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:84f58bed609b5669f5ad3d597901a4f1f86ee5b3c3708aaa55f05b4fe6e0f656"}, {file = "numpy-2.4.6-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:a0df0043bdb289bde1f62da130d20df23d58b45429f752bc7a8fc5325a225ecd"},
{file = "numpy-2.4.5-cp313-cp313t-win32.whl", hash = "sha256:7200c58f3f933ca61e66346667dcc8510bb111995e9ce15398a731e6a4afa4bb"}, {file = "numpy-2.4.6-cp313-cp313t-win32.whl", hash = "sha256:29a287e0cf63ff528da061de6b9f64a4618da591ca1046aafc54062e40ca7eab"},
{file = "numpy-2.4.5-cp313-cp313t-win_amd64.whl", hash = "sha256:c26c71080d35db5002102f5d9ff614d45de02aa1f7802943e691e063e5ee93bc"}, {file = "numpy-2.4.6-cp313-cp313t-win_amd64.whl", hash = "sha256:25c692919ac5a01f170a3bfcd62d745b24fd095c353d50812637d6fcab442e75"},
{file = "numpy-2.4.5-cp313-cp313t-win_arm64.whl", hash = "sha256:2caa576d1707b275cba1aeb60a5c50daa6fa2a3f28ecb08123bc05fd439005db"}, {file = "numpy-2.4.6-cp313-cp313t-win_arm64.whl", hash = "sha256:1e978ec1e8bd0e0e4de6bb75de9d30cbb74db6b6a2bb727618613703ca0167dd"},
{file = "numpy-2.4.5-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:889ca2c072315de638a5194a772aa1fa2df92bdd6175f6a222d4784040424b61"}, {file = "numpy-2.4.6-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:06ca2f61ec4385a07a6977c55ba998a4466c123642b4a32694d3128fce18c079"},
{file = "numpy-2.4.5-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:89e89304fb1f8c3f0ecfa4a7d48f311dd79771336a940e920159d643d1307e77"}, {file = "numpy-2.4.6-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:38efbc8de75c7a0fc1ac190162d892787f3f47b57cc291231aafee36b80982b7"},
{file = "numpy-2.4.5-cp314-cp314-macosx_14_0_arm64.whl", hash = "sha256:144fcc5a3a17679b2b82543b4a2d8dd29937230a7af13232b5f753872feb6361"}, {file = "numpy-2.4.6-cp314-cp314-macosx_14_0_arm64.whl", hash = "sha256:d581b735e177fdcdce6fed8e7e8880a3fb6ee4e3653a3ac6af01c6f4c03effc5"},
{file = "numpy-2.4.5-cp314-cp314-macosx_14_0_x86_64.whl", hash = "sha256:398bb16772b265b9fa5c07b07072646ea97137c10ffb62a9a087b277fc825c29"}, {file = "numpy-2.4.6-cp314-cp314-macosx_14_0_x86_64.whl", hash = "sha256:0a041d3d761dc3c35cc56ce0351506a02bcbc25f7b169f652435141a17db9096"},
{file = "numpy-2.4.5-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:fb352e7b8876da1249e72254736d6c58c505fa4e58a3d7e30efca241ca9ca9ce"}, {file = "numpy-2.4.6-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:40fdc1ae7125e518ea98e53e69a4ebc27e1fd50510c47b7ea130cf21e5e1d42b"},
{file = "numpy-2.4.5-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7341b08ff8124d7353939778e2707b8732d03c78c1c30e0815aba2dacbe1245a"}, {file = "numpy-2.4.6-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a2c306dea656c12c68f51f4cea133cbe78ca7435eb28c735eac1d3ebe73be6e8"},
{file = "numpy-2.4.5-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:deb01226f012539f3945261ffe1c10aec081a0fa0a5c925419933c70f3ae2d23"}, {file = "numpy-2.4.6-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:33111801a01c12a8a1e3721f0a9232f8cfc8ae2c6b7098167e6f623c6073f402"},
{file = "numpy-2.4.5-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:d888bdf7335f76878c3c7b264ac1ff089863e211ec81249f9fb5795c2183dc25"}, {file = "numpy-2.4.6-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ae506e6902902557576a26ff33eda8695e7ecb3cb36c3b573a0765dee114ebdb"},
{file = "numpy-2.4.5-cp314-cp314-win32.whl", hash = "sha256:15f90d1256e9b2320aff24fde44815b787ab6d7c49a1a11bfd8138b321c5f080"}, {file = "numpy-2.4.6-cp314-cp314-win32.whl", hash = "sha256:aaf159caa35993cb1f56fb9b8e4610d35758e7ca005412eb1daa856a78c9c4b1"},
{file = "numpy-2.4.5-cp314-cp314-win_amd64.whl", hash = "sha256:4bd2cd4ef9c0afa87de73723c0a33c0edff62143e1432917458e26d3d195d87f"}, {file = "numpy-2.4.6-cp314-cp314-win_amd64.whl", hash = "sha256:b507f5c4c1d508876d1819b6bf9a49d365b96320b5d4993426b33a23ca4b8261"},
{file = "numpy-2.4.5-cp314-cp314-win_arm64.whl", hash = "sha256:db304568c650e9d7039744d3575d0d287754debb2057d7c7b8cdfdc2c487a957"}, {file = "numpy-2.4.6-cp314-cp314-win_arm64.whl", hash = "sha256:6f41ae150c4e32db4f3310cdaf64b1593a03dbabe29eec77fc9b50fe64061df6"},
{file = "numpy-2.4.5-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:6de2883e0d2c63eae1bab1a84b390dca74aabb3d20ea1f5d58f360853c83abf3"}, {file = "numpy-2.4.6-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:ece3d2cfe132e7d51f44a832b303895e6f2d499c5e74dfbdb06ee246147a304a"},
{file = "numpy-2.4.5-cp314-cp314t-macosx_14_0_arm64.whl", hash = "sha256:06760fe73ae5005008748d182de612c733542af3cde063d532cd2127561b27be"}, {file = "numpy-2.4.6-cp314-cp314t-macosx_14_0_arm64.whl", hash = "sha256:e3e5193ef5a3dc73bceee50f7fdc2c90dbb76c42df8d8fae3d1067a583df579e"},
{file = "numpy-2.4.5-cp314-cp314t-macosx_14_0_x86_64.whl", hash = "sha256:4b51a01745cb04cc19278482207444b4d30728ce91c28d27a3bfae5fc6ff24c7"}, {file = "numpy-2.4.6-cp314-cp314t-macosx_14_0_x86_64.whl", hash = "sha256:17f9ade344e7d9b464a084d69bcf18fc691cb1db67c62ed80820bf4926d78f0e"},
{file = "numpy-2.4.5-cp314-cp314t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9a05636d7937d0936f271e5ba957fa8d746b5be3c2025caa1a2508f4fe521d40"}, {file = "numpy-2.4.6-cp314-cp314t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9cd5ffd25db4e7ba6a375693b3fc0fc1791ec636c17db3720da19bde7180ec43"},
{file = "numpy-2.4.5-cp314-cp314t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:14b86f56048ed09c3bbe48962a7dff077c2fd3274f8cf981800f3b38eac49cc3"}, {file = "numpy-2.4.6-cp314-cp314t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7d92c3819208a60205a12a245c91ad70cb0a85336659b19b834205573ac8456e"},
{file = "numpy-2.4.5-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:130d58151c4db23e9fa860b84784e219a3aa3e030acc88a493ea37006c4dfd4c"}, {file = "numpy-2.4.6-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:e85b752a1e912b70eaad4fafbd4d1238007ab221de2009b9a2f5ae7461239895"},
{file = "numpy-2.4.5-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:d475afc8cbe935ff5944f753d863bba774d7f4e1feaaa4102901e3e053ca5963"}, {file = "numpy-2.4.6-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:29cb7f67d10b479ff07c17d33e39f78c07f71c40ef30d63c153d340e96cd3fb4"},
{file = "numpy-2.4.5-cp314-cp314t-win32.whl", hash = "sha256:27f4a6dc26353a860b348961b9aa9e009835688b435cfa105e873b8dc2c726f5"}, {file = "numpy-2.4.6-cp314-cp314t-win32.whl", hash = "sha256:260a5d70215b61ab4fadf5c7baacd64821842975eea312125ed3c39a6391b063"},
{file = "numpy-2.4.5-cp314-cp314t-win_amd64.whl", hash = "sha256:76ac6e90f5e226011c88f9b7040a4bcae612518bc7e9adc127e697a13b28ad1a"}, {file = "numpy-2.4.6-cp314-cp314t-win_amd64.whl", hash = "sha256:81a1cca95ed5bb92aa8b10dd2cdc9a0d3853a50fad926c28b5d7e8ea54389627"},
{file = "numpy-2.4.5-cp314-cp314t-win_arm64.whl", hash = "sha256:7c392e2c1bf596701d3c6832be7567eab5d5b0a13865036c33365ee097d37f8b"}, {file = "numpy-2.4.6-cp314-cp314t-win_arm64.whl", hash = "sha256:0c9136e14ed34a9e343a31c533d78a9813a69a3148332bce5e9821cb2f996e66"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-macosx_10_15_x86_64.whl", hash = "sha256:6bf0bfc1c2e1db972e30b6cd3d4861f477f3af908b27799b239dc3cbe3eb4b95"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-macosx_10_15_x86_64.whl", hash = "sha256:55cced7c52e981362f708ad635198e97a752dfba412cc03c23bbf3bd8d5cd662"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:73d664413fb97229149c4711ef56531a6fe8c15c1c2626b0bbe497b84c287e70"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:d6da64deb6b8ed903e7560180a92f2d804ee1ba5eeb849ac2748b8c1aba1f6d7"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-macosx_14_0_arm64.whl", hash = "sha256:b35bee5ef99e8d227a07829bee2e864fcb65f7c157646fcd8ec8b4b45dd8b88f"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-macosx_14_0_arm64.whl", hash = "sha256:68a5124b13fa6cc2086764a20005d30bc0548146f7f5322f02fce212ca14317f"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-macosx_14_0_x86_64.whl", hash = "sha256:02981d0fc9f9ce147643d552966d47f329a02f7ecb3b113e84207242f20dfa83"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-macosx_14_0_x86_64.whl", hash = "sha256:948424b06129ce883307e8cff868c31396d8dc7630a59c61d70d98dbe70f222c"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0e63caf31a1df06338ae63d999f7a33a675ced62eea9c9b02db4b1c1f45cff38"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:5dbbdb29840ca3d91ee0fece42fc29278886d908280bfec0a5846c6f901a3eb0"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:d8fc52b85a7b45e474be53eddf08e006d22e381a4e41bcde8e4aa08da0e7d198"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8ad03c0965fb3c692200e74d458ca28c1dbb4ce96f9a479a8aa041ad5fabca02"},
{file = "numpy-2.4.5-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:40c71d50a4da1a7c317af419461052d3911a5770bfc5fd55baf52cc45e7a2c20"}, {file = "numpy-2.4.6-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:2803abfebfc990042cd494d8ce2d5f82e9d847af6d35ec486923aa19dbad5e73"},
{file = "numpy-2.4.5.tar.gz", hash = "sha256:ca670567a5683b7c1670ec03e0ddd5862e10934e92a70751d68d7b7b74ca7f9f"}, {file = "numpy-2.4.6.tar.gz", hash = "sha256:f3a3570c4a2a16746ac2c31a7c7c7b0c186b95ce902e33db6f28094ed7387dda"},
] ]
[[package]] [[package]]
@@ -974,6 +993,17 @@ files = [
{file = "packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661"}, {file = "packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661"},
] ]
[[package]]
name = "packbits"
version = "0.6"
description = "PackBits encoder/decoder"
optional = false
python-versions = "*"
groups = ["main"]
files = [
{file = "packbits-0.6.tar.gz", hash = "sha256:bc6b370bb34e04ac8cfa835e06c0484380affc6d593adb8009dd6c0f7bfff034"},
]
[[package]] [[package]]
name = "pep8" name = "pep8"
version = "1.7.1" version = "1.7.1"
@@ -1852,5 +1882,5 @@ h11 = ">=0.16.0,<1"
[metadata] [metadata]
lock-version = "2.1" lock-version = "2.1"
python-versions = ">=3.13" python-versions = ">=3.14"
content-hash = "3a8909800e4977eaaac71770d29b53b450a233fe7ddde5e8bb158c151428ed7e" content-hash = "18e1a5c8b085aa639b665279856f9e3c51a734a6a8d1d4dd135ca7e4f67e86ae"

1
pyproject.toml
View File

@@ -42,6 +42,7 @@ dependencies = [
"python-escpos (>=3.1,<4.0)", "python-escpos (>=3.1,<4.0)",
"pep8 (>=1.7.1,<2.0.0)", "pep8 (>=1.7.1,<2.0.0)",
"pylint (>=4.0.5,<5.0.0)", "pylint (>=4.0.5,<5.0.0)",
"brother-ql-inventree (>=1.3,<2.0)",
] ]
[tool.poetry] [tool.poetry]

217
src/main.py
View File

@@ -39,11 +39,14 @@ from flask_limiter.util import get_remote_address
from printer import Printer # The wrapper for the printer class from printer import Printer # The wrapper for the printer class
from raspberry import Raspberry # The Raspberry pi control Class from raspberry import Raspberry # The Raspberry pi control Class
from web import Web # Wrapper for the web routes and API from web import Web # Wrapper for the web routes and API
from print_queue import PrintQueue
from worker import PrintWorker
# Variables # We create the main Flask object
app = Flask(__name__) app = Flask(__name__)
socketio = SocketIO(app) socketio = SocketIO(app, cors_allowed_origins="*")
# Global variables
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"} ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
# Load the configuration file # Load the configuration file
@@ -82,10 +85,10 @@ except FileExistsError:
app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER) app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
except PermissionError: except PermissionError:
app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER) app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
exit(77) sys.exit(77)
# Output the config file # Output the config file
if os.getenv("LIPY_DEBUG") is True: if os.getenv("FLASK_DEBUG"):
pprint.pprint(configuration_file) pprint.pprint(configuration_file)
# We define the app module used by Flask # We define the app module used by Flask
@@ -113,19 +116,22 @@ rpi = Raspberry(
RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi() RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
# Queue creation
print_queue = PrintQueue(app)
# Web & API routes # Web & API management
web = Web(app, print_queue)
web = Web(app, printer) # Start worker thread
worker = PrintWorker(app, print_queue, printer, socketio)
if __name__ == "__main__": worker.start()
app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
# The rate limit
limiter = Limiter( limiter = Limiter(
get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"] get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
) )
# General routes
@app.route("/") @app.route("/")
@limiter.limit("1/second", override_defaults=False) @limiter.limit("1/second", override_defaults=False)
def index(): def index():
@@ -142,6 +148,83 @@ def webcam():
return render_template("webcam.html") 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"]
# 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 # API routes
# The api has the following methods # The api has the following methods
# api/print/{sms,img,letter,qr,barcode} # api/print/{sms,img,letter,qr,barcode}
@@ -163,93 +246,115 @@ def api_index():
@limiter.limit("6/minute", override_defaults=False) @limiter.limit("6/minute", override_defaults=False)
def api_print_sms(): def api_print_sms():
"""Prints a short message on a printer""" """Prints a short message on a printer"""
app.logger.debug("Printing an sms") app.logger.debug("Printing an sms via API")
try: try:
txt = request.form["txt"] txt = request.form["txt"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e)) app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e))
flash("Whoops, no forms submitted or missing signature : %s", str(e)) return str(e), 400
return redirect(url_for("index"))
try: try:
# comment: We try to get a signature # comment: We try to get a signature
sign = request.form["signature"] sign = request.form["signature"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
app.logger.warning( app.logger.warning(
"No signature found for this print, using default signature.", str(e) "No signature found for this print, using default signature. : %s", str(e)
) )
sign = configuration_file["defaults"]["signature"] sign = configuration_file["defaults"]["signature"]
try:
web.print_sms(txt, sign) # comment: We try to print the SMS
return redirect(url_for("index")) web.print_sms(txt, sign)
except RuntimeError as e:
return str(e), 500
# end try
return "OK", 200
@app.route("/api/print/img", methods=["POST"]) @app.route("/api/print/img", methods=["POST"])
@limiter.limit("6/minute", override_defaults=False) @limiter.limit("6/minute", override_defaults=False)
def api_print_image(): def api_print_image():
"""Prints an image on a printer""" """Prints an image on a printer"""
app.logger.debug("Printing an image") app.logger.debug("Printing an image via API")
try: try:
# comment: We try to get a signature # comment: We try to get a signature
sign = request.form["signature"] sign = request.form["signature"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
app.logger.warning( app.logger.warning(
"No signature found for this print, using default signature.", str(e) "No signature found for this print, using default signature. %s", str(e)
) )
sign = configuration_file["defaults"]["signature"] sign = configuration_file["defaults"]["signature"]
if request.method == "POST": # check if the post request has the file part
# check if the post request has the file part if "img" not in request.files:
if "img" not in request.files: app.logger.error("Whoops, no images submitted.")
app.logger.error("Whoops, no images submitted : %s ", str(e)) return "No image submitted", 400
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"] file = request.files["img"]
# If the user does not select a file, the browser submits an # If the user submits an empty file without a filename.
# empty file without a filename. if file.filename == "":
if file.filename == "": app.logger.error("Submitted file has no filename !")
app.logger.error("Submitted file has no filename !") return "Submitted file has no filename !", 400
flash("Submitted file has no filename !")
return redirect(url_for("index"))
try: try:
app.logger.debug("Sending the image to the printer.") app.logger.debug("Sending the image to the printer.")
web.print_image(file, sign) web.print_image(file, sign)
except Exception as e: except RuntimeError as e:
app.logger.error("The image could not be printed because : %s ", str(e)) return str(e), 500
flash("The image could not be printed because : %s ", str(e))
return redirect(url_for("index"))
else: return "OK", 200
app.logger.error("Method not allowed")
flash("Method not allowed")
return redirect(url_for("index"))
flash("Picture printed ! ")
return redirect(url_for("index"))
@app.route("/api/camera/picture") @app.route("/api/camera/picture", methods=["GET"])
def camera_picture(): def camera_picture():
"""Returns a picture taken by the camera""" """Returns a picture taken by the camera on a raspberry pi"""
if RASPBERRY_PI_CONNECTED: if RASPBERRY_PI_CONNECTED:
try: try:
return rpi.camera_picture() return rpi.camera_picture()
except Exception as e: except RuntimeError as e:
return jsonify({"message": "Error getting the stream : " + e}), 500 return jsonify({"message": "Error getting the stream : " + e}), 500
else: else:
return jsonify({"message": "No camera present"}), 500 return jsonify({"message": "No camera present"}), 500
@app.route("/api/queue", methods=["GET"])
def api_queue_status():
"""API endpoint for entire queue"""
return jsonify(web.get_queue_state())
@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") @app.route("/login")
@limiter.limit("1/second", override_defaults=False) @limiter.limit("1/second", override_defaults=False)
def login_page(): def login_page():
"""Unsued, logins""" """Unsued, logins"""
# web.login(username,password) # web.login(username,password)
return redirect(url_for("index")) return redirect(url_for("index")), 501
@app.route("/logout") @app.route("/logout")
@@ -257,7 +362,7 @@ def login_page():
def logout_page(): def logout_page():
"""Unused, logout""" """Unused, logout"""
# web.logout(username, password) # web.logout(username, password)
return redirect(url_for("index")) return redirect(url_for("index")), 501
@app.errorhandler(429) @app.errorhandler(429)
@@ -295,3 +400,7 @@ def camera_status():
socketio.emit("camera_status", True) socketio.emit("camera_status", True)
else: else:
socketio.emit("camera_status", False) socketio.emit("camera_status", False)
if __name__ == "__main__":
app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

135
src/print_queue.py Normal file
View File

@@ -0,0 +1,135 @@
"""
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} for t in self._queue]
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(),
}

157
src/printer.py
View File

@@ -1,12 +1,18 @@
# Importing the module to manage the connection to the printer. """
import escpos.printer as escp This class manages connexion to a Printer
from time import sleep, gmtime, strftime """
# import brother_ql
from time import sleep
import os.path import os.path
from PIL import Image, ImageEnhance, ImageOps
from PIL import Image, ImageEnhance
import numpy as np import numpy as np
# Importing the module to manage the connection to the printer.
import escpos.printer
class Printer(object):
class Printer():
""" """
# The connection is based on the ESC/POS library # The connection is based on the ESC/POS library
@@ -27,7 +33,7 @@ class Printer(object):
ready = False ready = False
def __init__(self, app, device_id, vendor_id): def __init__(self, app, device_id, vendor_id):
super(Printer, self).__init__() super().__init__()
self.app = app self.app = app
self.ready = False self.ready = False
self.printer = None self.printer = None
@@ -38,7 +44,9 @@ class Printer(object):
self.usb_args["idProduct"] = self.vendor_id self.usb_args["idProduct"] = self.vendor_id
def check_paper(self) -> bool: def check_paper(self) -> bool:
# Let's check paper status """
On printers that support it, we check that the printer has paper
"""
self.app.logger.debug("Checking paper status...") self.app.logger.debug("Checking paper status...")
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
status = self.printer.paper_status() status = self.printer.paper_status()
@@ -46,7 +54,7 @@ class Printer(object):
case 0: 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() self.printer.close()
raise Exception("No more paper in the printer") raise RuntimeError("No more paper in the printer")
case 1: case 1:
self.app.logger.warning( self.app.logger.warning(
"Printer needs paper to be changed very soon ! " "Printer needs paper to be changed very soon ! "
@@ -57,28 +65,42 @@ class Printer(object):
self.printer.close() self.printer.close()
def init_printer(self): def init_printer(self):
"""
Check if the printer online ? Is the communication with the printer successfull ?
"""
# TODO: This could happen directly when creating a new Printer class
if os.getenv("FLASK_DEBUG"):
waiting_elapsed = 3
else:
waiting_elapsed = 10
# Is the printer online ? Is the communication with the printer successfull ?
waiting_elapsed = 30
self.app.logger.debug("Waiting for printer to get online...") self.app.logger.debug("Waiting for printer to get online...")
while not self.ready: while not self.ready:
try: try:
# This also calls open(), which we need to close() # This also calls open(), which we need to close()
# or else the device will appear as busy. # or else the device will appear as busy.
p = escp.Usb(self.device_id, self.vendor_id, 0, profile="TM-P80") p = escpos.printer.Usb(
except Exception as e: self.device_id, self.vendor_id, 0, profile="TM-P80"
self.app.logger.error( )
"The USB device is not plugged in, trying again : " + str(e) except RuntimeError as e:
self.app.logger.error(
"The USB device is not plugged in, trying again %s : %s",
waiting_elapsed,
str(e),
) )
pass
try: try:
if p.is_online(): if p.is_online():
self.ready = True self.ready = True
self.app.logger.debug("Printer online !") self.app.logger.debug("Printer online !")
except Exception as e: except RuntimeError as e:
pass self.app.logger.error(
"Error while getting the printer online %s : %s",
waiting_elapsed,
str(e),
)
sleep(1) sleep(1)
waiting_elapsed -= 1 waiting_elapsed -= 1
@@ -115,15 +137,24 @@ class Printer(object):
return True return True
def print_sms(self, msg, signature="", bold=False): def _print_sms(self, msg, signature="", bold=False):
if not isinstance(msg, str):
self.app.logger.error(
"It is not possible to print a " + str(type(msg)) + ", only strings."
)
raise ValueError
# We make sure that the signature is not something too goofy
clean_msg = str(msg) + "\n" clean_msg = str(msg) + "\n"
clean_signature = str(signature) clean_signature = str(signature)
# Make checks on the size of the message being printed
if len(clean_msg) > 4096: if len(clean_msg) > 4096:
self.app.logger.warning( self.app.logger.warning(
"Could not print message of this length: " + str(len(clean_msg)) "Could not print message of this length: " + str(len(clean_msg))
) )
raise Exception( raise RuntimeError(
"Could not print message of this length :" "Could not print message of this length :"
+ str(len(clean_msg)) + str(len(clean_msg))
+ ", needs to be below 4096 caracters long." + ", needs to be below 4096 caracters long."
@@ -133,12 +164,14 @@ class Printer(object):
self.app.logger.warning( self.app.logger.warning(
"Could not print signature of this length: " + str(len(clean_signature)) "Could not print signature of this length: " + str(len(clean_signature))
) )
raise Exception( raise RuntimeError(
"Could not print signature of this length :" "Could not print signature of this length :"
+ str(len(clean_signature)) + str(len(clean_signature))
+ ", needs to be below 256 caracters long." + ", 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: try:
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.set(align="center", font="a", bold=bold) self.printer.set(align="center", font="a", bold=bold)
@@ -148,18 +181,21 @@ class Printer(object):
self.printer.close() self.printer.close()
except Exception as e: except Exception as e:
self.app.logger.error("Unable to print because : " + str(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.app.logger.info("Printed text")
return True return True
def print_img(self, path, sign="", center=True, process=False): def _print_img(self, path, signature="", center=True, process=False):
clean_signature = str(sign) clean_signature = str(signature)
if len(sign) > 256: if len(signature) > 256:
self.app.logger.warning( self.app.logger.warning(
"Could not print signature of this length: " + str(len(clean_signature)) "Could not print signature of this length: " + str(len(clean_signature))
) )
raise Exception( raise ValueError(
"Could not print signature of this length :" "Could not print signature of this length :"
+ str(len(clean_signature)) + str(len(clean_signature))
+ ", needs to be below 256 caracters long." + ", needs to be below 256 caracters long."
@@ -167,45 +203,60 @@ class Printer(object):
if not os.path.isfile(str(path)): if not os.path.isfile(str(path)):
self.app.logger.warning("File does not exist : " + str(path)) self.app.logger.warning("File does not exist : " + str(path))
raise Exception( raise OSError(
"The file path for this image :" "The file path for this image :"
+ str(path) + str(path)
+ " wasn't found. Please try again." + " wasn't found. Please try again."
) )
else:
self.app.logger.debug("Printing file from " + str(path)) self.app.logger.debug("Printing file from " + str(path))
if process: if process:
try: try:
self.app.logger.debug("Proccessing the image") self.app.logger.debug("Proccessing the image")
path = process_image(self, path) path = _process_image(self, path)
except Exception as e: except RuntimeError as e:
self.app.logger.error(str(e)) self.app.logger.error(
return False "Error while processing the image, aborting print : %s", str(e)
)
raise e
else: else:
self.app.logger.warning("Not proccessing the image") self.app.logger.warning("Not proccessing the image")
try: try:
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.image(path, center=center) self.printer.image(path, center=center)
self.printer.textln(signature)
self.printer.close() self.printer.close()
self.app.logger.debug("Printed an image : " + str(path)) self.app.logger.debug("Printed an image : " + str(path))
os.remove(path)
self.app.logger.debug("Removed image : " + str(path))
except Exception as e: except Exception as e:
self.printer.close()
self.app.logger.error(str(e)) self.app.logger.error(str(e))
return False raise RuntimeError("Could not print the picture") from e
finally:
try:
os.remove(path)
except OSError as e:
raise e
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
self.app.logger.info("Printed a picture") self.app.logger.info("Printed a picture")
return True return True
def qr(self, content): def _qr(self, content):
try: try:
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.qr(content, center=True) self.printer.qr(content, center=True)
self.printer.close() self.printer.close()
except Exception as e: except RuntimeError as e:
self.printer.close() self.printer.close()
self.app.logger.error(str(e)) self.app.logger.error(str(e))
return False return False
@@ -213,7 +264,7 @@ class Printer(object):
self.app.logger.info("Printed a QR") self.app.logger.info("Printed a QR")
return True return True
def cut(self): def _cut(self):
try: try:
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.cut() self.printer.cut()
@@ -226,8 +277,22 @@ class Printer(object):
self.app.logger.info("Did a cut") self.app.logger.info("Did a cut")
return True return True
def print_task(self, task_type, data):
"""Execute actual print based on task type"""
match (task_type.value):
case "text":
self._print_sms(data["txt"], signature=data["sign"])
case "image":
self._print_img(
data["img"], signature=data["sign"], process=data["process"]
)
case "cut":
self._cut()
case _:
raise RuntimeError("This task type is not supported")
def process_image(self, path):
def _process_image(self, path):
brightness_factor = 1.5 # Used only if image is too dark brightness_factor = 1.5 # Used only if image is too dark
brightness_threshold = 100 # Brightness threshold (0–255) brightness_threshold = 100 # Brightness threshold (0–255)
contrast_factor = 0.6 # Less than 1.0 = lower contrast contrast_factor = 0.6 # Less than 1.0 = lower contrast
@@ -241,11 +306,12 @@ def process_image(self, path):
original_img = original_img.convert("RGB") original_img = original_img.convert("RGB")
# Resize while maintaining aspect ratio # Resize while maintaining aspect ratio
original_img.thumbnail((max_width, max_height), Image.LANCZOS) original_img.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
self.app.logger.debug("Resized the image") self.app.logger.debug("Resized the image")
# # Convert to grayscale for dithering # # 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") # self.app.logger.debug("Dithered the image")
# Compute brightness of original image (grayscale average) # Compute brightness of original image (grayscale average)
@@ -273,16 +339,9 @@ def process_image(self, path):
# contrast_enhancer = ImageEnhance.Contrast(original_img) # contrast_enhancer = ImageEnhance.Contrast(original_img)
# original_img = contrast_enhancer.enhance(contrast_factor) # 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 # Convert to JPEG and save
jpeg_path = os.path.splitext(path)[0] + "_processed.jpg" jpeg_path = os.path.splitext(path)[0] + "_processed.jpg"
original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True) original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True)
self.app.logger.debug("Processed and saved image.")
return jpeg_path return jpeg_path

54
src/raspberry.py
View File

@@ -1,13 +1,21 @@
from flask_socketio import SocketIO """
from gpiozero import Button, LED, DigitalOutputDevice This class executes when we are on a raspberry Pi.
from time import sleep, gmtime, strftime
from PIL import Image 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 import io # To check if we are on a Raspberry Pi
import subprocess import subprocess
import os import os
from time import sleep, gmtime, strftime
from flask_socketio import SocketIO
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 : This class will manage three things :
- Connecting to a USB webcam - Connecting to a USB webcam
@@ -19,7 +27,7 @@ class Raspberry(object):
def __init__( def __init__(
self, self,
printer, print_queue,
app, app,
socketio, socketio,
button_gpio_port_number, button_gpio_port_number,
@@ -27,7 +35,7 @@ class Raspberry(object):
flash_gpio_port_number, flash_gpio_port_number,
is_flash_present, is_flash_present,
): ):
self.printer = printer self.print_queue = print_queue
self.socketio = socketio self.socketio = socketio
self.app = app self.app = app
@@ -38,9 +46,13 @@ class Raspberry(object):
self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg" self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
def is_raspberry_pi(self, raise_on_errors=False): def is_raspberry_pi(self, raise_on_errors=False):
"""
Checking if we are on a Raspberry Pi by checking
information on the /proc/cpuinfo file
"""
# Check if we are running on a raspberry pi # Check if we are running on a raspberry pi
try: try:
with io.open("/proc/cpuinfo", "r") as cpuinfo: with io.open("/proc/cpuinfo", "r", encoding="utf-8") as cpuinfo:
found = False found = False
for line in cpuinfo: for line in cpuinfo:
if line.startswith("Hardware"): if line.startswith("Hardware"):
@@ -141,7 +153,7 @@ class Raspberry(object):
f"Unable to take a picture. Error: {e.stderr.decode()}" f"Unable to take a picture. Error: {e.stderr.decode()}"
) )
return False return False
except Exception as e: except RuntimeError as e:
# Catch any unexpected errors # Catch any unexpected errors
self.app.logger.error(f"Unexpected error while taking picture: {str(e)}") self.app.logger.error(f"Unexpected error while taking picture: {str(e)}")
return False return False
@@ -198,7 +210,7 @@ class Raspberry(object):
output_path = image_path # Overwrite the original image if no output path is given output_path = image_path # Overwrite the original image if no output path is given
image.save(output_path) image.save(output_path)
except Exception as e: except RuntimeError as e:
self.app.logger.error(f"Error overlaying logo: {e}") self.app.logger.error(f"Error overlaying logo: {e}")
return False return False
@@ -225,7 +237,7 @@ class Raspberry(object):
image.save(output_path) image.save(output_path)
except Exception as e: except RuntimeError as e:
self.app.logger.error(f"Error cropping image to square: {e}") self.app.logger.error(f"Error cropping image to square: {e}")
return False return False
@@ -242,7 +254,7 @@ class Raspberry(object):
try: try:
self.flash.on() self.flash.on()
self.take_picture() self.take_picture()
except Exception as e: except RuntimeError as e:
self.app.logger.error( self.app.logger.error(
"Could not take a picture after the button press : " + str(e) "Could not take a picture after the button press : " + str(e)
) )
@@ -251,15 +263,11 @@ class Raspberry(object):
self.app.logger.debug("Printing picture") self.app.logger.debug("Printing picture")
self.led.on() self.led.on()
self.crop_to_square(self.image_path) self.crop_to_square(self.image_path)
self.printer.print_img("src/static/images/extase-club.png", process=True) self.print_queue.enqueue(ImageTask(self.image_path,signature="",process=True))
self.printer.print_img(self.image_path, process=True) self.print_queue.enqueue(TextTask(content="Imprimé par LittlePrynter", signature=""))
self.printer.print_sms("") time = strftime("%Y-%m-%d %H:%M", gmtime())
self.printer.print_sms("With Love From Société.Vide", signature="", bold=True) self.print_queue.enqueue(TextTask(content=time, signature=""))
self.printer.print_sms("Printed by LittlePrynter", signature="") self.print_queue.enqueue(CutTask())
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.led.off()
self.app.logger.debug("Done printing picture") self.app.logger.debug("Added a photomaton picture to the print queue")
return True return True

7
src/static/js/webcam.js
View File

@@ -255,7 +255,12 @@ function print_picture(data){
// headers:{ // headers:{
// 'Content-Type': 'multipart/form-data' // 'Content-Type': 'multipart/form-data'
// } // }
}).then(function(response) { console.log('Success:', response); } , true) }).then(function(response) {
console.log('Reponse:', response);
if(response.status != 200 ){
alert("The picture could not be printed be : " + response.statusText)
}
} , true)
.catch(error => console.error('Error:', error), false); .catch(error => console.error('Error:', error), false);
} }

98
src/task.py Normal file
View File

@@ -0,0 +1,98 @@
"""
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"
class PrintTask(ABC):
"""
A print task holds information about what we are looking to print.
"""
def __init__(self, task_type):
self.task_id = self._generate_id()
self.task_type = task_type
self.status = "pending" # pending, processing, completed, failed
print("Created a new " + str(self.task_type) + " with ID " + self.task_id)
@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 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

4
src/templates/index.html
View File

@@ -5,7 +5,7 @@
<div class="card"> <div class="card">
<h3 class="card-header">Print a short message</h3> <h3 class="card-header">Print a short message</h3>
<div class="card-body"> <div class="card-body">
<form class="form-group" action="/api/print/sms" method="post"> <form class="form-group" action="/web/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="txt" placeholder="200 chars or less " maxlength="200" 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"><br>
<input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer"> <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
@@ -18,7 +18,7 @@
<div class="card"> <div class="card">
<h3 class="card-header">Print an image</h3> <h3 class="card-header">Print an image</h3>
<div class="card-body"> <div class="card-body">
<form enctype="multipart/form-data" class="form-group" action="/api/print/img" method="post"> <form enctype="multipart/form-data" class="form-group" action="/web/print/img" method="post">
<input class="form-control" type="file" name="img" required><br> <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"><br>
<input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer"> <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">

70
src/user.py
View File

@@ -1,41 +1,41 @@
class User(object): # class User(object):
"""docstring for User.""" # """docstring for User."""
def __init__(self, arg): # def __init__(self, arg):
super(User, self).__init__() # super(User, self).__init__()
self.arg = arg # self.arg = arg
# @app.route('/login', methods=['POST','GET']) # # @app.route('/login', methods=['POST','GET'])
# @limiter.limit("100 per minute", error_message=error_handler_limiter) # # @limiter.limit("100 per minute", error_message=error_handler_limiter)
def login(): # def login():
if request.method == "POST": # if request.method == "POST":
if not session.get("logged_in"): # if not session.get("logged_in"):
if request.form["username"] and request.form["password"]: # if request.form["username"] and request.form["password"]:
# Get the json # # Get the json
with open("users.json") as f: # with open("users.json") as f:
users_file = json.load(f) # users_file = json.load(f)
for user in users_file["users"]: # for user in users_file["users"]:
if users_file["users"][user] == request.form["password"]: # if users_file["users"][user] == request.form["password"]:
session["logged_in"] = True # session["logged_in"] = True
session["user"] = request.form["username"] # session["user"] = request.form["username"]
if not session.get("logged_in"): # if not session.get("logged_in"):
flash("Mot de passe ou pseudo invalide.", "danger") # flash("Mot de passe ou pseudo invalide.", "danger")
return redirect(url_for("login")) # return redirect(url_for("login"))
else: # else:
return redirect(url_for("display_index_page")) # return redirect(url_for("display_index_page"))
else: # else:
flash("Incorrect logins") # flash("Incorrect logins")
return render_template("password.html") # return render_template("password.html")
else: # else:
return render_template("password.html") # return render_template("password.html")
else: # else:
return render_template("password.html") # return render_template("password.html")
@app.route("/logout") # @app.route("/logout")
def logout(): # def logout():
session["logged_in"] = False # session["logged_in"] = False
flash("Tu est déconnecté", "info") # flash("Tu est déconnecté", "info")
return redirect(url_for("login")) # return redirect(url_for("login"))

130
src/web.py
View File

@@ -1,56 +1,78 @@
from flask import Flask, request, flash from flask import flash
from werkzeug.utils import secure_filename from werkzeug.utils import secure_filename
from printer import Printer
import time import time
import os import os
from task import TextTask, ImageTask, CutTask
class Web(object): class Web(object):
"""docstring for 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, printer): def __init__(self, app, print_queue):
super(Web).__init__() super(Web).__init__()
self.printer = printer self.print_queue = print_queue
self.app = app self.app = app
def print_sms(self, texte, sign: str): def print_sms(self, texte, sign: str) -> bool:
# TODO: verify the texte before printing it here ? """
Get text and a signature, prints the text and cuts after that.
"""
self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign)) self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign))
try: try:
self.printer.print_sms(texte, sign) # We create two new tasks and add them directly to the queue
self.printer.cut() # 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())
except Exception as e: except Exception as e:
self.app.logger.error(e) self.app.logger.error(e)
flash("Error while printing the SMS : " + str(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("You message " + str(texte) + " has been printed :)") def print_image(self, image, sign: str) -> bool:
"""
def print_image(self, image, sign): Get an image and a signature, prints the image and cuts after that.
self.app.logger.debug("Uploading file") """
try: try:
self.app.logger.debug("Uploading file from " + str(sign)) file_uploaded = self.upload_file(image)
if self.upload_file(image): except Exception as e:
self.app.logger.debug("File has been uploaded, printing...") self.app.logger.error(e)
self.printer.print_img( raise RuntimeError("Could not upload file") from e
os.path.join(
self.app.config["UPLOAD_FOLDER"], if file_uploaded:
secure_filename(image.filename), self.app.logger.debug("File has been uploaded, printing...")
), try:
sign=sign, img = self.print_queue.enqueue(
process=True, ImageTask(
os.path.join(
self.app.config["UPLOAD_FOLDER"],
secure_filename(image.filename),
),
signature=sign,
process=True,
)
) )
self.printer.cut()
except Exception as e:
self.app.logger.error(e)
flash("Could not upload file." + str(e))
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: self.app.logger.info("Added two new tasks at position %s and %s", img, cut)
pass
def logout(username: str, password: str) -> bool: return True
pass
def login(self, username: str, password: str) -> bool:
"""Not implemented"""
return
def logout(self, username: str, password: str) -> bool:
"""Not implemented"""
return
def allowed_file(self, filename) -> bool: def allowed_file(self, filename) -> bool:
self.app.logger.debug("Is the filename allowed ?") self.app.logger.debug("Is the filename allowed ?")
@@ -62,22 +84,32 @@ class Web(object):
def upload_file(self, image) -> bool: def upload_file(self, image) -> bool:
self.app.logger.debug("Validating file") self.app.logger.debug("Validating file")
if image and self.allowed_file(image.filename): if image:
filename = secure_filename(image.filename) if self.allowed_file(image.filename):
self.app.logger.debug("File valid") filename = secure_filename(image.filename)
try: self.app.logger.debug("File valid")
image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename)) try:
self.app.logger.debug("File saved") image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
except Exception as e: except OSError as e:
self.app.logger.error("Could not save file") self.app.logger.error("Could not save file %s", e)
flash(str(e), "error") return False
return False
self.app.logger.debug( self.app.logger.debug(
"File saved to " "File saved to "
+ str(os.path.join(self.app.config["UPLOAD_FOLDER"], filename)) + str(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
) )
return True return True
else:
self.app.logger.error(
"Could not save file because the filename is forbidden"
)
return False
else: else:
self.app.logger.error("Could not save file " + str(filename)) self.app.logger.error(
"Could not save file, it seems to be null ? : " + str(filename)
)
return False return False
def get_queue_state(self):
"""Return current queue state"""
return self.print_queue.get_queue_state()

107
src/worker.py Normal file
View File

@@ -0,0 +1,107 @@
# This is the main printing thread
# As explained in the task file, this is where we command
# printing to happen.
import threading
import time
class PrintWorker(threading.Thread):
def __init__(self, app, print_queue, printer, socketio=None):
super().__init__(daemon=True)
self.app = app
self.print_queue = print_queue
self.printer = printer
self.socketio = socketio # Optional
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...")
def run(self):
"""Background thread that processes queue items"""
self.app.logger.info("Worker started working.")
while True:
if not self.running or not self.printer.ready:
time.sleep(0.2)
continue
try:
task = self.print_queue.dequeue()
except Exception as e:
self.app.logger.error("Could not get a new task ! %s ", str(e))
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)
self.state = "printing"
task.status = "processing"
self._emit_status(task.task_id, "processing")
print_data = task.get_print_data()
try:
self.printer.print_task(task.task_type, print_data)
except RuntimeError as e:
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._emit_status(task.task_id, "completed")
except RuntimeError as e:
task.status = "failed"
self.print_queue.mark_completed(task.task_id, "failed")
self._emit_status(task.task_id, "failed", error=str(e))
print(f"Print task {task.task_id} failed: {e}")
else:
# When they are no new tasks to handle, we put the thread to sleep.
self.state = "idle"
time.sleep(0.1)
def _emit_status(self, task_id, status, error=None):
"""Emit status update via Socket.IO if available"""
if not self.socketio:
return
room = f"task_{task_id}"
data = {
"task_id": task_id,
"status": status,
"position": None, # Task no longer in queue
}
if error:
data["error"] = error
self.socketio.emit("task_status", data, room=room)
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,
}