14 Commits

Author SHA1 Message Date
efa6bca56c Refactor for some name changes of libraries 2025-06-22 13:22:14 +02:00
4af20a9a91 Add packaging script 2025-06-20 14:25:15 +02:00
b0bd5f446f Finish V3.1.1 version bump 2025-06-20 14:13:59 +02:00
265288106e Fix issue with config parsing for filtering of cable 2025-06-20 14:07:26 +02:00
Janis Hutz
1c7b758a11 Delete .github/ISSUE_TEMPLATE/custom.md 2025-06-16 15:01:00 +00:00
Janis Hutz
44822e1cc4 Update LICENSE
Remove placeholder
2025-06-16 14:59:41 +00:00
4588caf974 Update README 2025-06-16 16:58:45 +02:00
223ab40bf8 Update README 2025-06-16 16:53:34 +02:00
7905cb851a Config, Lots of docs, Format
Added a config validator and documented code that was previously
undocumented, for the plot_generator scripts, documented them.
2025-06-16 16:36:18 +02:00
3a6cd6af3d Redesign app, prepare for 3.1.0 release 2025-06-16 12:21:45 +02:00
d6a5e90b3c Design temporary save 2025-06-15 19:54:19 +02:00
2b8f3c8aad Start reworking design 2025-06-15 15:33:11 +02:00
d875119071 Start design rework 2025-06-15 12:12:12 +02:00
b01232b552 Get test library running 2025-06-15 12:01:59 +02:00
34 changed files with 1626 additions and 1056 deletions

View File

@@ -1,10 +0,0 @@
---
name: Custom issue template
about: Describe this issue template's purpose here.
title: ''
labels: ''
assignees: ''
---

53
LICENSE
View File

@@ -619,56 +619,3 @@ Program, unless a warranty or assumption of liability accompanies a
copy of the Program in return for a fee. copy of the Program in return for a fee.
END OF TERMS AND CONDITIONS END OF TERMS AND CONDITIONS
How to Apply These Terms to Your New Programs
If you develop a new program, and you want it to be of the greatest
possible use to the public, the best way to achieve this is to make it
free software which everyone can redistribute and change under these terms.
To do so, attach the following notices to the program. It is safest
to attach them to the start of each source file to most effectively
state the exclusion of warranty; and each file should have at least
the "copyright" line and a pointer to where the full notice is found.
<one line to give the program's name and a brief idea of what it does.>
Copyright (C) <year> <name of author>
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
Also add information on how to contact you by electronic and paper mail.
If the program does terminal interaction, make it output a short
notice like this when it starts in an interactive mode:
<program> Copyright (C) <year> <name of author>
This program comes with ABSOLUTELY NO WARRANTY; for details type `show w'.
This is free software, and you are welcome to redistribute it
under certain conditions; type `show c' for details.
The hypothetical commands `show w' and `show c' should show the appropriate
parts of the General Public License. Of course, your program's commands
might be different; for a GUI interface, you would use an "about box".
You should also get your employer (if you work as a programmer) or school,
if any, to sign a "copyright disclaimer" for the program, if necessary.
For more information on this, and how to apply and follow the GNU GPL, see
<https://www.gnu.org/licenses/>.
The GNU General Public License does not permit incorporating your program
into proprietary programs. If your program is a subroutine library, you
may consider it more useful to permit linking proprietary applications with
the library. If this is what you want to do, use the GNU Lesser General
Public License instead of this License. But first, please read
<https://www.gnu.org/licenses/why-not-lgpl.html>.

View File

