# -*- coding: utf-8 -*-
"""
base_instrument.
Copyright <2022> <Christian Jung, www.heli2.de>
Permission is hereby granted, free of charge, to any person obtaining a copy of this software
and associated documentation files (the "Software"), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish, distribute,
sublicense, and/or sell copies of the Software, and to permit persons to whom the Software
is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial
portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
Created on Fri Jan 14 11:56:24 2022
@author: Christian Jung
INFO:
icons von https://github.com/spyder-ide/qtawesome
conda install qtawesome
Done:
read svg-files and connect id's to a mqtt-command
"""
import os
from PyQt5 import QtCore, QtWidgets, uic
import qtawesome as qta
from pylab_ml.misc.iasvg import InteractiveSvgWidget
from pylab_ml.common.mqtt_client import mylogger
from pylab_ml.common.data import str2num
__author__ = "Christian Jung"
__copyright__ = "Copyright 2022, Christian Jung"
__credits__ = ["Christan Jung"]
__email__ = "christian.jung@heli2.de"
__version__ = "0.0.5"
[docs]
def load_ui(window, my_uifile):
name = os.path.splitext(my_uifile)
if len(name) > 1 and name[1] == ".py":
my_uifile = my_uifile.replace(".py", ".ui")
if not os.path.exists(my_uifile):
raise Exception("'%s' doesn't exist" % my_uifile)
return uic.loadUi(my_uifile, window)
[docs]
class Gui(object):
"""
Base Class for Gui Applikations with an Sensor/Instrument
Needs:
- mqtt-message from the parent
Provides:
- basic-Window for Instruments
- receive and show mqtt_status in the statusbar
- subtopic
- function mqttConnectWidgets: if Objects-name or an svg-id start with MQTT than automatically
connect this object to gui2mqtt() and send the new value about mqtt
recognise QT-objects: QCombobox, QPushButton, QDoubleSpinBox, QSpinBox, QLCDNumber, QLabel
- if the Gui starts, it send the mqtt connect message
if the Gui receive a mqtt connect, it send the the mqtt get from the list self.mqtt_initlist + self.mqtt_cmds
"""
_states = {
"disconnect": ["disconnect", "color: rgb(255, 0, 0)"],
"connect": ["connect", "color: rgb(0, 255, 0)"],
"on": ["on", "color: rgb(0, 255, 0)"],
"off": [
"off",
"color: palette(HighlightedText)",
], # { color: palette(dark); }#QPalette::HighlightedText
}
_mqttPrefix = "MQTT_"
_mqttSet = "SET_"
_mqttGet = "GET_"
_tabPrefix = "TAB_"
[docs]
def __init__(self, grandparent=None, name=None, parentwindow=None):
"""
Init the Gui, create a window and load the ui-file with the same name as the class.
Parameters
----------
grandparent : object, optional
The parent object that contains the MQTT client and other shared resources. The default is None.
name : string, optional
The name of the instrument, used for MQTT topics and window title. The default is None.
parentwindow : QtWidgets.QWidget, optional
The parent window for the GUI. The default is None.
"""
self.logger = mylogger() if not hasattr(grandparent, "logger") else grandparent.logger
self.parentwindow = parentwindow if parentwindow is not None else grandparent
self.debug = False
window = QtWidgets.QMainWindow(parentwindow)
self.instName = name
self.instNameExtension = ""
self.grandparent = grandparent
load_ui(window, __file__)
self.gui = window
self.adjustUI()
self.geometry = self.gui.geometry
self.width = self.gui.width
self.height = self.gui.height
self.gui.closeEvent = self.close
self.gui.show()
self._onoff_ = None
self._id = ""
self._mqtt_status = "disconnect"
self.status = "disconnect"
self.topinstname = ""
self._subtopic = ""
self._lastsvgcolour = None
self.mqtt_initlist = []
self.mqtt_cmds = []
self.publish("mqtt_status", "connect") # I'm alive
# self.gui.myframe.setEnabled(True)
[docs]
def adjustUI(self):
""" Adjust the UI, set the title and icons and connect the quit action. """
self.gui.setWindowTitle(f" {os.path.basename(__file__)}-Gui (Version {__version__})")
self._translate = QtCore.QCoreApplication.translate
# set icons:
self.gui.actionQuit.setIcon(
qta.icon(
"ei.remove-sign",
color="white",
scale_factor=1.0,
color_active="orange",
)
)
# set connect:
self.gui.actionQuit.triggered.connect(self.gui.close)
self.gui.runToolBar = self.gui.addToolBar("Run")
self.gui.runMenu = []
[docs]
def retranslateUi(self, Form):
""" Retranslate the UI, set the text of the widgets. """
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "Form"))
self.checkBox.setText(_translate("Form", "CheckBox"))
self.label.setText(_translate("Form", "Label"))
[docs]
def setLabel(self, Label, value, ext=None):
"""
Set the text of a label.
Parameters
----------
Label : object
the label object.
value : string
this text will be displayed.
ext : string, optional
extention, The default is None.
Returns
-------
None.
"""
text = Label.text()
text = f"{text[:text.find(':')]}: {value}"
if ext is not None:
text = f"{text} {ext}"
Label.setText(text)
[docs]
def set_Geometry(self, geometry):
"""
Set the geometry of the GUI, check if the geometry is on the actual screen.
Parameters
----------
geometry : tuple
A tuple of the form (x, y, width, height) representing the geometry of the GUI.
"""
found = False
for i in range(0, QtWidgets.QDesktopWidget().screenCount()):
screen = QtWidgets.QDesktopWidget().screenGeometry(i)
if screen.x() < geometry[0] and screen.y() < geometry[1]:
found = True
if found:
self.gui.setGeometry(geometry[0], geometry[1], geometry[2], geometry[3])
else:
print(f"Warning: coudn't set last geometry for {self.instName}, it is out of the actual screen")
@property
def subtopic(self):
""" Return the subtopic of the instrument, which is the part after the topinstname. """
return self._subtopic
@subtopic.setter
def subtopic(self, value):
if value == []:
self._subtopic = value
return
split = value.split(".")
self._subtopic = [split[0]] if split[0] != self.topinstname else [split[1]]
[docs]
def filtersubtopic(self, value):
""" Filter the subtopic from the value, if the value starts with the topinstname. """
return (
value[value.find(self.topinstname) + len(self.topinstname) + 1:]
if value.split(".")[0] == self.topinstname
else value
)
[docs]
def unitOfMeasurement(self, value, unit=None):
"""
Convert a value with a unit to its numerical representation.
Parameters
----------
value : string
The value with a unit (e.g., "10 k").
unit : string, optional
The unit to use for conversion. If None, the unit is inferred from the value.
Returns
-------
float
The numerical representation of the value.
"""
split = value.split(" ")
result = str2num(split[0])
unit = split[1][0] if len(split) > 1 and unit is None else unit
mul = 1
if unit == "M":
mul = 1000000
elif unit == "k":
mul = 1000
elif unit == "m":
mul = 0.001
elif unit == "u":
mul = 0.000001
elif unit == "n":
mul = 0.000000001
return result * mul
[docs]
def objectName2mqttName(self, name):
"""
Filter from the object name the right MQTT name.
eg. objectname = MQTT_voltage -> Voltage listen on get and set
= MQTT_set_I_limit -> I_limit listen only on set
= MQTT_get_I_limit -> I_limit listen only on get
= TAB_ch -> Name for a tab-name
Parameters
----------
name : string
The object name to be filtered.
Returns
-------
listen : string
The type of MQTT command to listen for ("set", "get", or "").
result : string
The filtered MQTT name extracted from the object name.
"""
pos = name.find(self._mqttPrefix)
listen = ""
result = None
if pos == 0:
pos = len(self._mqttPrefix)
listen = ""
setget = name[pos:].find(self._mqttSet)
if setget == 0:
pos += len(self._mqttSet)
listen = "set"
setget = name[pos:].find(self._mqttGet)
if setget == 0:
pos += len(self._mqttSet)
listen = "get"
result = name[pos:]
else:
pos = name.find(self._tabPrefix)
if pos == 0:
result = self._tabPrefix
return listen, result
# =======================================================
# attributes which connect to an extern call (mqtt-command)
[docs]
def mqttreceive(self, instName, msg, check=False):
"""
Common MQTT receive messages, get raw MQTT data for receiving more information.
Parameters
----------
instName : string
The name of the instrument sending the MQTT message.
msg : dict
The MQTT message received, expected to have keys "cmd", "payload", and "type".
check : bool, optional
If True, only process the message if the command is in self.mqtt_cmds. The default is False.
Returns
-------
result : bool
True if the message was processed successfully, False otherwise.
"""
self.dprint(f' {instName}.mqttreceive from base class: search {msg["cmd"]}')
result = False
if "cmd" in tuple(msg.keys()) and "payload" in tuple(msg.keys()) and (msg["cmd"] in self.mqtt_cmds or not check):
typ = msg["type"].upper()
widget = None
for name in [
f'MQTT_{typ}_{msg["cmd"]}',
f'MQTT_{msg["cmd"]}',
]: # search for corresponding widget
if hasattr(self.myframe, name):
widget = getattr(self.myframe, name)
break
if widget is None and hasattr(self, "svgWidgets"): # search for corresponding svg-id
self.dprint(f' now search for {msg["cmd"]} in {self.svgWidgets.keys()}')
name = msg["cmd"].split(".")
if name[0] in self.svgWidgets:
self.dprint(f" {self.instName}.mqttreceive base: found svgWidgets: {name}")
if len(name) == 1:
svgwidget = self.svgWidgets[name]
else:
element = msg["cmd"][msg["cmd"].find(".") + 1:]
svgwidget = self.svgWidgets[name[0]]
svgwidget.setIdText(element, msg["payload"])
self.dprint(f" {self.instName}.mqttreceive base: set {name[0]}.{element}={msg['payload']}")
return True
if widget is None:
self.dprint(f' Warning {self.instName}.mqttreceive base: {instName}, cmd={msg["cmd"]} type= msg["type"] not implemented yet !')
return False
self.dprint(f' {self.instName}.mqttreceive base: {instName}, found {widget.objectName()}, type={type(widget)} value={msg["payload"]}')
value = str2num(msg["payload"], default=-1)
twidget = type(widget)
result = True
if twidget == dict:
widget["svgelement"].getchildren()[0].text = value
return result
if twidget == QtWidgets.QComboBox:
widget.blockSignals(True)
for index in range(0, widget.count()):
itemvalue = self.unitOfMeasurement(widget.itemText(index))
if type(value) in [int, float] and value >= itemvalue:
break
elif type(value) == str and value.find(widget.itemText(index).split(" ")[0]) == 0:
break
widget.setCurrentIndex(index)
self.dprint(f' {self.instName}.mqttreceive base: {instName}, {msg}, cmd={msg["cmd"]} done')
elif twidget == QtWidgets.QPushButton:
widget.blockSignals(True)
if type(value) == bool:
value = 0 if not value else 1
if value <= len(widget.textchoice):
widget.setText(widget.textchoice[value])
# if value == 1:
# widget.setStyleSheet("color: palette(HighlightedText);background-color: #19c9ff;")
# elif value == 0:
# widget.setStyleSheet("color: palette(HighlightedText);background-color: transparent;")
widget.setChecked(bool(value))
elif twidget in [
QtWidgets.QDoubleSpinBox,
QtWidgets.QSpinBox,
QtWidgets.QLCDNumber,
]:
widget.blockSignals(True)
widget.setProperty("value", value)
elif twidget == QtWidgets.QLabel:
widget.blockSignals(True)
widget.setText(value)
result = True
else:
self.dprint(f' {self.instName}.mqttreceive base: {instName}, {msg}, cmd={msg["cmd"]} widget={type(widget)} not yet implemented -> improve it !')
result = False
widget.blockSignals(False)
else:
self.dprint(f' {instName}.mqttreceive from base: {msg["cmd"]} not found in {self.mqtt_cmds}')
return result
@property
def id(self):
""" Return the id of the instrument, which is used for MQTT topics. """
return self._id
@id.setter
def id(self, value):
self._id = value
self.instNameExtension = self.id
self.mqtt_status
@property
def mqtt_status(self):
""" Return the MQTT status of the instrument, which is either "connect" or "disconnect". """
self.dprint(f" {self.instName}.mqtt_status == {self._mqtt_status}")
msg = f"{self.instName}: {self._mqtt_status}"
self.gui.myInstrument.setTitle(msg)
return self._mqtt_status
@mqtt_status.setter
def mqtt_status(self, value):
self.dprint(f" {self.instName}.mqtt_status := {value}")
self.status = value
if value == "disconnect":
self.publish("mqtt_status", "connect")
self.gui.myframe.setEnabled(False)
elif value == "connect":
self.gui.myframe.setEnabled(True)
if self._mqtt_status == "disconnect":
self.dprint(f" now ask for init values {self.mqtt_initlist}, {self.mqtt_cmds}")
for attribute in self.mqtt_initlist + self.mqtt_cmds:
if attribute.find("!") < 0:
self.dprint(f" {attribute}")
self.grandparent.mqtt.publish_get(self.instName, attribute)
self._mqtt_status = value
msg = f"{self.instName}: {value}"
self.gui.myInstrument.setTitle(msg)
[docs]
def gui_disconnect(self):
""" Disconnect the GUI, set the MQTT status to "disconnect" and disable the frame. """
value = "disconnect"
self.status = value
self._mqtt_status = value
self.publish("mqtt_status", value)
msg = f"{self.instName}: {value}"
self.gui.myInstrument.setTitle(msg)
self.gui.myframe.setEnabled(False)
# end extern calling attributes
# ======================================================================================
#
# connected GUI-function to buttons or menues
#
def _onoff(self):
""" Toggle the on/off state of the instrument, send the corresponding MQTT command and update the menu icon and status. """
if self._mqtt_status == "disconnect":
return
if self._onoff_ is None:
self.publish("off()")
self.status = "no channel defined"
elif self._onoff_ == 0:
self.publish("on()")
self.gui.runMenu[0].setIcon(qta.icon("ei.pause", color="red", scale_factor=1.0))
self._onoff_ = 1
self.status = "on"
else:
self.publish("off()")
self.gui.runMenu[0].setIcon(qta.icon("ei.play", color="green", scale_factor=1.0))
self._onoff_ = 0
self.status = "off"
@property
def status(self):
"""Return the current status message from the status bar."""
return self.gui.statusbar.currentMessage()
@status.setter
def status(self, value):
"""Set the status message and style in the status bar."""
style = None
msg = value
if type(value) == tuple:
msg = value[0]
style = value[1]
if value in self._states:
msg = self._states[value][0]
style = self._states[value][1]
style == "color: palette(HighlightedText);background-color: transparent;" if None else style
self.gui.statusbar.setStyleSheet(style)
self.gui.statusbar.showMessage(f"{msg}")
[docs]
def publish(self, cmd, value="", instName=None):
"""
Publish a command and value over MQTT.
Parameters
----------
cmd : string
The command to be published.
value : string, optional
The value associated with the command. The default is an empty string.
instName : string, optional
The name of the instrument. If None, the instance's name is used. The default is None.
"""
instName = self.instName if instName is None else instName
if hasattr(self.grandparent, "mqtt") and hasattr(self.grandparent.mqtt, "publish_set"):
self.grandparent.mqtt.publish_set(instName, cmd, value)
else:
print(f"Error: {instName}.{cmd} := {value} but no publish_set defined!")
[docs]
def gui2mqtt(self, widget):
"""
Function connect a Gui-widget to mqtt, it sends a mqtt set command with the value from the Gui-widget.
Parameters
----------
widget : QtWidgets.QWidget
The GUI widget that triggered the MQTT command, expected to have an objectName that follows the
Returns
-------
None.
"""
listen, name = self.objectName2mqttName(widget.objectName())
name = widget.objectName() if name is None else name
print(f"{self.instName}.gui2mqtt {name}")
typ = type(widget)
found = False
if typ == QtWidgets.QComboBox:
value = self.unitOfMeasurement(widget.currentText())
found = True
elif typ == QtWidgets.QPushButton:
value = 1 if widget.isChecked() else 0
found = True
elif typ in [QtWidgets.QDoubleSpinBox, QtWidgets.QSpinBox]:
value = widget.value()
found = True
if found:
self.publish(name, value)
else:
print(f"Warning: {self.instName}.gui2mqtt {name} widget={type(widget)} not yet implemented -> improve it !!!!!!")
[docs]
def svg2mqtt(self, name, value):
"""
Publish the name and value over mqtt.
Overwrite this function if you need a more complex form.
Parameters
----------
name : string
The name of the SVG element that was changed.
value : string
The new value of the SVG element.
"""
self.publish(name, value)
[docs]
def isVisible(self):
""" Return whether the GUI is currently visible. """
return self.gui.isVisible()
[docs]
def hide(self):
""" Hide the GUI. """
self.gui.hide()
[docs]
def show(self):
""" Show the GUI. """
self.gui.show()
[docs]
def close(self, event=None):
""" Close the GUI, publish a disconnect message and call the appclosed method of the grandparent if it exists. """
self.publish("mqtt_status", "disconnect")
if hasattr(self.grandparent, "appclosed"):
self.grandparent.appclosed(self.instName)
self.gui.close()
[docs]
def dprint(self, msg):
""" Print a debug message if debugging is enabled. """
if self.debug:
print(msg)
if __name__ == "__main__":
pass
import sys
# from mqtt_client import mqtt_init
# broker = 127.0.0.1'
# message_client = 'ate/DT1604092/matrix'
# mqttclient = mqtt_init() # prepare mqtt
# mqttclient.init(None, message_client=message_client.split('/')[0]) # mqtt client connect to broker
# mqttclient.init()
app = QtWidgets.QApplication(sys.argv)
window = Gui(app, "My Instrument")
app.exec_()