85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
from kivy.app import App
|
|
from kivy.uix.tabbedpanel import TabbedPanel, TabbedPanelItem
|
|
from kivy.uix.boxlayout import BoxLayout
|
|
from kivy.uix.label import Label
|
|
from kivy.uix.button import Button
|
|
from kivy.clock import Clock
|
|
import threading
|
|
|
|
from comm_modbus_device import class_modbus_comm_device
|
|
from menu_mine import mqtt_pack_measure, mqtt_pack_param, mqtt_pack_alarm
|
|
|
|
class DataTab(BoxLayout):
|
|
def __init__(self, device, data_pack, title, **kwargs):
|
|
super().__init__(orientation='vertical', **kwargs)
|
|
self.device = device
|
|
self.data_pack = data_pack
|
|
self.labels = {}
|
|
self.title = title
|
|
self.build_ui()
|
|
|
|
def build_ui(self):
|
|
title_label = Label(text=f"[b]{self.title}[/b]", markup=True, size_hint_y=None, height=40)
|
|
self.add_widget(title_label)
|
|
|
|
for item in self.data_pack:
|
|
name = item.get("name")
|
|
label = Label(text=f"{name}: --", size_hint_y=None, height=30)
|
|
self.labels[name] = label
|
|
self.add_widget(label)
|
|
|
|
refresh_btn = Button(text="Refresh", size_hint_y=None, height=40)
|
|
refresh_btn.bind(on_press=self.refresh_data)
|
|
self.add_widget(refresh_btn)
|
|
|
|
def refresh_data(self, instance):
|
|
def _task():
|
|
for item in self.data_pack:
|
|
try:
|
|
name = item.get("name")
|
|
value = self.device.comm_get_value_from_menu_item(item)
|
|
if value is not None:
|
|
text = f"{name}: {value}"
|
|
else:
|
|
text = f"{name}: Read failed"
|
|
Clock.schedule_once(lambda dt, n=name, t=text: self.update_label(n, t))
|
|
except Exception as e:
|
|
Clock.schedule_once(lambda dt: self.update_label(name, f"{name}: Error"))
|
|
threading.Thread(target=_task, daemon=True).start()
|
|
|
|
def update_label(self, name, text):
|
|
if name in self.labels:
|
|
self.labels[name].text = text
|
|
|
|
class MainPanel(TabbedPanel):
|
|
def __init__(self, device, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.do_default_tab = False
|
|
|
|
self.measure_tab = DataTab(device, mqtt_pack_measure, "Measurements")
|
|
self.add_widget(self.create_tab("Measure", self.measure_tab))
|
|
|
|
self.param_tab = DataTab(device, mqtt_pack_param, "Parameters")
|
|
self.add_widget(self.create_tab("Params", self.param_tab))
|
|
|
|
self.alarm_tab = DataTab(device, mqtt_pack_alarm, "Alarms")
|
|
self.add_widget(self.create_tab("Alarms", self.alarm_tab))
|
|
|
|
def create_tab(self, tab_title, content_widget):
|
|
tab = TabbedPanelItem(text=tab_title)
|
|
tab.add_widget(content_widget)
|
|
return tab
|
|
|
|
class ModbusTabbedApp(App):
|
|
def build(self):
|
|
self.device = class_modbus_comm_device()
|
|
from modbus_tk import modbus_tcp
|
|
self.device.master = modbus_tcp.TcpMaster('192.168.1.119', 502)
|
|
self.device.master.set_timeout(5.0)
|
|
# self.device.set_slave_id(1)
|
|
return MainPanel(self.device)
|
|
|
|
if __name__ == '__main__':
|
|
ModbusTabbedApp().run()
|