@@ -26,7 +26,7 @@
BiogasControllerApp has just received a major rewrite, where I focused on code-readability, documentation and stability. The documentation in the code is aimed at beginners and does contain some unnecessary extra comments BiogasControllerApp has just received a major rewrite, where I focused on code-readability, documentation and stability. The documentation in the code is aimed at beginners and does contain some unnecessary extra comments
If you are here to read the code, the files you are most likely looking for can be found in the `lib` folder. If you want to understand and have a look at all of the application, start with the `biogascontrollerapp.py` file ***If you are here to read the code, the files you are most likely looking for can be found in the `util` folder. If you want to understand and have a look at all of the application, start with the `biogascontrollerapp.py` file***
# Installation # Installation
To install it, navigate to the releases tab on the right hand side. Click the current release, scroll down to assets and select the version appropriate for your operating system. To install it, navigate to the releases tab on the right hand side. Click the current release, scroll down to assets and select the version appropriate for your operating system.
@@ -37,6 +37,11 @@ That means:
Compared to older versions, the new BiogasControllerApp doesn't install itself as an app and only resides in a folder where you can launch it using the executable or the `launch.sh` script. Compared to older versions, the new BiogasControllerApp doesn't install itself as an app and only resides in a folder where you can launch it using the executable or the `launch.sh` script.
## Troubleshooting
If you get a warning from Windows, the reason for this is that this app bundle is unsigned (since a signing certificate is about USD 350/year), so it might warn you about that. You can safely click "Run anyway" or the like to bypass that problem.
If this makes you uncomfortable, you may simply install python and install the necessary dependencies (see below) and run the app using Python.
# Features # Features
- Read data the microcontroller in ENATECH sends - Read data the microcontroller in ENATECH sends
- Configure the microcontroller (Coefficients & Temperature). Old settings will be pre-loaded - Configure the microcontroller (Coefficients & Temperature). Old settings will be pre-loaded
@@ -47,7 +52,7 @@ Compared to older versions, the new BiogasControllerApp doesn't install itself a
- Documented code so you can more easily understand what is happening - Documented code so you can more easily understand what is happening
# Issues # Issues
If you encounter any bugs or other weird behaviour, please open an issue on this GitHub repository. If you encounter any bugs or other weird behaviour, please open an issue on this GitHub repository, contact me on my [support page](https://support.janishutz.com) or send me an [email](mailto:development@janishutz.com)
# Documentation # Documentation
You may find documentation for this project in its wiki here on GitHub. The code is also documented with explanations what it does You may find documentation for this project in its wiki here on GitHub. The code is also documented with explanations what it does
@@ -55,15 +60,15 @@ You may find documentation for this project in its wiki here on GitHub. The code
# Officially Supported OS # Officially Supported OS
- Microsoft Windows 10, 11 (through the provided compiled package, might work on older versions as well) - Microsoft Windows 10, 11 (through the provided compiled package, might work on older versions as well)
- Microsoft Windows XP, Vista, 7, 8, 10, 11 (through running the package with Python yourself) - Microsoft Windows XP, Vista, 7, 8, 10, 11 (through running the package with Python yourself)
- MacOS 10.9 (Mavericks) or later (required by Python)
- GNU/Linux: All distros that support Python 3.8 or later (use `install-linux.sh` to install and `launch.sh` to launch for convenience) - GNU/Linux: All distros that support Python 3.8 or later (use `install-linux.sh` to install and `launch.sh` to launch for convenience)
- FreeBSD: If you have Pyhton 3.8 or later installed - FreeBSD: If you have Pyhton 3.8 or later installed
## Dependencies ## Dependencies
Only needed if you run with python directly Only needed if you run with python directly
- Python 3.10 - latest (only tested on this version, but should work down to at least 3.8) - Python 3.10 - latest (only tested on this version, but should work down to at least 3.8)
- kivy[base] - kivy[base]==2.3.1
- pyserial - kivymd==1.1.1
- pyserial==3.5
To install them, run `pip install -r requirements.txt` To install them, run `pip install -r requirements.txt`

View File

@@ -4,11 +4,12 @@
Currently only the newest versions get security updates as security updates are also part of a release. Currently only the newest versions get security updates as security updates are also part of a release.
Only Version 3 is supported due to the poor code quality of V2.3.0 and below. Only Version 3.1 and later are supported due to the poor code quality of V2.3.0 and different UI before.
| Version | Supported | | Version | Supported |
| ------- | ------------------ | | ------- | ------------------ |
| 3.0.0 | ✅ | | 3.1.X | ✅ |
| 3.0.X | ❎ |
| 2.3.0 | ❎ | | 2.3.0 | ❎ |
| 2.2.0 | ❎ | | 2.2.0 | ❎ |
| 2.1.0 | ❎ | | 2.1.0 | ❎ |

View File

@@ -4,7 +4,11 @@
# ╰────────────────────────────────────────────────╯ # ╰────────────────────────────────────────────────╯
# #
# So you would like to read the source code? Nice! # So you would like to read the source code? Nice!
# Just be warned, this application uses Thread and a UI Toolkit called #
# If you simply want to know how the connection stuff works, then head to
# the util/ folder and check out the com.py file!
#
# Just be warned, this application uses Threads and a UI Toolkit called
# Kivy to run. If you are unsure of what functions do, consider # Kivy to run. If you are unsure of what functions do, consider
# checking out the kivy docs at https://kivy.org/doc. # checking out the kivy docs at https://kivy.org/doc.
# It also uses the pyserial library for communication with the micro- # It also uses the pyserial library for communication with the micro-
@@ -12,95 +16,9 @@
# #
# ──────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────
# Load the config file # Print a welcome message
import configparser print(
import time """
config = configparser.ConfigParser()
config.read("./config.ini")
# Introducing tariffs to Python imports.
# It was too funny of an idea to miss out on
# You can enable or disable this in the config.
# It is disabled by default
if config["Tariffs"]["impose_tariffs"] == "True":
try:
import tariff
tariff.set({
"kivy": int(config["Tariffs"]["kivy_rate"]),
"serial": int(config["Tariffs"]["pyserial_rate"]),
})
except Exception as e:
print(e)
print("You cannot evade the tariffs. I will impose impose a tariff of 1000000% on the launch of this app!")
time.sleep(2000000)
import os
from typing import override
from lib.com import Com, ComSuperClass
import lib.test.com
# Load config and disable kivy log if necessary
if config["Dev"]["verbose"] == "True":
pass
else:
os.environ["KIVY_NO_CONSOLELOG"] = "1"
# Load kivy modules. Kivy is the UI framework used. See https://kivy.org
# from kivy.core.window import Window, Config
from kivy.uix.screenmanager import ScreenManager
from kivy.app import App
# Store the current app version
app_version = f"{config['Info']['version']}{config['Info']['subVersion']}"
# ╭────────────────────────────────────────────────╮
# │ Screens │
# ╰────────────────────────────────────────────────╯
# Import all the screens (= pages) used in the app
from gui.home.home import HomeScreen
from gui.credits.credits import CreditsScreen
from gui.program.program import ProgramScreen
from gui.about.about import AboutScreen
from gui.main.main import MainScreen
# ╭────────────────────────────────────────────────╮
# │ Screen Manager │
# ╰────────────────────────────────────────────────╯
# Kivy uses a screen manager to manage pages in the application
class BiogasControllerApp(App):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.screen_manager = ScreenManager()
@override
def build(self):
com: ComSuperClass = Com()
if config["Dev"]["use_test_library"] == "True":
com = lib.test.com.Com()
self.icon = "./BiogasControllerAppLogo.png"
self.title = "BiogasControllerApp-" + app_version
self.screen_manager.add_widget(HomeScreen(com, name="home"))
self.screen_manager.add_widget(MainScreen(com, name="main"))
self.screen_manager.add_widget(ProgramScreen(com, name="program"))
self.screen_manager.add_widget(CreditsScreen(name="credits"))
self.screen_manager.add_widget(AboutScreen(name="about"))
return self.screen_manager
# Disallow this file to be imported
if __name__ == "__main__":
print("""
┏━━┓━━━━━━━━━━━━━━━━━━━━┏━━━┓━━━━━━━━━┏┓━━━━━━━━┏┓━┏┓━━━━━━━━┏━━━┓━━━━━━━━ ┏━━┓━━━━━━━━━━━━━━━━━━━━┏━━━┓━━━━━━━━━┏┓━━━━━━━━┏┓━┏┓━━━━━━━━┏━━━┓━━━━━━━━
┃┏┓┃━━━━━━━━━━━━━━━━━━━━┃┏━┓┃━━━━━━━━┏┛┗┓━━━━━━━┃┃━┃┃━━━━━━━━┃┏━┓┃━━━━━━━━ ┃┏┓┃━━━━━━━━━━━━━━━━━━━━┃┏━┓┃━━━━━━━━┏┛┗┓━━━━━━━┃┃━┃┃━━━━━━━━┃┏━┓┃━━━━━━━━
┃┗┛┗┓┏┓┏━━┓┏━━┓┏━━┓━┏━━┓┃┃━┗┛┏━━┓┏━┓━┗┓┏┛┏━┓┏━━┓┃┃━┃┃━┏━━┓┏━┓┃┃━┃┃┏━━┓┏━━┓ ┃┗┛┗┓┏┓┏━━┓┏━━┓┏━━┓━┏━━┓┃┃━┗┛┏━━┓┏━┓━┗┓┏┛┏━┓┏━━┓┃┃━┃┃━┏━━┓┏━┓┃┃━┃┃┏━━┓┏━━┓
@@ -110,9 +28,181 @@ if __name__ == "__main__":
━━━━━━━━━━━┏━┛┃━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┃┃━━┃┃━━ ━━━━━━━━━━━┏━┛┃━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┃┃━━┃┃━━
━━━━━━━━━━━┗━━┛━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┗┛━━┗┛━━ ━━━━━━━━━━━┗━━┛━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┗┛━━┗┛━━
Version 3.0.0 Version 3.2.0
=> Initializing.... => Initializing....
""") """
)
# Load the config file
import time
from util.config import read_config, set_verbosity, str_to_bool
verbose = str_to_bool(read_config("Dev", "verbose", "False", type_to_validate="bool"))
verbose = verbose if verbose != None else False
# Introducing tariffs to Python imports.
# It was too funny of an idea to miss out on
# You can enable or disable this in the config.
# It is disabled by default
if str_to_bool(
read_config("Tariffs", "impose_tariffs", "False", type_to_validate="bool")
):
try:
import tariff
tariff.set(
{
"kivy": int(
read_config("Tariffs", "kivy_rate", "0", type_to_validate="int")
),
"serial": int(
read_config("Tariffs", "pyserial_rate", "0", type_to_validate="int")
),
}
)
except Exception as e:
print(e)
print(
"You cannot evade the tariffs. I will impose impose a tariff of 1000000% on the launch of this app!"
)
time.sleep(2000000)
import os
from typing import override
from util.com import Com
from util.interface import ControllerConnection
import util.test.com
# Load config and disable kivy log if necessary
if verbose:
pass
else:
os.environ["KIVY_NO_CONSOLELOG"] = "1"
# Load kivy modules. Kivy is the UI framework used. See https://kivy.org
from kivy.core.window import Window
from kivy.uix.screenmanager import ScreenManager
from kivymd.app import MDApp
# Set Window size
Window.size = (
int(int(read_config("UI", "width", "800", type_to_validate="int"))),
int(int(read_config("UI", "height", "600", type_to_validate="int"))),
)
# ╭────────────────────────────────────────────────╮
# │ Screens │
# ╰────────────────────────────────────────────────╯
# Import all the screens (= pages) used in the app
from gui.home.home import HomeScreen
from gui.program.program import ProgramScreen
from gui.about.about import AboutScreen
from gui.main.main import MainScreen
# ╭────────────────────────────────────────────────╮
# │ Screen Manager │
# ╰────────────────────────────────────────────────╯
# Kivy uses a screen manager to manage pages in the application
colors = [
"Red",
"Pink",
"Purple",
"DeepPurple",
"Indigo",
"Blue",
"LightBlue",
"Cyan",
"Teal",
"Green",
"LightGreen",
"Lime",
"Yellow",
"Amber",
"Orange",
"DeepOrange",
"Brown",
"Gray",
"BlueGray",
]
class BiogasControllerApp(MDApp):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.screen_manager = ScreenManager()
@override
def build(self):
# Configure com
filters = [
x.strip()
for x in read_config(
"Connection",
"filters",
"USB-Serial Controller; Prolific USB-Serial Controller",
).split(";")
]
baudrate = int(
read_config("Connection", "baudrate", "19200", type_to_validate="int")
)
com: ControllerConnection = Com(
baudrate,
filters,
)
if str_to_bool(
read_config("Dev", "use_test_library", "False", type_to_validate="bool")
):
com = util.test.com.Com(
int(read_config("Dev", "fail_sim", "20", type_to_validate="int")),
baudrate,
filters,
)
com.set_port_override(read_config("Connection", "port_override", "None"))
self.theme_cls.theme_style = read_config(
"UI", "theme", "Dark", ["Dark", "Light"]
)
self.theme_cls.material_style = "M3"
self.theme_cls.primary_palette = read_config(
"UI", "primary_color", "Green", colors
)
self.theme_cls.accent_palette = read_config(
"UI", "accent_color", "Lime", colors
)
self.theme_cls.theme_style_switch_animation = False
if verbose:
print("\n", "-" * 20, "\n")
self.icon = "./BiogasControllerAppLogo.png"
self.title = "BiogasControllerApp-V3.2.0"
self.screen_manager.add_widget(HomeScreen(com, name="home"))
self.screen_manager.add_widget(MainScreen(com, name="main"))
self.screen_manager.add_widget(ProgramScreen(com, name="program"))
self.screen_manager.add_widget(AboutScreen(name="about"))
return self.screen_manager
def change_theme(self):
self.theme_cls.theme_style = (
"Dark" if self.theme_cls.theme_style == "Light" else "Light"
)
# Disallow this file to be imported
if __name__ == "__main__":
set_verbosity(verbose)
# Start the application
BiogasControllerApp().run() BiogasControllerApp().run()
print("\n => Exiting!") print("\n => Exiting!")

View File

@@ -1,11 +1,32 @@
***CHANGELOG*** ***CHANGELOG***
V3.2.0
- Fixed a bug with comport assignment
- Refactored some naming
- Added more comments
V3.0-beta V3.1.0
- Redesigned GUI - Completely redesigned User Interface using KivyMD
- Added config option for themes
V3.0.1
- Install script fixes
- Packaging fixes
V3.0.0
- Small UI fixes
- Consolidated multiple previously separate screens - Consolidated multiple previously separate screens
- Completely rewritten backend - Completely rewritten backend
- Improved stability - Improved stability
- Cleaned, documented code - Cleaned, documented code
- Reduced overhead of connecting
- Improved hooking reliability
- Removed installer, simpler setup now possible
- Removed official MacOS support as it didn't really work before anyway
- Added additional config options
- Improved linguistics
- Bugfixes
OLD VERSIONS OLD VERSIONS
------------ ------------

View File

@@ -1,21 +1,25 @@
[Ports] [Connection]
specificport = None port_override = None
baudrate = 19200
# List the names as which the adapter cable will show up separated by commas below
# For ENATECH, the below is likely correct. The name cannot contain a semicolon
filters = USB-Serial Controller; Prolific USB-Serial Controller
[UI] [UI]
sizeh = 600 height = 600
sizew = 800 width = 800
# Can be Dark or Light
theme = Dark
primary_color = Green
accent_color = Lime
[Dev] [Dev]
verbose = True verbose = False
log_level = DEBUG
use_test_library = False use_test_library = False
# One time out of how many (plus one) it should fail
fail_sim = 20
[Tariffs] [Tariffs]
impose_tariffs = False impose_tariffs = False
kivy_rate = 50 kivy_rate = 0
pyserial_rate = 500 pyserial_rate = 0
[Info]
version = V2.3.0
subversion =

View File

@@ -1,3 +0,0 @@
from gui.popups.popups import *

5
gui/README.md Normal file
View File

@@ -0,0 +1,5 @@
# GUI
This folder contains all files that are used for the GUI of the app.
It is written in KivyMD, so if you don't know what that is and you don't want to learn it,
there isn't much of use in here for you! - Just so you're warned

View File

@@ -1,37 +1,52 @@
<AboutScreen>: <AboutScreen>:
name: "about" name: "about"
canvas.before: MDFloatLayout:
Color: Image:
rgba: (10,10,10,0.1) source: "BiogasControllerAppLogo.png"
Rectangle: pos_hint: {"top": 0.9}
size: self.size size_hint_y: .3
pos: self.pos radius: 36, 36, 0, 0
GridLayout: allow_stretch: True
keep_ratio: True
MDGridLayout:
cols: 1 cols: 1
Label: MDLabel:
text: "About" text: "About"
font_size: 40 font_size: 40
color: (0, 113, 0, 1) halign: 'center'
valign: 'center'
bold: True bold: True
FloatLayout: italic: True
GridLayout: theme_text_color: 'Secondary'
pos_hint: {"x":0.05, "y":0.05} pos_hint: {'center_x': 0, 'center_y': 0}
size_hint: 0.9, 0.9
cols: 3 MDFillRoundFlatButton:
Button: pos_hint: {'x': 0.1, 'y': 0.05}
text: "Back" text: "Back"
background_color: (255,0,0,0.6)
on_release: on_release:
app.root.current = "home" app.root.current = "home"
root.manager.transition.direction = "up" root.manager.transition.direction = "up"
Button:
text: "Report a\nBug" MDFillRoundFlatButton:
background_color: (255,0,0,0.6) pos_hint: {'right': 0.9, 'y': 0.05}
text: "Report a Bug"
on_release: on_release:
root.report_issue() root.goto("issues")
Button:
text: "Credits" MDFillRoundFlatButton:
background_color: (255,0,0,0.6) pos_hint: {'right': 0.48, 'y': 0.05}
text: "Wiki"
on_release: on_release:
app.root.current = "credits" root.goto("wiki")
root.manager.transition.direction = "left"
MDFillRoundFlatButton:
pos_hint: {'x': 0.52, 'y': 0.05}
text: "Repo"
on_release:
root.goto("repo")
Label:
text: "This is a simple controller application that allows you to read data from and configure the microcontroller used in ENATECH at KSWO. It is written in Python using KivyMD as its UI framework.\n\nThis software is free Software licensed under the GNU General Public License Version 3 and as such comes with absolutely no warranty."
pos_hint: {'x': 0.05, 'top': 0.42}
text_size: self.width, None
size_hint: 0.9, None

View File

@@ -1,13 +1,38 @@
from kivy.uix.screenmanager import Screen from kivy.uix.screenmanager import Screen
from kivymd.uix.dialog import MDDialog
from kivymd.uix.button import MDFlatButton
from kivy.lang import Builder from kivy.lang import Builder
import webbrowser import webbrowser
from gui.popups.popups import SingleRowPopup # Simple about screen
class AboutScreen(Screen): class AboutScreen(Screen):
def report_issue(self): def __init__(self, **kw):
SingleRowPopup().open("Opened your web-browser") # Prepare dialog
webbrowser.open('https://github.com/janishutz/BiogasControllerApp/issues', new=2) self.opened_web_browser_dialog = MDDialog(
title="Open Link",
text="Your webbrowser has been opened. Continue there",
buttons=[
MDFlatButton(
text="Ok",
on_release=lambda _: self.opened_web_browser_dialog.dismiss(),
),
],
)
super().__init__(**kw)
Builder.load_file('./gui/about/about.kv') def goto(self, loc: str):
# Open web browser with links
if loc == "wiki":
webbrowser.open(
"https://github.com/janishutz/BiogasControllerApp/wiki", new=2
)
elif loc == "issues":
webbrowser.open(
"https://github.com/janishutz/BiogasControllerApp/issues", new=2
)
elif loc == "repo":
webbrowser.open("https://github.com/janishutz/BiogasControllerApp", new=2)
self.opened_web_browser_dialog.open()
Builder.load_file("./gui/about/about.kv")

View File

@@ -1,27 +0,0 @@
<CreditsScreen>:
name: "credits"
canvas.before:
Color:
rgba: (10,10,10,0.1)
Rectangle:
size: self.size
pos: self.pos
FloatLayout:
Button:
text: "back"
size_hint: 0.4, 0.2
pos_hint: {"x":0.3, "y":0.1}
on_release:
app.root.current = "about"
root.manager.transition.direction = "right"
GridLayout:
cols:1
pos_hint:{"x":0.05, "y":0.35}
size_hint: 0.9, 0.5
Label:
text: "This is a controller sofware that helps you reprogram and monitor the micro-controller used in ENATECH at KSWO"
Label:
text: "Written by: Janis Hutz\nDesigned by: Janis Hutz\nDesign language: Kivy"
Label:
text: "This software is free Software licensed under the GPL V3 (GNU General Public License) and as such comes with absolutely no warranty. In return, you can use, modify, distribute or use any of the code of this software in your own project, if you reuse the same license. For more infos, you can find a copy of this license in the project folder."
text_size: self.width, None

View File

@@ -1,8 +0,0 @@
from kivy.uix.screenmanager import Screen
from kivy.lang import Builder
class CreditsScreen(Screen):
pass
Builder.load_file('./gui/credits/credits.kv')

View File

@@ -1,45 +1,62 @@
<HomeScreen>: <HomeScreen>:
name: "home" name: "home"
canvas.before: MDFloatLayout:
Color: Image:
rgba: (10,10,10,0.1) source: "BiogasControllerAppLogo.png"
Rectangle: pos_hint: {"top": 0.9}
size: self.size size_hint_y: .3
pos: self.pos radius: 36, 36, 0, 0
GridLayout: allow_stretch: True
cols:1 keep_ratio: True
Label:
text: "BiogasanlageControllerApp" MDGridLayout:
cols: 1
MDLabel:
text: "BiogasControllerApp"
font_size: 50 font_size: 50
color: (0, 113, 0, 1) halign: 'center'
bold:True valign: 'center'
italic:True bold: True
FloatLayout: italic: True
GridLayout: theme_text_color: 'Secondary'
pos_hint: {'center_x': 0, 'center_y': 0}
MDGridLayout:
spacing: 20
size_hint: None, None
size: self.minimum_size
cols: 2 cols: 2
size_hint: 0.8, 0.8 pos_hint: {'center_x': 0.5, 'center_y': 0.3 }
pos_hint: {"x": 0.1, "y": 0.1} MDFillRoundFlatButton:
Button: font_size: 30
text: "Start" text: "Start"
background_color: (255, 0, 0, 0.6) on_release: root.start()
font_size: 30
on_release: MDFillRoundFlatButton:
root.start()
Button:
text: "Quit" text: "Quit"
background_color: (255, 0, 0, 0.6)
font_size: 30 font_size: 30
on_release: pos_hint: {"x": 0.7, "center_y": 0}
root.quit() on_release: root.quit()
Label:
text: "You are running version V3.0.1" MDLabel:
text: "You are running version V3.2.0"
font_size: 13 font_size: 13
pos_hint: {"y": -0.45, "x":0.05} pos_hint: {"y": -0.45, "x":0}
Button: halign: 'center'
MDFlatButton:
text: "About" text: "About"
font_size: 13 font_size: 13
size_hint: 0.07, 0.06 size_hint: 0.07, 0.06
pos_hint: {"x":0.01, "y":0.01} pos_hint: {"x":0.01, "y":0.01}
background_color: (50, 0, 0, 0.2)
on_release: on_release:
root.to_about() root.to_about()
# MDFlatButton:
# text: "Change Theme"
# font_size: 13
# size_hint: 0.07, 0.06
# pos_hint: {"right":0.99, "y":0.01}
# on_release:
# app.change_theme()

View File

@@ -1,71 +1,126 @@
from kivy.uix.screenmanager import Screen from kivy.base import Clock
from kivymd.app import MDApp
from kivymd.uix.button import MDFlatButton
from kivymd.uix.dialog import MDDialog
from kivymd.uix.screen import MDScreen
from kivy.lang import Builder from kivy.lang import Builder
from gui.popups.popups import DualRowPopup, QuitPopup, TwoActionPopup
from lib.com import ComSuperClass
import platform import platform
from util.interface import ControllerConnection
# Information for errors encountered when using pyserial # Information for errors encountered when using pyserial
information = { information = {
"Windows": { "Windows": {
"2": "Un- and replug the cable and ensure you have the required driver(s) installed", "2": "Un- and replug the cable and ensure you have the required driver(s) installed",
"13": "You are probably missing a required driver or your cable doesn't work. Consult the wiki for more information", "13": "You are probably missing a required driver or your cable doesn't work. Consult the wiki for more information",
"NO_COM": "Could not find a microcontroller. Please ensure you have one connected and the required driver(s) installed" "NO_COM": "Could not find a microcontroller. Please ensure you have one connected and the required driver(s) installed",
}, },
"Linux": { "Linux": {
"2": "Un- and replug the cable, or if you haven't plugged a controller in yet, do that", "2": "Un- and replug the cable, or if you haven't plugged a controller in yet, do that",
"13": "Incorrect permissions at /dev/ttyUSB0. Open a terminal and type: sudo chmod 777 /dev/ttyUSB0", "13": "Incorrect permissions at /dev/ttyUSB0. Open a terminal and type: sudo chmod 777 /dev/ttyUSB0",
"NO_COM": "Could not find a microcontroller. Please ensure you have one connected" "NO_COM": "Could not find a microcontroller. Please ensure you have one connected",
} },
} }
# This is the launch screen, i.e. what you see when you start up the app # This is the launch screen, i.e. what you see when you start up the app
class HomeScreen(Screen): class HomeScreen(MDScreen):
def __init__(self, com: ComSuperClass, **kw): def __init__(self, com: ControllerConnection, **kw):
self._com = com; self._com = com
self.connection_error_dialog = MDDialog(
title="Connection",
text="Failed to connect. See Details for more information and troubleshooting guide",
buttons=[
MDFlatButton(
text="Cancel",
on_release=lambda _: self.connection_error_dialog.dismiss(),
),
MDFlatButton(
text="Details", on_release=lambda _: self.open_details_popup()
),
],
)
self.quit_dialog = MDDialog(
title="Exit BiogasControllerApp",
text="Do you really want to exit BiogasControllerApp?",
buttons=[
MDFlatButton(
text="Cancel",
on_release=lambda _: self.quit_dialog.dismiss(),
),
MDFlatButton(text="Quit", on_release=lambda _: self._quit()),
],
)
super().__init__(**kw) super().__init__(**kw)
def _quit(self):
self._com.close()
MDApp.get_running_app().stop()
def start(self):
Clock.schedule_once(lambda _: self._start())
# Go to the main screen if we can establish connection or the check was disabled # Go to the main screen if we can establish connection or the check was disabled
# in the configs # in the configs
def start(self): def _start(self):
if self._com.connect(): if self._com.connect():
self.manager.current = 'main' self.manager.current = "main"
self.manager.transition.direction = 'right' self.manager.transition.direction = "right"
else: else:
TwoActionPopup().open('Failed to connect', 'Details', self.open_details_popup) self.connection_error_dialog.open()
print('ERROR connecting') print("[ COM ] Connection failed!")
# Open popup for details as to why the connection failed # Open popup for details as to why the connection failed
def open_details_popup(self): def open_details_popup(self):
DualRowPopup().open("Troubleshooting tips", self._generate_help()) self.connection_error_dialog.dismiss()
self.details_dialog = MDDialog(
title="Troubleshooting",
text=self._generate_help(),
buttons=[
MDFlatButton(
text="Ok", on_release=lambda _: self.details_dialog.dismiss()
)
],
)
self.details_dialog.open()
def _generate_help(self) -> str: def _generate_help(self) -> str:
operating_system = platform.system() operating_system = platform.system()
if operating_system == "Windows" or operating_system == "Linux": if operating_system == "Windows" or operating_system == "Linux":
port = self._com.get_comport(); port = self._com.get_comport()
information["Linux"]["13"] = f"Incorrect permissions at {port}. Resolve by running 'sudo chmod 777 {port}'" if port == "Sim":
return "Running in simulator, so this error is just simulated"
information["Linux"][
"13"
] = f"Incorrect permissions at {port}. Resolve by running 'sudo chmod 777 {port}'"
if port == "": if port == "":
return information[operating_system]["NO_COM"] return information[operating_system]["NO_COM"]
err = self._com.get_error() err = self._com.get_error()
if err != None: if err != None:
return information[operating_system][str(err.errno)] return information[operating_system][str(err.errno)]
else: else:
return "No error message available" return "No error message available"
else: else:
return "You are running on an unsupported Operating System. No help available" return (
"You are running on an unsupported Operating System. No help available"
)
# Helper to open a Popup to ask user whether to quit or not # Helper to open a Popup to ask user whether to quit or not
def quit(self): def quit(self):
QuitPopup(self._com).open() self.quit_dialog.open()
# Switch to about screen # Switch to about screen
def to_about(self): def to_about(self):
self.manager.current = 'about' self.manager.current = "about"
self.manager.transition.direction = 'down' self.manager.transition.direction = "down"
# Load the design file for this screen (.kv files) # Load the design file for this screen (.kv files)
# The path has to be relative to root of the app, i.e. where the biogascontrollerapp.py # The path has to be relative to root of the app, i.e. where the biogascontrollerapp.py
# file is located # file is located
Builder.load_file('./gui/home/home.kv') Builder.load_file("./gui/home/home.kv")

View File

@@ -1,89 +1,124 @@
<MainScreen>: <MainScreen>:
on_pre_enter: root.reset() on_pre_enter: root.reset()
name: "main" name: "main"
canvas.before: MDFloatLayout:
Color: MDGridLayout:
rgba: (10,10,10,0.1) cols: 1
Rectangle: pos_hint: {'x': 0, 'y': 0.4}
size: self.size MDLabel:
pos: self.pos
GridLayout:
FloatLayout:
Label:
pos_hint: {"y":0.4}
text: "READOUT" text: "READOUT"
font_size: 40 font_size: 40
color: (0, 113, 0, 1) halign: 'center'
valign: 'center'
pos_hint: {'center_x': 0, 'center_y': 0}
bold: True bold: True
GridLayout:
MDGridLayout:
cols:4 cols:4
size_hint: 0.8, 0.3 size_hint: 0.8, 0.3
pos_hint: {"x":0.1, "y":0.4} pos_hint: {"x":0.1, "y":0.4}
Label: MDLabel:
text: "Sensor 1: " text: "Sensor 1: "
font_size: 20 font_size: 20
Label: MDLabel:
id: sensor1 id: sensor1
text: "" text: ""
Label: size_hint: 1, 1
halign: 'left'
text_size: self.size
MDLabel:
text: "Sensor 2: " text: "Sensor 2: "
font_size: 20 font_size: 20
Label: MDLabel:
id: sensor2 id: sensor2
text: "" text: ""
Label: size_hint: 1, 1
halign: 'left'
text_size: self.size
MDLabel:
text: "Sensor 3: " text: "Sensor 3: "
font_size: 20 font_size: 20
Label: MDLabel:
id: sensor3 id: sensor3
text: "" text: ""
Label: size_hint: 1, 1
halign: 'left'
text_size: self.size
MDLabel:
text: "Sensor 4: " text: "Sensor 4: "
font_size: 20 font_size: 20
Label: MDLabel:
id: sensor4 id: sensor4
text: "" text: ""
Button: size_hint: 1, 1
halign: 'left'
text_size: self.size
MDFillRoundFlatButton:
text: "Connect" text: "Connect"
size_hint: 0.2, 0.1 size_hint: 0.15, 0.09
pos_hint: {"x": 0.5, "y": 0.05} pos_hint: {"x": 0.03, "y": 0.05}
background_color: (255, 0, 0, 0.6)
on_release: on_release:
root.start() root.start()
Button:
MDFillRoundFlatButton:
text: "Disconnect" text: "Disconnect"
size_hint: 0.2, 0.1 size_hint: 0.15, 0.09
pos_hint: {"x": 0.7, "y": 0.05} pos_hint: {"x": 0.2, "y": 0.05}
background_color: (255, 0, 0, 0.6)
on_release: on_release:
root.end() root.end()
Button:
MDFillRoundFlatButton:
text: "Back" text: "Back"
size_hint: 0.3, 0.1 size_hint: 0.15, 0.09
pos_hint: {"x":0.05, "y":0.05} pos_hint: {"right": 0.95, "y":0.05}
background_color: (255, 0, 0, 0.6) md_bg_color: app.theme_cls.primary_dark
on_release: on_release:
root.end() root.end()
app.root.current = "home" app.root.current = "home"
root.manager.transition.direction = "left" root.manager.transition.direction = "left"
ToggleButton:
MDGridLayout:
cols: 2
size_hint: 0.15, 0.1
pos_hint: {"x":0.1, "y":0.15}
MDLabel:
text: "Fast Mode"
valign: "center"
MDSwitch:
id: mode_selector id: mode_selector
size_hint: 0.15, 0.1 on_active: root.switch_mode()
pos_hint: {"x":0.1, "y":0.2} icon_active: "check"
text: "Normal Mode" if self.state == "normal" else "Fast Mode"
on_text: root.switch_mode(mode_selector.text) MDFillRoundFlatButton:
background_color: (255,0,0,0.6) if self.state == "normal" else (0,0,255,0.6)
Button:
text: "Configuration" text: "Configuration"
size_hint: 0.15, 0.1 size_hint: 0.1, 0.07
pos_hint: {"x":0.7, "y":0.2} pos_hint: {"x":0.45, "y":0.06}
background_color: (255, 0, 0, 0.6) md_bg_color: app.theme_cls.accent_dark
on_release: on_release:
root.end() root.end()
app.root.current = "program" app.root.current = "program"
root.manager.transition.direction = "down" root.manager.transition.direction = "down"
Label:
MDGridLayout:
size_hint: 0.2, None
spacing: 0
padding: 0
cols: 1
pos_hint: {'right': 0.95, 'top': 0.95}
MDLabel:
id: status id: status
text: "Status will appear here" text: "Status will appear here"
font_size: 10 font_size: 10
pos_hint: {"x":0.4, "y": 0.3} halign: 'right'
MDGridLayout:
size_hint: None, None
spacing: 0
padding: 0
cols: 1
pos_hint: {'right': 0.95, 'top': 0.925}
MDLabel:
id: port
text: "Port: Not connected"
font_size: 10
halign: 'right'

View File

@@ -1,17 +1,18 @@
from ctypes import ArgumentError from ctypes import ArgumentError
from time import time from time import time
from typing import List, override from typing import List, override
from kivy.uix.screenmanager import Screen from kivymd.uix.screen import MDScreen
from kivy.lang import Builder from kivy.lang import Builder
from gui.popups.popups import SingleRowPopup, TwoActionPopup, empty_func
from kivy.clock import Clock, ClockEvent from kivy.clock import Clock, ClockEvent
from kivymd.uix.button import MDFlatButton
from kivymd.uix.dialog import MDDialog
import queue import queue
import threading import threading
# Load utilities # Load utilities
from lib.instructions import Instructions from util.instructions import Instructions
from lib.com import ComSuperClass from util.interface import ControllerConnection
from lib.decoder import Decoder from util.decoder import Decoder
# TODO: Consider consolidating start and stop button # TODO: Consider consolidating start and stop button
@@ -26,13 +27,13 @@ synced_queue: queue.Queue[List[str]] = queue.Queue()
# ╰────────────────────────────────────────────────╯ # ╰────────────────────────────────────────────────╯
# Using a Thread to run this in parallel to the UI to improve responsiveness # Using a Thread to run this in parallel to the UI to improve responsiveness
class ReaderThread(threading.Thread): class ReaderThread(threading.Thread):
_com: ComSuperClass _com: ControllerConnection
_decoder: Decoder _decoder: Decoder
_instructions: Instructions _instructions: Instructions
# This method allows the user to set Com object to be used. # This method allows the user to set Com object to be used.
# The point of this is to allow for the use of a single Com object to not waste resources # The point of this is to allow for the use of a single Com object to not waste resources
def set_com(self, com: ComSuperClass): def set_com(self, com: ControllerConnection):
"""Set the Com object to be used in this """Set the Com object to be used in this
Args: Args:
@@ -53,7 +54,7 @@ class ReaderThread(threading.Thread):
# Hook to output stream # Hook to output stream
if self._instructions.hook_main(): if self._instructions.hook_main():
# We are now hooked to the stream (i.e. data is synced) # We are now hooked to the stream (i.e. data is synced)
synced_queue.put(["HOOK"]) synced_queue.put(["HOOK", self._com.get_comport()])
# making it exit using the stop function # making it exit using the stop function
while self._run: while self._run:
@@ -70,17 +71,23 @@ class ReaderThread(threading.Thread):
for i in range(4): for i in range(4):
# The slicing that happens here uses offsets automatically calculated from the sensor id # The slicing that happens here uses offsets automatically calculated from the sensor id
# This allows for short code # This allows for short code
try:
data.append( data.append(
f"Tadc: { f"Tadc: {
self._decoder.decode_int(received[12 * i:12 * i + 4]) self._decoder.decode_int(received[12 * i:12 * i + 4])
}\nTemperature: { }\nTemp: {
self._decoder.decode_float(received[12 * i + 5:12 * i + 11]) round(self._decoder.decode_float(received[12 * i + 5:12 * i + 11]) * 1000) / 1000
}\nDuty-Cycle: { }°C\nDC: {
self._decoder.decode_float_long(received[48 + 5 * i: 52 + 5 * i]) / 65535.0 * 100 round((self._decoder.decode_float_long(received[48 + 5 * i: 52 + 5 * i]) / 65535.0 * 100) * 1000) / 1000
}%" }%"
) )
except:
data.append("Bad data")
# Calculate the frequency of updates # Calculate the frequency of updates
data.append(str(1 / (time() - start_time))) data.append(
str(round((1 / (time() - start_time)) * 1000) / 1000) + " Hz"
)
synced_queue.put(data)
else: else:
# Send error message to the UI updater # Send error message to the UI updater
synced_queue.put(["ERR_HOOK"]) synced_queue.put(["ERR_HOOK"])
@@ -94,45 +101,87 @@ class ReaderThread(threading.Thread):
# │ Main App Screen │ # │ Main App Screen │
# ╰────────────────────────────────────────────────╯ # ╰────────────────────────────────────────────────╯
# This is the main screen, where you can read out data # This is the main screen, where you can read out data
class MainScreen(Screen): class MainScreen(MDScreen):
_event: ClockEvent _event: ClockEvent
# The constructor if this class takes a Com object to share one between all screens # The constructor if this class takes a Com object to share one between all screens
# to preserve resources and make handling better # to preserve resources and make handling better
def __init__(self, com: ComSuperClass, **kw): def __init__(self, com: ControllerConnection, **kw):
# Set some variables # Set some variables
self._com = com self._com = com
self._event = None self._event = None
self._fast_mode = False
# Set up Dialog for erros
self.connection_error_dialog = MDDialog(
title="Connection",
text="Failed to connect. Do you wish to retry?",
buttons=[
MDFlatButton(
text="Cancel",
on_release=lambda _: self.connection_error_dialog.dismiss(),
),
MDFlatButton(text="Retry", on_release=lambda _: self.start()),
],
)
self.mode_switch_error_dialog = MDDialog(
title="Mode Switch",
text="Failed to change mode. Please try again",
buttons=[
MDFlatButton(
text="Ok",
on_release=lambda _: self.mode_switch_error_dialog.dismiss(),
),
],
)
# Prepare the reader thread # Prepare the reader thread
self._reader = ReaderThread() self._prepare_reader()
self._reader.setDaemon(True) self._has_run = False
self._reader.set_com(com)
self._has_connected = False self._has_connected = False
# Call the constructor for the Screen class # Call the constructor for the Screen class
super().__init__(**kw) super().__init__(**kw)
# Start the connection to the micro-controller to read data from it. def _prepare_reader(self):
# This also now starts the reader thread to continuously read out data # Prepares the reader thread
self._reader = ReaderThread()
self._reader.daemon = True
self._reader.set_com(self._com)
# Small helper function that makes the UI not freeze by offloading
def start(self): def start(self):
Clock.schedule_once(lambda _: self._start())
# Start the connection to the micro-controller to read data from it.
# This also starts the reader thread to continuously read out data
def _start(self):
# Prevent running multiple times
self.connection_error_dialog.dismiss()
if self._has_connected:
return
# Some UI config
self.ids.status.text = "Connecting..." self.ids.status.text = "Connecting..."
if self._com.connect(): if self._com.connect():
print("Acquired connection") print("[ COM ] Connection Acquired")
# Prevent multiple connections
self._has_connected = True self._has_connected = True
self._has_run = True
if self._has_run:
self._prepare_reader()
# Start communication # Start communication
self._reader.start() self._reader.start()
print("Reader has started") print("[ COM ] Reader has started")
Clock.schedule_interval(self._update_screen, 0.5)
# Schedule UI updates
self._event = Clock.schedule_interval(self._update_screen, 0.5)
else: else:
self.ids.status.text = "Connection failed" self.ids.status.text = "Connection failed"
TwoActionPopup().open( self.connection_error_dialog.open()
"Failed to connect. Do you want to retry?",
"Cancel",
empty_func,
"Retry",
self.start,
)
# End connection to micro-controller and set it back to normal mode # End connection to micro-controller and set it back to normal mode
def end(self, set_msg: bool = True): def end(self, set_msg: bool = True):
@@ -142,32 +191,52 @@ class MainScreen(Screen):
if self._event != None: if self._event != None:
self._event.cancel() self._event.cancel()
self._reader.stop() self._reader.stop()
# Join the thread to end it safely
try:
self._reader.join()
except:
pass
# Go back to Normal Mode on the Controller
# This is so you don't accidentally forget!
try: try:
self._com.send("NM") self._com.send("NM")
except: except:
pass pass
self._com.close() self._com.close()
if set_msg: if set_msg:
self.ids.status.text = "Connection terminated" self.ids.status.text = "Connection terminated"
self.ids.port.text = "Port: Not connected"
self._has_connected = False
print("Connection terminated") print("Connection terminated")
# A helper function to update the screen. Is called on an interval # A helper function to update the screen. Is called on an interval
def _update_screen(self, dt): def _update_screen(self, _):
update = [] update = []
try: try:
update = synced_queue.get_nowait() update = synced_queue.get_nowait()
except: except:
pass pass
if len(update) == 0: if len(update) == 0:
# There are no updates to process, don't block and simply try again next time # There are no updates to process, don't block and simply try again next time
return return
if len(update) == 1: if len(update) == 1:
# Sync errors
if update[0] == "ERR_HOOK": if update[0] == "ERR_HOOK":
self.ids.status.text = "Hook failed" self.ids.status.text = "Hook failed"
self.end(False) self.end(False)
elif update[0] == "HOOK":
if len(update) == 2:
# Connection successful
if update[0] == "HOOK":
self.ids.status.text = "Connected to controller" self.ids.status.text = "Connected to controller"
self.ids.port.text = "Port: " + update[1]
else: else:
# Update the UI
self.ids.sensor1.text = update[0] self.ids.sensor1.text = update[0]
self.ids.sensor2.text = update[1] self.ids.sensor2.text = update[1]
self.ids.sensor3.text = update[2] self.ids.sensor3.text = update[2]
@@ -181,11 +250,12 @@ class MainScreen(Screen):
self.ids.sensor3.text = "" self.ids.sensor3.text = ""
self.ids.sensor4.text = "" self.ids.sensor4.text = ""
self.ids.status.text = "Status will appear here" self.ids.status.text = "Status will appear here"
self.ids.port.text = "Port: Not connected"
# Switch the mode for the micro-controller # Switch the mode for the micro-controller
def switch_mode(self, new_mode: str): def switch_mode(self):
# Store if we have been connected to the micro-controller before mode was switched # Store if we have been connected to the micro-controller before mode was switched
was_connected = self._reader.is_alive was_connected = self._has_connected
# Disconnect from the micro-controller # Disconnect from the micro-controller
self.end() self.end()
@@ -193,14 +263,15 @@ class MainScreen(Screen):
# Try to set the new mode # Try to set the new mode
try: try:
if new_mode == "Normal Mode": if self._fast_mode:
self._com.send("NM") self._com.send("NM")
else: else:
self._com.send("FM") self._com.send("FM")
except: except:
SingleRowPopup().open("Failed to switch modes") self.mode_switch_error_dialog.open()
return return
self.ids.status.text = "Mode set"
# If we have been connected, reconnect # If we have been connected, reconnect
if was_connected: if was_connected:
self.start() self.start()

View File

@@ -1,111 +0,0 @@
<QuitPopup>:
title: "BiogasControllerApp"
font_size: 50
size_hint: 0.5, 0.4
auto_dismiss: False
GridLayout:
cols: 1
Label:
text: "Are you sure you want to leave?"
font_size: 20
GridLayout:
cols:2
Button:
text: "Yes"
font_size: 15
on_release:
root.quit()
app.stop()
Button:
text: "No"
font_size: 15
on_press:
root.dismiss()
<SingleRowPopup>:
title: "INFORMATION"
size_hint: 0.7, 0.5
auto_dismiss: True
GridLayout:
cols: 1
Label:
id: msg
text: "Message"
text_size: self.width, None
halign: 'center'
GridLayout:
cols: 1
Button:
text: "Ok"
on_release:
root.dismiss()
<TwoActionPopup>:
title: "WARNING!"
font_size: 50
size_hint: 0.5, 0.4
auto_dismiss: False
GridLayout:
cols:1
Label:
id: msg
text: "Message"
font_size: 20
halign: 'center'
GridLayout:
cols:2
Button:
id: btn1
text: "Details"
on_release:
root.action_one()
root.dismiss()
Button:
id: btn2
text:"Ok"
on_release:
root.action_two()
root.dismiss()
<DualRowPopup>:
title: "Details"
font_size: 50
size_hint: 0.7, 0.6
auto_dismiss: False
GridLayout:
cols:1
Label:
id: msg_title
text: "Message title"
font_size: 20
Label:
id: msg_body
text: "Message body"
font_size: 14
Button:
text:"Ok"
on_release:
root.dismiss()
<LargeTrippleRowPopUp>:
title: "DETAILS"
font_size: 50
size_hint: 1, 0.7
auto_dismiss: False
GridLayout:
cols: 1
Label:
id: msg_title
text: "title"
font_size: 20
Label:
id: msg_body
text: "Message"
font_size: 13
Label:
text: msg_extra
font_size: 13
Button:
text:"Ok"
on_release:
root.dismiss()

View File

@@ -1,63 +0,0 @@
from typing import Callable
from kivy.uix.popup import Popup
from kivy.lang import Builder
from lib.com import ComSuperClass
# Just an empty function
def empty_func():
pass
# ╭────────────────────────────────────────────────╮
# │ Popups │
# ╰────────────────────────────────────────────────╯
# Below, you can find various popups with various designs that can be used in the app
class QuitPopup(Popup):
def __init__(self, com: ComSuperClass, **kw):
self._com = com;
super().__init__(**kw)
def quit(self):
self._com.close()
class SingleRowPopup(Popup):
def open(self, message, *_args, **kwargs):
self.ids.msg.text = message
return super().open(*_args, **kwargs)
class DualRowPopup(Popup):
def open(self, title: str, message: str, *_args, **kwargs):
self.ids.msg_title.text = title
self.ids.msg_body.text = message
return super().open(*_args, **kwargs)
class LargeTrippleRowPopup(Popup):
def open(self, title: str, message: str, details: str, *_args, **kwargs):
self.ids.msg_title.text = title
self.ids.msg_body.text = message
self.ids.msg_extra.text = details
return super().open(*_args, **kwargs)
class TwoActionPopup(Popup):
def open(self,
message: str,
button_one: str,
action_one: Callable[[], None],
button_two: str = 'Ok',
action_two: Callable[[], None] = empty_func,
*_args,
**kwargs
):
self.ids.msg.text = message
self.ids.btn1.text = button_one
self.ids.btn2.text = button_two
self.action_one = action_one
self.action_two = action_two
return super().open(*_args, **kwargs)
# Load the design file for this screen (.kv files)
# The path has to be relative to root of the app, i.e. where the biogascontrollerapp.py
# file is located
Builder.load_file('./gui/popups/popups.kv')

View File

@@ -1,131 +1,123 @@
<ProgramScreen>: <ProgramScreen>:
name: "program" name: "program"
on_enter: self.config_loader = root.load_config() on_enter: self.config_loader = root.load_config()
canvas.before:
Color:
rgba: (10,10,10,0.1)
Rectangle:
size: self.size
pos: self.pos
FloatLayout: FloatLayout:
Label: MDGridLayout:
cols: 1
pos_hint: {'x': 0, 'y': 0.4}
MDLabel:
text: "Configuration" text: "Configuration"
font_size: 40 font_size: 40
color: (0, 113, 0, 1) halign: 'center'
valign: 'center'
pos_hint: {'center_x': 0, 'center_y': 0}
bold: True bold: True
pos_hint: {"y":0.4}
GridLayout: MDGridLayout:
size_hint: 0.8, 0.5 cols: 1
pos_hint: {"x":0.1, "y":0.2} pos_hint: {'x': 0, 'y': 0.33}
MDLabel:
text: "Change the configuration of the microcontroller"
font_size: 18
halign: 'center'
valign: 'center'
pos_hint: {'center_x': 0, 'center_y': 0}
italic: True
MDGridLayout:
cols: 1
pos_hint: {'x': 0, 'y': 0.25}
MDLabel:
id: status
text: "Loading..."
font_size: 17
halign: 'center'
bold: True
MDGridLayout:
size_hint: 0.9, 0.5
spacing: 10
pos_hint: {"x":0.05, "y":0.2}
cols: 4 cols: 4
Label: MDTextField:
text: "Sensor 1, a:"
TextInput:
id: s1_a id: s1_a
multiline: False hint_text: 'Sensor 1 a'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 1, b:"
TextInput:
id: s1_b id: s1_b
multiline: False hint_text: 'Sensor 1 b'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 1, c:"
TextInput:
id: s1_c id: s1_c
multiline: False hint_text: 'Sensor 1 c'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 1, Temp:"
TextInput:
id: s1_t id: s1_t
multiline: False hint_text: 'Sensor 1 Temperature'
input_filter: "float" on_text: root.validate_float(self)
Label:
text: "Sensor 2, a:" MDTextField:
TextInput:
id: s2_a id: s2_a
multiline: False hint_text: 'Sensor 2 a'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 2, b:"
TextInput:
id: s2_b id: s2_b
multiline: False hint_text: 'Sensor 2 b'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 2, c:"
TextInput:
id: s2_c id: s2_c
multiline: False hint_text: 'Sensor 2 c'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 2, Temp:"
TextInput:
id: s2_t id: s2_t
multiline: False hint_text: 'Sensor 2 Temperature'
input_filter: "float" on_text: root.validate_float(self)
Label:
text: "Sensor 3, a:" MDTextField:
TextInput:
id: s3_a id: s3_a
multiline: False hint_text: 'Sensor 3 a'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 3, b:"
TextInput:
id: s3_b id: s3_b
multiline: False hint_text: 'Sensor 3 b'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 3, c:"
TextInput:
id: s3_c id: s3_c
multiline: False hint_text: 'Sensor 3 c'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 3, Temp:"
TextInput:
id: s3_t id: s3_t
multiline: False hint_text: 'Sensor 3 Temperature'
input_filter: "float" on_text: root.validate_float(self)
Label:
text: "Sensor 4, a:" MDTextField:
TextInput:
id: s4_a id: s4_a
multiline: False hint_text: 'Sensor 4 a'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 4, b:"
TextInput:
id: s4_b id: s4_b
multiline: False hint_text: 'Sensor 4 b'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 4, c:"
TextInput:
id: s4_c id: s4_c
multiline: False hint_text: 'Sensor 4 c'
input_filter: "float" on_text: root.validate_float(self)
Label: MDTextField:
text: "Sensor 4, Temp:"
TextInput:
id: s4_t id: s4_t
multiline: False hint_text: 'Sensor 4 Temperature'
input_filter: "float" on_text: root.validate_float(self)
Button:
MDFillRoundFlatButton:
size_hint: 0.1, 0.07
text: "Back" text: "Back"
size_hint: 0.1, 0.1
pos_hint: {"x":0.1, "y":0.1} pos_hint: {"x":0.1, "y":0.1}
background_color: (255, 0, 0, 0.6) background_color: (255, 0, 0, 0.6)
on_release: on_release:
app.root.current = "main" app.root.current = "main"
root.manager.transition.direction = "up" root.manager.transition.direction = "up"
Button: MDFillRoundFlatButton:
size_hint: 0.15, 0.09
text: "Save" text: "Save"
size_hint: 0.2, 0.1
pos_hint: {"x":0.6, "y":0.1} pos_hint: {"x":0.6, "y":0.1}
background_color: (255, 0, 0, 0.6)
on_release: on_release:
root.save() root.save()

View File

@@ -1,10 +1,11 @@
from typing import List from typing import List
from kivy.uix.screenmanager import Screen from kivymd.uix.screen import MDScreen
from kivy.lang import Builder from kivy.lang import Builder
from lib.decoder import Decoder from util.decoder import Decoder
from lib.instructions import Instructions from util.instructions import Instructions
from gui.popups.popups import SingleRowPopup, TwoActionPopup, empty_func from util.instructions import ControllerConnection
from lib.com import ComSuperClass from kivymd.uix.button import MDFlatButton
from kivymd.uix.dialog import MDDialog
from kivy.clock import Clock from kivy.clock import Clock
@@ -13,18 +14,67 @@ from kivy.clock import Clock
name_map = ["a", "b", "c", "t"] name_map = ["a", "b", "c", "t"]
class ProgramScreen(Screen): class ProgramScreen(MDScreen):
def __init__(self, com: ComSuperClass, **kw): def __init__(self, com: ControllerConnection, **kw):
self._com = com self._com = com
self._instructions = Instructions(com) self._instructions = Instructions(com)
self._decoder = Decoder() self._decoder = Decoder()
# Configure Dialog
self.connection_error_dialog = MDDialog(
title="Connection",
text="Failed to connect. Do you wish to retry?",
buttons=[
MDFlatButton(
text="Cancel",
on_release=lambda _: self.connection_error_dialog.dismiss(),
),
MDFlatButton(text="Retry", on_release=lambda _: self.load_config()),
],
)
self.missing_fields_error_dialog = MDDialog(
title="Save",
text="Some fields are missing entries. Please fill them out and try again",
buttons=[
MDFlatButton(
text="Ok",
on_release=lambda _: self.missing_fields_error_dialog.dismiss(),
),
],
)
self.save_error_dialog = MDDialog(
title="Save",
text="Failed to save data. Please try again",
buttons=[
MDFlatButton(
text="Ok",
on_release=lambda _: self.save_error_dialog.dismiss(),
),
],
)
self.save_success_dialog = MDDialog(
title="Save",
text="Data saved successfully!",
buttons=[
MDFlatButton(
text="Ok",
on_release=lambda _: self.save_success_dialog.dismiss(),
),
],
)
super().__init__(**kw) super().__init__(**kw)
# Load the config (async to not freeze the UI)
def load_config(self): def load_config(self):
Clock.schedule_once(self._load) Clock.schedule_once(lambda _: self._load())
# Load the current configuration from the micro-controller # Load the current configuration from the micro-controller
def _load(self, dt: float): def _load(self):
self.ids.status.text = "Loading..."
# Hook to the microcontroller's data stream (i.e. sync up with it) # Hook to the microcontroller's data stream (i.e. sync up with it)
if self._instructions.hook("RD", ["\n", "R", "D", "\n"]): if self._instructions.hook("RD", ["\n", "R", "D", "\n"]):
config: List[List[str]] = [] config: List[List[str]] = []
@@ -37,13 +87,7 @@ class ProgramScreen(Screen):
received = self._com.receive(28) received = self._com.receive(28)
except: except:
# Open error popup # Open error popup
TwoActionPopup().open( self.connection_error_dialog.open()
"Failed to connect to micro-controller, retry?",
"Cancel",
empty_func,
"Retry",
lambda: self._load(0),
)
return return
# Create a list of strings to store the config for the sensor # Create a list of strings to store the config for the sensor
@@ -58,14 +102,11 @@ class ProgramScreen(Screen):
# Add it to the config # Add it to the config
config.append(config_sensor_i) config.append(config_sensor_i)
self.ids.status.text = ""
self._set_ui(config)
else: else:
TwoActionPopup().open( self.connection_error_dialog.open()
"Failed to connect to micro-controller, retry?",
"Cancel",
empty_func,
"Retry",
lambda: self._load(0),
)
# Set the elements of the UI to the values of the config # Set the elements of the UI to the values of the config
def _set_ui(self, config: List[List[str]]): def _set_ui(self, config: List[List[str]]):
@@ -92,17 +133,42 @@ class ProgramScreen(Screen):
return data return data
# Transmit the changed data to the micro-controller to reconfigure it
def save(self): def save(self):
Clock.schedule_once(lambda _: self._save())
# Transmit the changed data to the micro-controller to reconfigure it
def _save(self):
self.ids.status.text = "Saving..."
data = self._read_ui() data = self._read_ui()
if data == None: if data == None:
SingleRowPopup().open("Some fields are missing values!") self.missing_fields_error_dialog()
else: else:
try: try:
self._instructions.change_config(data) self._instructions.change_config(data)
except: except:
SingleRowPopup().open("Could not save data!") self.save_error_dialog.open()
SingleRowPopup().open("Data saved successfully") return
self.save_success_dialog.open()
self.ids.status.text = "Saved!"
Clock.schedule_once(self.reset_update, 5)
def reset_update(self, _):
self.ids.status.text = ""
def validate_float(self, instance):
text = instance.text
# Allow only digits and one dot
if text.count(".") > 1 or any(c not in "0123456789." for c in text):
# Remove invalid characters
clean_text = "".join(c for c in text if c in "0123456789.")
# Remove extra dots
if clean_text.count(".") > 1:
first_dot = clean_text.find(".")
clean_text = clean_text[: first_dot + 1] + clean_text[
first_dot + 1 :
].replace(".", "")
instance.text = clean_text
# Load the design file for this screen (.kv files) # Load the design file for this screen (.kv files)

View File

@@ -1,127 +0,0 @@
from abc import ABC, abstractmethod
from typing import Optional
import serial
import struct
import serial.tools.list_ports
class ComSuperClass(ABC):
def __init__(self, baudrate: Optional[int] = 19200, filters: Optional[list[str]] = None) -> None:
self._serial: Optional[serial.Serial] = None
self._filters = filters if filters != None else [ 'USB-Serial Controller', 'Prolific USB-Serial Controller' ]
self._port_override = ''
self._baudrate = baudrate if baudrate != None else 19200
self._err = None
def set_port_override(self, override: str) -> None:
"""Set the port override, to disable port search"""
self._port_override = override
def get_error(self) -> serial.SerialException | None:
return self._err
@abstractmethod
def get_comport(self) -> str:
pass
@abstractmethod
def connect(self) -> bool:
pass
@abstractmethod
def close(self) -> None:
pass
@abstractmethod
def receive(self, byte_count: int) -> bytes:
pass
@abstractmethod
def send(self, msg: str) -> None:
pass
@abstractmethod
def send_float(self, msg: float) -> None:
pass
class Com(ComSuperClass):
def _connection_check(self) -> bool:
if self._serial == None:
return self._open()
if self._serial != None:
if not self._serial.is_open:
self._serial.open()
return True
else:
return False
def get_comport(self) -> str:
"""Find the comport the microcontroller has attached to"""
if self._port_override != '':
return self._port_override
# Catch all errors and simply return an empty string if search unsuccessful
try:
# Get an array of all used comports
ports = [comport.device for comport in serial.tools.list_ports.comports()]
# Filter for specific controller
for comport in ports:
for filter in self._filters:
if filter in comport:
return comport
except Exception as e:
self._err = e
return ""
def _open(self) -> bool:
comport = self.get_comport()
# Comport search returns empty string if search unsuccessful
if comport == '':
try:
self._serial = serial.Serial(comport, self._baudrate, timeout=5)
except serial.SerialException as e:
self._err = e
return False
return True
else:
return False
def connect(self) -> bool:
"""Try to find a comport and connect to the microcontroller. Returns the success as a boolean"""
return self._connection_check()
def close(self) -> None:
"""Close the serial connection, if possible"""
if self._serial != None:
try:
self._serial.close()
except:
pass
def receive(self, byte_count: int) -> bytes:
"""Recieve bytes from microcontroller over serial. Returns bytes. Might want to decode using functions from lib.tools"""
self._connection_check()
if self._serial != None:
return self._serial.read(byte_count)
else:
raise Exception('ERR_CONNECTING')
def send(self, msg: str) -> None:
"""Send a string over serial connection. Will open a connection if none is available"""
self._connection_check()
if self._serial != None:
self._serial.write(msg.encode())
else:
raise Exception('ERR_CONNECTING')
def send_float(self, msg: float) -> None:
"""Send a float number over serial connection"""
self._connection_check()
if self._serial != None:
self._serial.write(bytearray(struct.pack('>f', msg))[0:3])
else:
raise Exception('ERR_CONNECTING')

View File

@@ -1,18 +0,0 @@
import struct
class Decoder:
def decode_ascii(self, value: bytes) -> str:
try:
return value.decode()
except:
return 'Error'
def decode_float(self, value: bytes) -> float:
return struct.unpack('>f', bytes.fromhex(str(value, 'ascii') + '00'))[0]
def decode_float_long(self, value: bytes) -> float:
return struct.unpack('>f', bytes.fromhex(str(value, 'ascii') + '0000'))[0]
def decode_int(self, value: bytes) -> int:
# return int.from_bytes(value, 'big')
return int(value, base=16)

View File

@@ -1,148 +0,0 @@
"""
Library to be used in standalone mode (without microcontroller, for testing functionality)
It simulates the behviour of an actual microcontroller being connected
"""
from typing import Optional
import queue
import random
import time
import struct
from lib.com import ComSuperClass
# This file contains a Com class that can be used to test the functionality
# even without a microcontroller. It is not documented in a particularly
# beginner-friendly way, nor is the code written with beginner-friendliness
# in mind. It is the most complicated piece of code of the entire application
# All double __ prefixed properties and methods are not available in the actual one
instruction_lut: dict[str, list[str]] = {
"PR": ["\n", "P", "R", "\n"],
"PT": ["\n", "P", "T", "\n"],
"RD": ["\n", "R", "D", "\n"],
"NM": ["\n", "N", "M", "\n"],
"FM": ["\n", "F", "M", "\n"],
}
class SimulationError(Exception):
pass
class Com(ComSuperClass):
def __init__(
self, baudrate: int = 19200, filters: Optional[list[str]] = None
) -> None:
# Calling the constructor of the super class to assign defaults
print("\n\nWARNING: Using testing library for communication!\n\n")
super().__init__(baudrate, filters)
# Initialize queue with values to be sent on call of recieve
self.__simulated_data: queue.Queue[bytes] = queue.Queue()
self.__simulated_data_remaining = 0
# Initially, we are in normal mode (which leads to slower data intervals)
self.__mode = "NM"
def set_port_override(self, override: str) -> None:
"""Set the port override, to disable port search"""
self._port_override = override
def get_comport(self) -> str:
return "test" if self._port_override != "" else self._port_override
def connect(self) -> bool:
# Randomly return false in 1 in 20 ish cases
if random.randint(0, 20) == 1:
print("Simulating error to connect")
return False
return True
def close(self) -> None:
pass
def receive(self, byte_count: int) -> bytes:
data = []
# If queue is too short, refill it
if self.__simulated_data_remaining < byte_count:
self.__fill_queue()
for _ in range(byte_count):
if self.__mode == "NM":
time.sleep(0.001)
try:
data.append(self.__simulated_data.get_nowait())
self.__simulated_data_remaining -= 1
except Exception as e:
print("ERROR: Simulation could not continue")
raise SimulationError(
"Simulation encountered an error with the simulation queue. The error encountered: \n"
+ str(e)
)
return b''.join(data)
def send(self, msg: str) -> None:
# Using LUT to reference
readback = instruction_lut.get(msg)
if readback != None:
for i in range(len(readback)):
self.__simulated_data.put(bytes(readback[i], "ascii"))
if msg == "RD":
# Handle ReadData readback
# self.__simulated_data.put(ord(""))
pass
def send_float(self, msg: float) -> None:
# Encode float as 8 bytes (64 bit)
ba = struct.pack("d", msg)
for byte in ba:
self.__simulated_data.put(byte.to_bytes())
def __fill_queue_alternative(self):
for _ in range(4):
for _ in range(4):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes(" ", "ascii"))
for _ in range(6):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes(" ", "ascii"))
for _ in range(3):
for _ in range(4):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes(" ", "ascii"))
for _ in range(4):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes("\n", "ascii"))
self.__simulated_data_remaining = 68
def __fill_queue(self):
for _ in range(4):
self.__add_integer_as_hex(self.__generate_random_int(200))
self.__simulated_data.put(bytes(" ", "ascii"))
self.__add_float_as_hex(self.__generate_random_float(50))
self.__simulated_data.put(bytes(" ", "ascii"))
self.__simulated_data_remaining += 2
for _ in range(3):
self.__add_integer_as_hex(self.__generate_random_int(65535))
self.__simulated_data.put(bytes(" ", "ascii"))
self.__add_integer_as_hex(self.__generate_random_int(65535))
self.__simulated_data.put(bytes("\n", "ascii"))
self.__simulated_data_remaining += 4
print("Length:", self.__simulated_data_remaining)
def __generate_random_int(self, max: int) -> int:
return random.randint(0, max)
def __generate_random_float(self, max: int) -> float:
return random.random() * max
def __add_character_as_hex(self, data: str):
pass
def __add_integer_as_hex(self, data: int):
pass
def __add_float_as_hex(self, data: float):
pass

