#!/usr/bin/env python # -*- coding: UTF-8 -*- # # generated by wxGlade 0.9.5 on Fri Feb 7 20:10:41 2020 # import wx import wx.grid import wx.propgrid import gettext import re import sys import pprint from pubsub import pub import pubsub.utils.notification import easymodbus.modbusClient from registers import regdict from stabiliti import StabilitiController as SC from faults import faultInfo from FaultView import FaultFrame print('pubsub API version', pub.VERSION_API) pubsub.utils.notification.useNotifyByWriteFile(sys.stdout) class WebRelay: def __init__(self, n_relay=10): self.n_relay = n_relay self.relay_status = [False for i in range(self.n_relay)] class Controller: def __init__(self, view): self.sc = None self.view = view self.timer = wx.Timer() self.timer.Bind(wx.EVT_TIMER, self.OnTimer) self.refresh_rate = 2000 self.mode = False self.mbclient = None pub.subscribe(self.startTimer, 'monitoring_started') pub.subscribe(self.stopTimer, 'monitoring_stopped') pub.subscribe(self.applyNetworkConfigs, 'apply_network_configs') pub.subscribe(self.resetNetworkConfigs, 'reset_network_configs') pub.subscribe(self.resetPcsRemote, 'reset_pcs_remote') pub.subscribe(self.checkSysOpMode, 'check_system_op_mode') pub.subscribe(self.startSystemOp, "start_system_op_manual") pub.subscribe(self.stopSystemOp, "stop_system_op_manual") pub.subscribe(self.turnRelayOn, "turn_relay_on") pub.subscribe(self.turnRelayOff, "turn_relay_off") ''' def WriteSingleCoil(self, startingAddress, value) Write single Coil to Server device (Function code 5) startingAddress: Coil to be written value: Coil Value to be written def WriteMultipleCoils(self, startingAddress, values) Write multiple coils to Server device (Function code 15) startingAddress : First coil to be written values: Coil Values [0..quantity-1] to be written ''' def turnRelayOn(self, relay_id): ''' :param relay_id: id of relay to turn on :return: None ''' self.mbclient.write_single_coil(relay_id-1, True) def turnRelayOff(self, relay_id): ''' :param relay_id: id of relay to turn off :return: None ''' self.mbclient.write_single_coil(relay_id-11, False) def startSystemOp(self, mode, val): self.sc.setPortModes(mode) self.sc.setPortSetpoints(val) self.sc.setUserStart() def stopSystemOp(self): self.sc.setUserStop() def startTimer(self): self.mbclient.connect() self.timer.Start(self.refresh_rate) def stopTimer(self): self.mbclient.close() self.timer.Stop() def applyNetworkConfigs(self, ncDict): if self.mbclient: if self.mbclient.is_connected(): self.mbclient.close() self.mbclient = easymodbus.modbusClient.ModbusClient(ncDict["ipaddr"], ncDict["port"]) self.mbclient.unitidentifier = ncDict["uid"] self.refresh_rate = ncDict.pop("refresh") def resetNetworkConfigs(self): if self.mbclient: if self.mbclient.is_connected(): self.mbclient.close() self.mbclient = None def resetPcsRemote(self): if self.sc: self.sc.resetPcs() def checkSysOpMode(self): if self.sc: self.mode = self.sc.checkOpMode() a = "Auto" if self.mode else "Manual" self.view.text_op_mode.SetValue(a) def OnTimer(self, event): # Read registers relay_status = self.mbclient.read_coils(0,10) # Process registers # Show Registers for i, rstatus in enumerate(relay_status): if rstatus: self.view.indicators[i].SetValue("ON") self.view.indicators[i].SetBackgroundColour(wx.Colour(0, 255, 0)) else: self.view.indicators[i].SetValue("OFF") self.view.indicators[i].SetBackgroundColour(wx.Colour(255, 0, 0)) class MyFrame(FaultFrame): def __init__(self, *args, **kwds): FaultFrame.__init__(self, *args, **kwds) pub.subscribe(self.update_display, 'read_all_registers') self.indicators = [ self.text_ctrl_1, self.text_ctrl_2, self.text_ctrl_3, self.text_ctrl_4, self.text_ctrl_5, self.text_ctrl_6, self.text_ctrl_7, self.text_ctrl_8, self.text_ctrl_9, self.text_ctrl_10, ] def update_status(self, stats): self.text_ctrl_system_status.SetValue(stats[0]) self.text_ctrl_ac1_status.SetValue(stats[1]) self.text_ctrl_dc2_status.SetValue(stats[2]) self.text_ctrl_dc3_status.SetValue(stats[3]) def update_display(self, reg, info): ports = {k: v for (k, v) in info.items() if re.search(r'^p[1-3]_', k)} etc = {k: v for (k, v) in info.items() if not re.search(r'^p[1-3]_', k)} for k, v in ports.items(): self.property_page.SetPropertyValue(k, str(v.convertValue(reg[v.address]))) self.property_page.SetPropertyAttribute(k, "Units", str((reg[v.address]))) for k, v in etc.items(): self.property_page_etc.SetPropertyValue(k, str(v.convertValue(reg[v.address]))) self.property_page_etc.SetPropertyAttribute(k, "Units", str((reg[v.address]))) def OnFaultSelect(self, event): # wxGlade: MyFrame. pub.sendMessage("fault_selected", pos=(event.GetRow(), event.GetCol())) event.Skip() def SetFaultGrid(self, fActivity, fOccurence): if len(fActivity) == len(fOccurence) == 64: for i in range(8): for j in range(8): if fActivity[i * 8 + j] == '1': self.fault_grid.SetCellBackgroundColour(i, j, wx.LIGHT_GREY) else: self.fault_grid.SetCellBackgroundColour(i, j, wx.WHITE) if fOccurence[i * 8 + j] == '1': self.fault_grid.SetCellValue(i, j, "O") else: self.fault_grid.SetCellValue(i, j, "") def ToggleMonitoring(self, event): # wxGlade: MyFrame. btnLabel = self.button_1.GetLabel() if self.button_1.Value: pub.sendMessage("monitoring_started") self.button_1.SetLabel("Stop") else: pub.sendMessage("monitoring_stopped") self.button_1.SetLabel("Start") def OnApply(self, event): if self.button_1.Enabled: return self.input_uid.Disable() self.input_port.Disable() self.input_ip.Disable() self.input_refresh.Disable() ncDict = { "ipaddr": self.input_ip.GetValue(), "port": int(self.input_port.GetValue()), "uid": int(self.input_uid.GetValue()), "refresh": int(self.input_refresh.GetValue()), } pub.sendMessage("apply_network_configs", ncDict=ncDict) self.button_1.Enable() def OnReset(self, event=None): if not self.button_1.Enabled: return pub.sendMessage("reset_network_configs") self.button_1.SetValue(False) self.ToggleMonitoring(None) self.button_1.Disable() self.input_uid.Enable() self.input_port.Enable() self.input_ip.Enable() self.input_refresh.Enable() def OnButtonRemoteReset(self, event): pub.sendMessage("reset_pcs_remote") self.OnReset() def OnCheckSysOpMode(self, event): pub.sendMessage("check_system_op_mode") def OnRelayOn(self, event): pub.sendMessage("turn_relay_on", relay_id = event.GetId()) event.Skip() def OnRelayOff(self, event): pub.sendMessage("turn_relay_off", relay_id = event.GetId()) event.Skip() def OnRelayPulse(self, event): event.Skip() # end of class MyFrame class MyApp(wx.App): def OnInit(self): self.frame = MyFrame(None, wx.ID_ANY, "") self.c = Controller(self.frame) self.SetTopWindow(self.frame) self.frame.Show() return True # end of class MyApp if __name__ == "__main__": gettext.install("app") # replace with the appropriate catalog name app = MyApp(0) app.MainLoop() if app.c.sc: app.c.sc.checkClose()