#!/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 gettext.install(None) from pubsub import pub import re from registers import regdict print('pubsub API version', pub.VERSION_API) from pubsub.utils.notification import useNotifyByWriteFile import sys useNotifyByWriteFile(sys.stdout) import pprint from stabiliti import StabilitiController as SC from faults import faultInfo from FaultView import FaultFrame 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 pub.subscribe(self.getFaultInfo, 'fault_selected') 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") 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.timer.Start(self.refresh_rate) def stopTimer(self): self.timer.Stop() def applyNetworkConfigs(self, ncDict): if self.sc: self.sc.checkClose() self.refresh_rate = ncDict.pop("refresh") self.sc = SC(**ncDict) def resetNetworkConfigs(self): if self.sc: self.sc.checkClose() self.sc = 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 getFaultInfo(self, pos): y, x = pos fid = x + y*8 if 0 <= fid <= 63: try: info_text = faultInfo[fid] except IndexError: info_text = "Fault Number: {} (Reserved)".format(fid) self.view.text_fault_doc.SetValue(info_text) status_text = "Not Connected" if self.sc: fDict = self.sc.readFaultDetail(fid) status_text = pprint.pformat(fDict) self.view.text_fault_status.SetValue(status_text) def OnTimer(self, event): # Read registers self.sc.readAllRegisters() # Process registers # Show Registers factivity = self.sc.readFaultActivity() foccurence = self.sc.readFaultOccurence() self.view.SetFaultGrid(factivity, foccurence) stats = list(range(4)) stats[0], _ = (self.sc.getSystemStatus()) stats[1], _ = (self.sc.getPortStatusAC1()) stats[2], _ = (self.sc.getPortStatusDC2()) stats[3], _ = (self.sc.getPortStatusDC3()) self.view.update_status(list(map("\n".join, stats))) ''' for item in self.writeQueue: name, value = item self.sc.write(name, value) ''' class MyFrame(FaultFrame): def __init__(self, *args, **kwds): FaultFrame.__init__(self, *args, **kwds) self.property_page = self.property_grid_1.AddPage("Ports") self.property_page_etc = self.property_grid_1.AddPage("Etc") for k in regdict: prop = wx.propgrid.StringProperty(label=k, value="") prop.SetHelpString(str(regdict[k])) if re.search(r'^p[1-3]_', k): self.property_page.Append(prop) else: self.property_page_etc.Append(prop) self.property_page.Sort() self.property_page_etc.Sort() # self.property_page.SetPropertyAttributeAll("Raw", 0) # self.property_page_etc.SetPropertyAttributeAll("Raw", 0) self.property_grid_1.SetPropertyAttributeAll("Units", 0) self.property_grid_1.SetColumnCount(3, self.property_page.GetIndex()) self.property_grid_1.SetColumnCount(3, self.property_page_etc.GetIndex()) self.property_grid_1.ShowHeader() pub.subscribe(self.update_display, 'read_all_registers') 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 OnOpStart(self, event): mode = (self.choice_ac1_control.GetStringSelection(), self.choice_dc2_control.GetStringSelection(), self.choice_dc3_control.GetStringSelection()) inputs = { "p1_real_pwr_setpt": self.text_ctrl_ac1_realpwr.Value, "p1_react_pwr_setpt": self.text_ctrl_ac1_reactpwr.Value, "p2_current_setpt": self.text_ctrl_dc2_current.Value, "p2_power_setpt": self.text_ctrl_dc2_power.Value, "p2_voltage_setpt": self.text_ctrl_dc2_voltage.Value, "p3_current_setpt": self.text_ctrl_dc3_current.Value, "p3_power_setpt": self.text_ctrl_dc3_power.Value, "p3_voltage_setpt": self.text_ctrl_dc3_voltage.Value, } pub.sendMessage("start_system_op_manual", mode=mode, val=inputs) def OnOpStop(self, event): pub.sendMessage("stop_system_op_manual") # 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__": app = MyApp(0) app.MainLoop() if app.c.sc: app.c.sc.checkClose()