44
package.sh Executable file
View File

@@ -0,0 +1,44 @@
#!/bin/sh
# packaging script
rm -rf ./build/
rm -rf ./dist/
# build windows package
wine pyinstaller BiogasControllerApp.spec
if [ $? -ne 0 ]; then
echo -e "\nBuild unsuccessful, aborting..."
exit 1
fi
# Build successful
cp -r ./gui ./dist
cp -r ./util ./dist
cp ./biogascontrollerapp.py ./dist/
cp ./BiogasControllerAppLogo.png ./dist/
cp ./changelog ./dist/
cp ./config.ini ./dist/
cp ./README.md ./dist/
cp ./requirements.txt ./dist/
cp ./SECURITY.md ./dist/
# Remove build directories
rm -rf ./build/
rm -rf ./dist/biogascontrollerapp/
# Create Windows archive (zip)
zip -9r BiogasControllerApp-Windows.zip ./dist
# Create Linux archive (tar)
rm ./dist/BiogasControllerApp.exe
cp ./install-linux.sh ./dist/
cp ./launch.sh ./dist/
ouch compress -y ./dist/ biogascontrollerapp-linux.tar.gz
rm -rf ./dist
echo "Done!"

View File

@@ -6,53 +6,85 @@ n = int(input("Sensor number to be printed: "))
file = "" file = ""
def generate_plot(): def generate_plot():
reader = csv.reader(file, delimiter=',') # Read data using the CSV library
reader = csv.reader(file, delimiter=",")
# Create a list from the data
data = list(reader) data = list(reader)
data.sort(key=lambda imp: float(imp[2]))
lenght = len(data) # Sort the list using a lambda sort descriptor
# A lambda function is an anonymous function (= an unnamed function),
# which makes it convenient. A sort descriptor is a function that
# (usually, but not here) returns a value indicating which of two values
# come before or after in the ordering.
# Here, instead we simply return a floating point value for each data point
data.sort(key=lambda data_point: float(data_point[2]))
# Store the x and y coordinates in two arrays
x = [] x = []
y = [] y = []
for _ in range(lenght): for _ in range(len(data)):
extract = data.pop(0) # Extract the data point
sensor = int(extract.pop(0)) data_point = data.pop(0)
sensor = int(data_point.pop(0))
if sensor == n: if sensor == n:
ye = extract.pop(0) y.append(float(data_point.pop(0)))
xe = extract.pop(0) x.append(float(data_point.pop(0)))
y.append(float(ye))
x.append(float(xe))
# Use Numpy's polyfit function to fit a 2nd degree polynomial to the points using quadratic regression
# This function returns an array with the coefficients
fit = np.polyfit(x, y, 2) fit = np.polyfit(x, y, 2)
# The formula to output to the plot
formula = f"F(U) = {round(float(fit[0]), 4)}U^2+{round(float(fit[1]), 4)}U+{round(float(fit[2]), 4)}" formula = f"F(U) = {round(float(fit[0]), 4)}U^2+{round(float(fit[1]), 4)}U+{round(float(fit[2]), 4)}"
fit_fn = np.poly1d(fit) # Create a fit function from the previously determined coefficients
fit_fn = np.poly1d(fit) # Returns a function that takes a list of x-coordinate as argument
# Plot the line on the graph
plt.plot(x, fit_fn(x), color="BLUE", label="T(U)") plt.plot(x, fit_fn(x), color="BLUE", label="T(U)")
# Scatter Plot the data points that we have
plt.scatter(x, y, color="MAGENTA", marker="o", label="Data") plt.scatter(x, y, color="MAGENTA", marker="o", label="Data")
# Label the graph
plt.ylabel("Temperature") plt.ylabel("Temperature")
plt.xlabel("Voltage") plt.xlabel("Voltage")
title = 'Sensor MCP9701A #{}'.format(n) plt.title("Sensor MCP9701A #{}".format(n))
plt.title(title)
# Scale the axis appropriately
plt.axis((0.6, 2.0, 15.0, 70.0)) plt.axis((0.6, 2.0, 15.0, 70.0))
# Print a legend and set the graph to be annotated
plt.legend(loc="lower right") plt.legend(loc="lower right")
plt.annotate(formula, xy=(0.85, 60)) plt.annotate(formula, xy=(0.85, 60))
# Enable the background grid
plt.grid(True) plt.grid(True)
# Finally, show the graph
plt.show() plt.show()
# Get user input whether to save the plot or not
saveit = input("Do you wish to save the plot? (y/N) ").lower() saveit = input("Do you wish to save the plot? (y/N) ").lower()
if saveit == "y": if saveit == "y":
plt.savefig("Sensor"+str(n)+".png") # Save the plot as Sensor[Number] (e.g. Sensor9) as png, pdf and svg
plt.savefig("Sensor"+str(n)+".pdf", format="pdf") plt.savefig("Sensor" + str(n) + ".png")
plt.savefig("Sensor"+str(n)+".svg", format="svg") plt.savefig("Sensor" + str(n) + ".pdf", format="pdf")
plt.savefig("Sensor" + str(n) + ".svg", format="svg")
print("==> Images saved") print("==> Images saved")
else: else:
print("==> Images discarded") print("==> Images discarded")
# Since we have defined a function above as a function, this here is executed first
filename = input("Please enter a file path to the csv file to be plotted: ") filename = input("Please enter a file path to the csv file to be plotted: ")
# Try to open the file
try: try:
file = open(filename, "r") file = open(filename, "r")
generate_plot() generate_plot()

