258 lines
7.8 KiB
Python
258 lines
7.8 KiB
Python
#!/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 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.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.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 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
|
|
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.<event_handler>
|
|
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.<event_handler>
|
|
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):
|
|
event.Skip()
|
|
|
|
def OnRelayOff(self, event):
|
|
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()
|