View File

@@ -4,29 +4,36 @@ import matplotlib.pyplot as plt
import csv import csv
import os import os
# Get user input for various data
path = input("Path to csv-file to be plotted: ") path = input("Path to csv-file to be plotted: ")
print("For the below, it is recommended to enter data in this format: yyyy-mm-dd-hh-mm")
date = input("Date & time at which the measurement was taken (approx.): ") date = input("Date & time at which the measurement was taken (approx.): ")
group = input("Group-name: ") group = input("Group-name: ")
saveit = input("Should the graph be saved? (y/n) ").lower() saveit = input("Should the graph be saved? (y/n) ").lower()
imp = open(path, "r") imp = open(path, "r")
reader = csv.reader(imp, delimiter=',') reader = csv.reader(imp, delimiter=",")
rohdaten = list(reader) data = list(reader)
lenght = len(rohdaten)
x = [] x = []
y = [] y = []
for i in range(lenght): for i in range(len(data)):
extract = rohdaten.pop(0) # Extract the data
extract = data.pop(0)
x.append(float(extract.pop(0))) x.append(float(extract.pop(0)))
y.append(float(extract.pop(0))) y.append(float(extract.pop(0)))
# Set up plot
plt.plot(x, y, color="MAGENTA") plt.plot(x, y, color="MAGENTA")
plt.xlabel("Time") plt.xlabel("Time")
plt.ylabel("Voltage") plt.ylabel("Voltage")
title = f"GC - Biogasanlage {date}"
plt.title(title) plt.title(f"GC - Biogasanlage {date}")
plt.grid(True) plt.grid(True)
if saveit == "y":
# Check if user wants to save the image
if saveit == "n":
print("didn't save images")
else:
pos = 0 pos = 0
for letter in path[::-1]: for letter in path[::-1]:
if letter == "/": if letter == "/":
@@ -40,11 +47,7 @@ if saveit == "y":
os.mkdir(save_path) os.mkdir(save_path)
except FileExistsError: except FileExistsError:
pass pass
plt.savefig(save_path) plt.savefig(f"{save_path}/GC-{date}-{group}.png")
os.rename(f"{save_path}/.png", f"{save_path}/GC-{date}-{group}.png")
print(f"saved images to {save_path}") print(f"Saved images to {save_path}")
else:
print("didn't save images")
plt.show() plt.show()

View File

@@ -1,2 +1,3 @@
kivy[base] kivy[base]==2.3.1
pyserial kivymd==1.1.1
pyserial==3.5

136
util/com.py Normal file
View File

@@ -0,0 +1,136 @@
from typing import override
import serial
import struct
import serial.tools.list_ports
from util.interface import ControllerConnection
# ┌ ┐
# │ Main Com Class Implementation │
# └ ┘
# Below you can find what you were most likely looking for. This is the implementation of the communication with the microcontroller.
# You may also be interested in the decoder.py and instructions.py file, as the decoding and the hooking / syncing process are
# implemented there. It is recommended that you do NOT read the test/com.py file, as that one is only there for simulation purposes
# and is much more complicated than this here, if you are not well versed with Python or are struggling with the basics
# All variables starting in self are bound to the object and can be changed by any consumer of this library. The Com class
# inherits from the ControllerConnection class (found in interface.py), which implements some of the methods (functions)
# this class exposes, namely the constructor, set_port_override and get_error. They are not further relevant for the code below
# though, so you can safely ignore it.
class Com(ControllerConnection):
def _connection_check(self) -> bool:
if self._serial == None:
return self._open()
if self._serial != None:
if not self._serial.is_open:
self._serial.open()
return True
else:
return False
@override
def get_comport(self) -> str:
"""Find the comport the microcontroller has attached to"""
if self._port_override != "":
return self._port_override
# Catch all errors and simply return an empty string if search unsuccessful
try:
# Get an array of all used comports
ports = [comport.device for comport in serial.tools.list_ports.comports()]
# Filter for specific controller
for comport in ports:
for filter in self._filters:
if filter in comport:
return comport
except Exception as e:
self._err = e
return ""
def _open(self) -> bool:
"""Open the connection. Internal function, not to be called directly, use connect instead
Returns:
Boolean indicates if connection was successful or not
"""
# Get the com port the controller has connected to
comport = self.get_comport()
# Comport search returns empty string if search unsuccessful
if comport != "":
# Try to generate a new Serial object with the configuration of this class
# self._baudrate contains the baud rate and defaults to 19200
try:
self._serial = serial.Serial(comport, self._baudrate, timeout=5)
except serial.SerialException as e:
# If an error occurs, catch it, handle it and store the error
# for the UI and return False to indicate failed connection
self._err = e
return False
# Connection succeeded, return True
return True
else:
# Haven't found a comport
return False
@override
def connect(self) -> bool:
"""Try to find a comport and connect to the microcontroller. Returns the success as a boolean"""
return self._connection_check()
@override
def close(self) -> None:
"""Close the serial connection, if possible"""
if self._serial != None:
try:
self._serial.close()
except:
pass
@override
def receive(self, byte_count: int) -> bytes:
"""Receive bytes from microcontroller over serial. Returns bytes. Might want to decode using functions from lib.decoder"""
# Check connection
self._connection_check()
# Ignore this boilerplate (extra code), the body of the if is the only thing important.
# The reason for the boilerplate is that the type checker will notice that self._serial can be
# None, thus showing errors.
if self._serial != None:
return self._serial.read(byte_count)
else:
raise Exception("ERR_CONNECTING")
@override
def send(self, msg: str) -> None:
"""Send a string over serial connection. Will open a connection if none is available"""
# Check connection
self._connection_check()
# Ignore this boilerplate (extra code), the body of the if is the only thing important.
# The reason for the boilerplate is that the type checker will notice that self._serial can be
# None, thus showing errors.
if self._serial != None:
self._serial.write(msg.encode())
else:
raise Exception("ERR_CONNECTING")
@override
def send_float(self, msg: float) -> None:
"""Send a float number over serial connection"""
# Check connection
self._connection_check()
# Ignore this boilerplate (extra code), the body of the if is the only thing important.
# The reason for the boilerplate is that the type checker will notice that self._serial can be
# None, thus showing errors.
if self._serial != None:
self._serial.write(bytearray(struct.pack(">f", msg))[0:3])
else:
raise Exception("ERR_CONNECTING")

147
util/config.py Normal file
View File

@@ -0,0 +1,147 @@
# This library is used to validate the config file
import configparser
from typing import List
# Load the config
config = configparser.ConfigParser()
config.read("./config.ini")
global first_error
first_error = True
global is_verbose
is_verbose = True
# Set the verbosity if needed
def set_verbosity(verbose: bool):
global is_verbose
is_verbose = verbose
print("\n", "-" * 20, "\nValidating configuration...\n")
def str_to_bool(val: str) -> bool | None:
"""Convert a string to boolean, converting "True" and "true" to True, same for False
Args:
val: The value to try to convert
Returns:
Returns either a boolean if conversion was successful, or None if not a boolean
"""
return {"True": True, "true": True, "False": False, "false": False}.get(val, None)
def read_config(
key_0: str,
key_1: str,
default: str,
valid_entries: List[str] = [],
type_to_validate: str = "",
) -> str:
"""Read the configuration, report potential configuration issues and validate each entry
Args:
key_0: The first key (top level)
key_1: The second key (where the actual key-value pair is)
default: The default value to return if the check fails
valid_entries: [Optiona] The entries that are valid ones to check against
type_to_validate: [Optional] Data type to validate
Returns:
[TODO:return]
"""
# Try loading the keys
tmp = {}
try:
tmp = config[key_0]
except KeyError:
print_config_error(key_0, key_1, "", default, "unknown", index=1)
return default
value = ""
try:
value = tmp[key_1]
except KeyError:
print_config_error(key_0, key_1, "", default, "unknown")
return default
if len(value) == 0:
print_config_error(key_0, key_1, value, default, "not_empty")
# Validate input
if type_to_validate != "":
# Need to validate
if type_to_validate == "int":
try:
int(value)
except ValueError:
print_config_error(key_0, key_1, value, default, "int")
return default
if type_to_validate == "float":
try:
float(value)
except ValueError:
print_config_error(key_0, key_1, value, default, "float")
return default
if type_to_validate == "bool":
if str_to_bool(value) == None:
print_config_error(key_0, key_1, value, default, "bool")
return default
if len(valid_entries) > 0:
# Need to validate the names
try:
valid_entries.index(value)
except ValueError:
print_config_error(
key_0, key_1, value, default, "oneof", valid_entries=valid_entries
)
return default
return value
def print_config_error(
key_0: str,
key_1: str,
value: str,
default: str,
expected: str,
valid_entries: List[str] = [],
msg: str = "",
index: int = 1,
):
"""Print configuration errors to the shell
Args:
key_0: The first key (top level)
key_1: The second key (where the actual value is to be found)
expected: The data type expected. If unknown key, set to "unknown" and set index; If should be one of, use "oneof" and set valid_entries list
msg: The message to print
index: The index in the chain (i.e. if key_0 or key_1)
"""
if not is_verbose:
return
print(f" ==> Using default setting ({default}) for {key_0}.{key_1}")
if expected == "unknown":
# The field was unknown
print(f' -> Unknown field "{key_0 if index == 0 else key_1}"')
elif expected == "oneof":
print(
f' -> Invalid name "{value}". Has to be one of', ", ".join(valid_entries)
)
elif expected == "not_empty":
print(" -> Property is unexpectedly None")
elif expected == "bool":
print(f' -> Boolean property expected, but instead found "{value}".')
else:
print(f" -> Expected a config option of type {expected}.")
if msg != "":
print(msg)

24
util/decoder.py Normal file
View File

@@ -0,0 +1,24 @@
import struct
# Decoder to decode various sent values from the microcontroller
class Decoder:
# Decode an ascii character
def decode_ascii(self, value: bytes) -> str:
try:
return value.decode()
except:
return "Error"
# Decode a float (6 bits)
def decode_float(self, value: bytes) -> float:
return struct.unpack(">f", bytes.fromhex(str(value, "ascii") + "00"))[0]
# Decode a float, but with additional offsets
def decode_float_long(self, value: bytes) -> float:
return struct.unpack(">f", bytes.fromhex(str(value, "ascii") + "0000"))[0]
# Decode an int
def decode_int(self, value: bytes) -> int:
# return int.from_bytes(value, 'big')
return int(value, base=16)

View File

@@ -1,21 +1,17 @@
from lib.com import ComSuperClass import util.decoder
import lib.decoder
import time import time
# TODO: Load filters (for comport search) from util.interface import ControllerConnection
decoder = lib.decoder.Decoder()
decoder = util.decoder.Decoder()
# Class that supports sending instructions to the microcontroller, # Class that supports sending instructions to the microcontroller,
# as well as hooking to data stream according to protocol # as well as hooking to data stream according to protocol
class Instructions: class Instructions:
def __init__(self, com: ComSuperClass) -> None: def __init__(self, com: ControllerConnection) -> None:
self._com = com self._com = com
# Set a port override (to use a specific COM port)
def set_port_override(self, override: str) -> None:
self._com.set_port_override(override)
# Helper method to hook to the data stream according to protocol. # Helper method to hook to the data stream according to protocol.
# You can specify the sequence that the program listens to to sync up, # You can specify the sequence that the program listens to to sync up,
# as an array of strings, that should each be of length one and only contain # as an array of strings, that should each be of length one and only contain
@@ -39,11 +35,13 @@ class Instructions:
# Only run for a limited amount of time # Only run for a limited amount of time
while time.time() - start < 5: while time.time() - start < 5:
# If the decoded ascii character is equal to the next expected character, move pointer right by one # Receive and decode a single byte and decode as ASCII
# If not, jump back to start data = decoder.decode_ascii(self._com.receive(1))
if decoder.decode_ascii(self._com.receive(1)) == sequence[pointer]: if data == sequence[pointer]:
# Increment the pointer (move to next element in the List)
pointer += 1 pointer += 1
else: else:
# Jump back to start
pointer = 0 pointer = 0
# If the pointer has reached the end of the sequence, return True, as now the hook was successful # If the pointer has reached the end of the sequence, return True, as now the hook was successful
@@ -53,7 +51,7 @@ class Instructions:
# If we time out, which is the only way in which this code can be reached, return False # If we time out, which is the only way in which this code can be reached, return False
return False return False
# Used to hook to the main data stream, as that hooking mechanism is differen # Used to hook to the main data stream, as that hooking mechanism is different
def hook_main(self) -> bool: def hook_main(self) -> bool:
# Record start time to respond to timeout # Record start time to respond to timeout
start = time.time() start = time.time()
@@ -61,16 +59,26 @@ class Instructions:
# Wait to find a CR character (enter) # Wait to find a CR character (enter)
char = decoder.decode_ascii(self._com.receive(1)) char = decoder.decode_ascii(self._com.receive(1))
while char != "\n": while char != "\n":
# Check for timeout
if time.time() - start > 3: if time.time() - start > 3:
return False return False
# Set the next character by receiving and decoding it as ASCII
char = decoder.decode_ascii(self._com.receive(1)) char = decoder.decode_ascii(self._com.receive(1))
# Store the position in the hooking process # Store the position in the hooking process
state = 0 state = 0
distance = 0 distance = 0
# While we haven't timed out and have not reached the last state execute
# The last state indicates that the sync was successful
while time.time() - start < 5 and state < 3: while time.time() - start < 5 and state < 3:
# Receive the next char and decode it as ASCII
char = decoder.decode_ascii(self._com.receive(1)) char = decoder.decode_ascii(self._com.receive(1))
# The character we look for when syncing is Space (ASCII char 32 (decimal))
# It is sent every 4 bits. If we have received 3 with the correct distance from
# the previous in a row, we are synced
if char == " ": if char == " ":
if distance == 4: if distance == 4:
state += 1 state += 1
@@ -82,6 +90,7 @@ class Instructions:
else: else:
distance += 1 distance += 1
# Read 5 more bits to correctly sync up
self._com.receive(5) self._com.receive(5)
return state == 3 return state == 3

66
util/interface.py Normal file
View File

@@ -0,0 +1,66 @@
from abc import ABC, abstractmethod
from typing import Optional
import serial
# If you don't know what OOP is, you can safely ignore this file
#
# The below class is abstract to have a consistent, targetable interface
# for both the real connection module and the simulation module
#
# For the interested, a quick rundown of what the benefits are of doing it this way is:
# This class provides a way to have two wholly different implementations that have
# the same function interface (i.e. all functions take the same arguments)
#
# Another benefit of having classes is that we can pass a single instance around to
# various components and have one shared instance that all can modify, reducing some
# overhead.
#
# The actual implementation of most functions (called methods in OOP) are implemented
# in the Com class below.
class ControllerConnection(ABC):
def __init__(
self, baudrate: Optional[int] = 19200, filters: Optional[list[str]] = None
) -> None:
self._serial: Optional[serial.Serial] = None
self._filters = (
filters
if filters != None
else ["USB-Serial Controller", "Prolific USB-Serial Controller"]
)
self._port_override = ""
self._baudrate = baudrate if baudrate != None else 19200
self._err = None
def set_port_override(self, override: str) -> None:
"""Set the port override, to disable port search"""
if override != "" and override != "None":
self._port_override = override
def get_error(self) -> serial.SerialException | None:
return self._err
@abstractmethod
def get_comport(self) -> str:
pass
@abstractmethod
def connect(self) -> bool:
pass
@abstractmethod
def close(self) -> None:
pass
@abstractmethod
def receive(self, byte_count: int) -> bytes:
pass
@abstractmethod
def send(self, msg: str) -> None:
pass
@abstractmethod
def send_float(self, msg: float) -> None:
pass

274
util/test/com.py Normal file
View File

@@ -0,0 +1,274 @@
"""
Library to be used in standalone mode (without microcontroller, for testing functionality)
It simulates the behviour of an actual microcontroller being connected
"""
# ────────────────────────────────────────────────────────────────────
# ┌ ┐
# │ Testing Module For Com │
# └ ┘
# This file contains a Com class that can be used to test the functionality
# even without a microcontroller. It is not documented in a particularly
# beginner-friendly way, nor is the code written with beginner-friendliness
# in mind. It is the most complicated piece of code of the entire application
# ────────────────────────────────────────────────────────────────────
# Just be warned, more OOP concepts and less documentation can be found here.
# Code starts here
# ────────────────────────────────────────────────────────────────────
from typing import List, Optional, override
import queue
import random
import time
import struct
from util.interface import ControllerConnection
# All double __ prefixed properties and methods are not available in the actual impl
instruction_lut: dict[str, list[str]] = {
"PR": ["\n", "P", "R", "\n"],
"PT": ["\n", "P", "T", "\n"],
"RD": ["\n", "R", "D", "\n"],
"NM": ["\n", "N", "M", "\n"],
"FM": ["\n", "F", "M", "\n"],
}
reconfig = ["a", "b", "c", "t"]
class SimulationError(Exception):
pass
class SensorConfig:
a: float
b: float
c: float
t: float
def __init__(
self, a: float = 20, b: float = 30, c: float = 10, t: float = 55
) -> None:
self.a = a
self.b = b
self.c = c
self.t = t
class Com(ControllerConnection):
def __init__(
self, fail_sim: int, baudrate: int = 19200, filters: Optional[list[str]] = None
) -> None:
# Calling the constructor of the super class to assign defaults
print("\n\nWARNING: Using testing library for communication!\n\n")
super().__init__(baudrate, filters)
# Initialize queue with values to be sent on call of recieve
self.__simulated_data: queue.Queue[bytes] = queue.Queue()
self.__simulated_data_remaining = 0
self.__reconf_sensor = 0
self.__reconf_step = 0
self.__fail_sim = fail_sim
self.__config: List[SensorConfig] = [
SensorConfig(),
SensorConfig(),
SensorConfig(),
SensorConfig(),
]
# Initially, we are in normal mode (which leads to slower data intervals)
self.__mode = "NM"
@override
def get_comport(self) -> str:
return "Sim" if self._port_override == "" else self._port_override
@override
def connect(self) -> bool:
# Randomly return false in 1 in fail_sim ish cases
if random.randint(0, self.__fail_sim) == 0:
print("Simulating error to connect")
return False
return True
@override
def close(self) -> None:
pass
@override
def receive(self, byte_count: int) -> bytes:
data = []
# If queue is too short, refill it
if self.__simulated_data_remaining < byte_count:
self.__fill_queue()
for _ in range(byte_count):
if self.__mode == "NM":
time.sleep(0.005)
try:
data.append(self.__simulated_data.get_nowait())
self.__simulated_data_remaining -= 1
except Exception as e:
print("ERROR: Simulation could not continue")
raise SimulationError(
"Simulation encountered an error with the simulation queue. The error encountered: \n"
+ str(e)
)
return b"".join(data)
@override
def send(self, msg: str) -> None:
# Using LUT to reference
readback = instruction_lut.get(msg)
if readback != None:
for i in range(len(readback)):
self.__add_ascii_char(readback[i])
if msg == "RD":
self.__set_read_data_data()
elif msg == "PR":
self.__reconf_sensor = 0
self.__reconf_step = 0
self.__add_ascii_char("a")
self.__add_ascii_char("0")
self.__add_ascii_char("\n")
def __set_read_data_data(self) -> None:
# Send data for all four sensors
for i in range(4):
self.__add_float_as_hex(self.__config[i].a)
self.__add_ascii_char(" ")
self.__add_float_as_hex(self.__config[i].b)
self.__add_ascii_char(" ")
self.__add_float_as_hex(self.__config[i].c)
self.__add_ascii_char(" ")
self.__add_float_as_hex(self.__config[i].t)
self.__add_ascii_char("\n")
@override
def send_float(self, msg: float) -> None:
if self.__reconf_step == 0:
self.__config[self.__reconf_sensor].a = msg
elif self.__reconf_step == 1:
self.__config[self.__reconf_sensor].b = msg
elif self.__reconf_step == 2:
self.__config[self.__reconf_sensor].c = msg
elif self.__reconf_step == 3:
self.__config[self.__reconf_sensor].t = msg
if self.__reconf_step == 3:
self.__reconf_step = 0
self.__reconf_sensor += 1
else:
self.__reconf_step += 1
if self.__reconf_sensor == 4:
return
self.__add_ascii_char(reconfig[self.__reconf_step])
self.__add_ascii_char(str(self.__reconf_sensor))
self.__add_ascii_char("\n")
def __fill_queue(self):
# Simulate a full cycle
for _ in range(4):
self.__add_integer_as_hex(self.__generate_random_int(200))
self.__simulated_data.put(bytes(" ", "ascii"))
self.__add_float_as_hex(self.__generate_random_float(50))
self.__simulated_data.put(bytes(" ", "ascii"))
self.__simulated_data_remaining += 2
for _ in range(3):
self.__add_integer_as_hex(self.__generate_random_int(65535))
self.__simulated_data.put(bytes(" ", "ascii"))
self.__simulated_data_remaining += 1
self.__add_integer_as_hex(self.__generate_random_int(65535))
self.__simulated_data.put(bytes("\n", "ascii"))
self.__simulated_data_remaining += 1
def __generate_random_int(self, max: int) -> int:
return random.randint(0, max)
def __generate_random_float(self, max: int) -> float:
return random.random() * max
def __add_ascii_char(self, ascii_string: str):
self.__simulated_data.put(ord(ascii_string).to_bytes(1))
self.__simulated_data_remaining += 1
def __add_two_byte_value(self, c: int):
"""putchhex
Args:
c: The char (as integer)
"""
# First nibble (high)
high_nibble = (c >> 4) & 0x0F
high_char = chr(high_nibble + 48 if high_nibble < 10 else high_nibble + 55)
self.__simulated_data.put(high_char.encode())
# Second nibble (low)
low_nibble = c & 0x0F
low_char = chr(low_nibble + 48 if low_nibble < 10 else low_nibble + 55)
self.__simulated_data.put(low_char.encode())
self.__simulated_data_remaining += 2
def __add_integer_as_hex(self, c: int):
"""Writes the hexadecimal representation of the high and low bytes of integer `c` (16-bit) to the simulated serial port."""
if not (0 <= c <= 0xFFFF):
raise ValueError("Input must be a 16-bit integer (065535)")
# Get high byte (most significant byte)
hi_byte = (c >> 8) & 0xFF
# Get low byte (least significant byte)
lo_byte = c & 0xFF
# Call putchhex for the high byte and low byte
self.__add_two_byte_value(hi_byte)
self.__add_two_byte_value(lo_byte)
def __add_float_as_hex(self, f: float):
"""Converts a float to its byte representation and sends the bytes using putchhex."""
# Pack the float into bytes (IEEE 754 format)
packed = struct.pack(">f", f) # Big-endian format (network byte order)
# Unpack the bytes into 3 bytes: high, mid, low
high, mid, low = packed[0], packed[1], packed[2]
# Send each byte as hex
self.__add_two_byte_value(high)
self.__add_two_byte_value(mid)
self.__add_two_byte_value(low)