lpcmanager

Moved CanFestival plugin customization code to LPCCanFestival.py
revamp
2018-02-05, Edouard Tisserant
163b46d0412b
Moved CanFestival plugin customization code to LPCCanFestival.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import fnmatch
import shutil
import zipfile
import tempfile
from ProjectController import ProjectController
def mycopytree(src, dst):
"""
Copy content of a directory to an other, omit hidden files
@param src: source directory
@param dst: destination directory
"""
for i in os.listdir(src):
if not i.startswith('.'):
srcpath = os.path.join(src, i)
dstpath = os.path.join(dst, i)
if os.path.isdir(srcpath):
if os.path.exists(dstpath):
shutil.rmtree(dstpath)
os.makedirs(dstpath)
mycopytree(srcpath, dstpath)
elif os.path.isfile(srcpath):
shutil.copy2(srcpath, dstpath)
[SIMULATION_MODE, TRANSFER_MODE] = range(2)
if arch in PLC_module:
class LPCProjectNodeEditor(ProjectNodeEditor):
pass
else:
class LPCProjectNodeEditor(ProjectNodeEditor):
SHOW_PARAMS = False
ENABLE_REQUIRED = False
_StatusMethods = [
{"bitmap": "Debug",
"name": _("Simulate"),
"tooltip": _("Simulate PLC"),
"method": "_Simulate"},
{"bitmap": "Run",
"name": _("Run"),
"shown": False,
"tooltip": _("Start PLC"),
"method": "_Run"},
{"bitmap": "Stop",
"name": _("Stop"),
"shown": False,
"tooltip": _("Stop Running PLC"),
"method": "_Stop"},
{"bitmap": "Build",
"name": _("Build"),
"tooltip": _("Build project into build folder"),
"method": "_Build"},
{"bitmap": "Clean",
"name": _("Clean"),
"enabled": False,
"tooltip": _("Clean project build folder"),
"method": "_Clean"},
{"bitmap": "Transfer",
"name": _("Transfer"),
"shown": False,
"tooltip": _("Transfer PLC"),
"method": "_Transfer"},
]
_MethodFromPLCState = {
"Started": [("_Simulate", False),
("_Run", False),
("_Stop", True),
("_Build", True),
("_Transfer", True)],
"Stopped": [("_Simulate", False),
("_Run", True),
("_Stop", False),
("_Build", True),
("_Transfer", True)],
"Connected": [("_Simulate", "not simulating"),
("_Run", True),
("_Stop", True if arch in PLC_module else "simulating"),
("_Build", True),
("_Transfer", True)],
"Disconnected": [("_Simulate", "not simulating"),
("_Run", False),
("_Stop", False if arch in PLC_module else "simulating"),
("_Build", True),
("_Transfer", False)],
"Empty" :[("_Simulate", "not simulating"),
("_Run", False),
("_Stop", False),
("_Build", True),
("_Transfer", True)],
}
if arch in PLC_module:
_StatusMethods += [
{"bitmap": "Connect",
"name": _("Connect"),
"shown": True,
"tooltip": _("Connect to the target PLC"),
"method": "_Connect"},
{"bitmap": "Disconnect",
"name": _("Disconnect"),
"shown": False,
"tooltip": _("Disconnect from PLC"),
"method": "_Disconnect"},
{"bitmap": "UpdateFw",
"name": _("UpdateFw"),
"shown": True,
"tooltip": _("Update the PLC firmware"),
"method": "_UpdateFw"},
]
_MethodFromPLCState["Disconnected"] += [("_Connect", True),
("_Disconnect", False)]
_MethodFromPLCState["Connected"] += [("_Connect", False),
("_Disconnect", True)]
class LPCProjectController(ProjectController):
StatusMethods = _StatusMethods
# ConfNodeMethods = []
EditorType = LPCProjectNodeEditor
def __init__(self, frame, logger, buildpath):
self.arch = arch
self.OrigBuildPath = buildpath
ProjectController.__init__(self, frame, logger)
self.CTNChildrenTypes += [("LPCBus", LPCBus, "LPC bus"), ("CanOpen", LPCCanOpen, "CanOpen bus")]
self.CTNType = "LPC"
self.OnlineMode = "NORMAL" if arch in PLC_module else "OFF"
self.LPCConnector = None
self.ConnectorPath = None
self.CurrentMode = None
self.previous_mode = None
self.SimulationBuildPath = None
self.AbortTransferTimer = None
# Firmware update running status
self.firmawreUpadateIsRunning = False
self.building = False
# Bind mouse double click event on URI_location in Beremiz
self.UriOptions = True
def GetProjectName(self):
return self.Project.getname()
def GetDefaultTargetName(self):
if self.CurrentMode == SIMULATION_MODE:
return ProjectController.GetDefaultTargetName(self)
else:
return "LPC"
def GetTarget(self):
target = ProjectController.GetTarget(self)
if self.CurrentMode != SIMULATION_MODE and arch not in PLC_module:
target.getcontent().setBuildPath(self.BuildPath)
return target
def _getBuildPath(self):
if self.CurrentMode == SIMULATION_MODE:
if self.SimulationBuildPath is None:
self.SimulationBuildPath = os.path.join(tempfile.mkdtemp(), os.path.basename(self.ProjectPath), "build")
return self.SimulationBuildPath
else:
return ProjectController._getBuildPath(self)
def ToZIPFile(self):
# MD5 = self.GetLastBuildMD5()
try:
path_export_file = self.BuildPath[:-5] + "\\" + self._builder.exe[:-3] + ".xEye"
zf = zipfile.ZipFile(path_export_file, mode='w', compression=zipfile.ZIP_DEFLATED)
for extrafilespath in [self._getExtraFilesPath(),
self._getProjectFilesPath()]:
dir = extrafilespath.split("\\")[-1]
for name in os.listdir(extrafilespath):
zf.write(extrafilespath + '\\' + str(name), dir + '\\' + str(name), zipfile.ZIP_DEFLATED)
zf.write(self._builder.exe_path, self._builder.exe)
zf.write(self.BuildPath + '\\lastbuildPLC.md5', 'lastbuildPLC.md5')
zf.close()
self.logger.write(_("Export file is successfully created on location: %s\n") % path_export_file)
except Exception, e:
self.logger.write(_("Export file is not created because eror: %s\n") % e)
def _Build(self):
self._Clean()
save = self.ProjectTestModified()
if save:
self.SaveProject()
self.AppFrame._Refresh(TITLE, FILEMENU)
if self.BuildPath is not None:
mycopytree(self.OrigBuildPath, self.BuildPath)
if ProjectController._Build(self):
self.ToZIPFile()
if save:
wx.CallAfter(self.AppFrame.RefreshAll)
def SetProjectName(self, name):
return self.Project.setname(name)
def SetOnlineMode(self, mode, path=None):
# SetOnlineMode is only for MC8
if arch in PLC_module:
return None
mode = mode.upper()
if self.OnlineMode != mode:
if mode not in ["OFF", ""]:
self.OnlineMode = mode
self.ConnectorPath = path
uri = "LPC://%s/%s" % (self.OnlineMode, path)
try:
self.LPCConnector = connectors.ConnectorFactory(uri, self)
except Exception, msg:
self.logger.write_error(_("Exception while connecting %s!\n") % uri)
self.logger.write_error(traceback.format_exc())
# Did connection success ?
if self.LPCConnector is None:
# Oups.
self.logger.write_error(_("Connection failed to %s!\n") % uri)
else:
self.OnlineMode = "OFF"
self.LPCConnector = None
self.ConnectorPath = None
self.ApplyOnlineMode()
def ApplyOnlineMode(self):
if self.CurrentMode != SIMULATION_MODE:
self.KillDebugThread()
self._SetConnector(self.LPCConnector)
# Init with actual PLC status and print it
self.UpdateMethodsFromPLCStatus()
if self.LPCConnector is not None and self.OnlineMode == "APPLICATION":
self.CompareLocalAndRemotePLC()
if self.previous_plcstate is not None:
status = _(self.previous_plcstate)
else:
status = ""
self.logger.write(_("PLC is %s\n") % status)
# if self.StatusTimer and not self.StatusTimer.IsRunning():
# # Start the status Timer
# self.StatusTimer.Start(milliseconds=2000, oneShot=False)
if self.previous_plcstate == "Started":
if self.DebugAvailable() and self.GetIECProgramsAndVariables():
self.logger.write(_("Debug connect matching running PLC\n"))
self._connect_debug()
else:
self.logger.write_warning(_("Debug do not match PLC - stop/transfert/start to re-enable\n"))
elif self.StatusTimer and self.StatusTimer.IsRunning():
self.StatusTimer.Stop()
if self.CurrentMode == TRANSFER_MODE:
if self.OnlineMode == "BOOTLOADER":
self.BeginTransfer()
elif self.OnlineMode == "APPLICATION":
self.CurrentMode = None
self.AbortTransferTimer.Stop()
self.AbortTransferTimer = None
self.logger.write(_("PLC transferred successfully\n"))
# Update a PLCOpenEditor Pou variable location
def UpdateProjectVariableLocation(self, old_leading, new_leading):
self.Project.updateElementAddress(old_leading, new_leading)
self.BufferProject()
# Update a PLCOpenEditor Pou variable name
def UpdateProjectVariableName(self, old_name, new_name):
self.Project.updateElementName(old_name, new_name)
self.BufferProject()
def RemoveProjectVariableByAddress(self, address):
self.Project.removeVariableByAddress(address)
self.BufferProject()
def RemoveProjectVariableByFilter(self, leading):
self.Project.removeVariableByFilter(leading)
self.BufferProject()
def AddProjectDefaultConfiguration(self, config_name="config", res_name="resource1"):
ProjectController.AddProjectDefaultConfiguration(self, config_name, res_name)
self.SetEditedResourceInfos(
self.ComputeConfigurationResourceName(config_name, res_name),
[{"Name": "main_task",
"Triggering": "Cyclic",
"Interval": "T#50ms",
"Priority": 0}],
[])
def LoadProject(self, ProjectPath, BuildPath=None):
"""
Load a project contained in a folder
@param ProjectPath: path of the project folder
"""
if os.path.basename(ProjectPath) == "":
ProjectPath = os.path.dirname(ProjectPath)
# Verify that project contains a PLCOpen program
plc_file = os.path.join(ProjectPath, "plc.xml")
if os.path.isfile(plc_file):
# Load PLCOpen file
result = self.OpenXMLFile(plc_file)
if result:
return result
else:
self.CreateNewProject({"companyName": "",
"productName": "",
"productVersion": "",
"projectName": "",
"pageSize": (0, 0),
"scaling": {}})
if len(self.GetProjectConfigNames()) == 0:
self.AddProjectDefaultConfiguration()
# Change XSD into class members
self._AddParamsMembers()
self.Children = {}
# Keep track of the root confnode (i.e. project path)
self.ProjectPath = ProjectPath
self.BuildPath = self._getBuildPath()
if self.OrigBuildPath is not None:
mycopytree(self.OrigBuildPath, self.BuildPath)
# If dir have already be made, and file exist
if os.path.isdir(self.CTNPath()) and os.path.isfile(self.ConfNodeXmlFilePath()):
# Load the confnode.xml file into parameters members
result = self.LoadXMLParams()
if result:
return result
# Load and init all the children
self.LoadChildren()
canopen_child = self.GetChildByName("CanOpen")
if canopen_child is None:
canopen = self.CTNAddChild("CanOpen", "CanOpen", 0)
canopen.LoadChildren()
canopen.CTNRequestSave()
if self.CTNTestModified():
self.SaveProject()
if wx.GetApp() is None:
self.RefreshConfNodesBlockLists()
else:
wx.CallAfter(self.RefreshConfNodesBlockLists)
if arch in PLC_module:
self.SetParamsAttribute('BeremizRoot.TargetType', 'MC9')
return None
def IsPLCStarted(self):
return self.previous_plcstate == "Started" or self.previous_mode == SIMULATION_MODE
def ShowMethod(self, name, val):
simulating = self.CurrentMode == SIMULATION_MODE
if type(val) == str:
if val.endswith("simulating"):
if val.startswith("not"):
val = not simulating
else:
val = simulating
ProjectController.ShowMethod(self, name, val)
def UpdateMethodsFromPLCStatus(self):
simulating = self.CurrentMode == SIMULATION_MODE
if self.OnlineMode == "OFF":
if simulating:
status, log_count = self._connector.GetPLCstatus()
self.UpdatePLCLog(log_count)
status = "Disconnected"
elif self.OnlineMode == "BOOTLOADER":
status = "Connected"
else:
if self._connector is not None:
status, log_count = self._connector.GetPLCstatus()
if status == "Disconnected":
self._SetConnector(None, False)
else:
self.UpdatePLCLog(log_count)
else:
status = "Disconnected"
if self.previous_plcstate != status or self.previous_mode != self.CurrentMode:
for args in _MethodFromPLCState.get(status, []):
self.ShowMethod(*args)
self.previous_plcstate = status
self.previous_mode = self.CurrentMode
if self.AppFrame is not None:
self.AppFrame.RefreshStatusToolBar()
connection_text = _("Connected to: ")
status_text = ""
if simulating:
connection_text += _("Simulation")
status_text += _("ON")
if status == "Disconnected":
if not simulating:
self.AppFrame.ConnectionStatusBar.SetStatusText(_(status), 1)
self.AppFrame.ConnectionStatusBar.SetStatusText('', 2)
else:
self.AppFrame.ConnectionStatusBar.SetStatusText(connection_text, 1)
self.AppFrame.ConnectionStatusBar.SetStatusText(status_text, 2)
else:
if simulating:
connection_text += " (%s)"
status_text += " (%s)"
else:
connection_text += "%s"
status_text += "%s"
self.AppFrame.ConnectionStatusBar.SetStatusText(connection_text % self.ConnectorPath, 1)
self.AppFrame.ConnectionStatusBar.SetStatusText(status_text % _(status), 2)
return True
return False
def Generate_plc_declare_locations(self):
"""
Declare used locations in order to simulatePLC in a black box
"""
return """#include "iec_types_all.h"
#define __LOCATED_VAR(type, name, ...) \
type beremiz_##name;\
type *name = &beremiz_##name;
#include "LOCATED_VARIABLES.h"
#undef __LOCATED_VAR
"""
def Generate_lpc_retain_array_sim(self):
"""
Support for retain array in Simulation
"""
return """/* Support for retain array */
#define USER_RETAIN_ARRAY_SIZE 2000
#define NUM_OF_COLS 3
unsigned char readOK = 0;
unsigned int foundIndex = USER_RETAIN_ARRAY_SIZE;
unsigned int retainArray[USER_RETAIN_ARRAY_SIZE][NUM_OF_COLS];
unsigned int __GetRetainData(unsigned char READ, unsigned int INDEX, unsigned int COLUMN)
{
if(READ == 1)
{
if((0<=INDEX) && (INDEX<USER_RETAIN_ARRAY_SIZE) && (0<=COLUMN) && (COLUMN<NUM_OF_COLS))
{
readOK = 1;
return retainArray[INDEX][COLUMN];
}
}
readOK = 0;
return 0;
}
unsigned char __SetRetainData(unsigned char WRITE, unsigned int INDEX, unsigned int WORD1, unsigned int WORD2, unsigned int WORD3)
{
if(WRITE == 1)
{
if((0<=INDEX) && (INDEX<USER_RETAIN_ARRAY_SIZE))
{
retainArray[INDEX][0] = WORD1;
retainArray[INDEX][1] = WORD2;
retainArray[INDEX][2] = WORD3;
return 1;
}
}
return 0;
}
unsigned char __FindRetainData(unsigned char SEARCH, unsigned int START_IDX, unsigned int END_IDX, unsigned int WORD1, unsigned int WORD2, unsigned int WORD3)
{
unsigned int i;
if((SEARCH==1) && (0<=START_IDX) && (START_IDX<USER_RETAIN_ARRAY_SIZE) && (START_IDX<=END_IDX) && (END_IDX<USER_RETAIN_ARRAY_SIZE))
{
for(i=START_IDX;i<=END_IDX;i++)
{
if((retainArray[i][0] == WORD1) && (retainArray[i][1] == WORD2) && (retainArray[i][2] == WORD3))
{
foundIndex = i;
return 1;
}
}
}
foundIndex = USER_RETAIN_ARRAY_SIZE; /* Data not found => return index that is out of array bounds */
return 0;
}
/* Since Beremiz debugger doesn't like pointer-by-reference stuff or global varibles, separate function is a must */
unsigned char __GetReadStatus(unsigned char dummy)
{
return readOK;
}
unsigned int __GetFoundIndex(unsigned char dummy)
{
return foundIndex;
}
"""
def _Simulate(self):
"""
Method called by user to Simulate PLC
"""
uri = "LOCAL://"
try:
self._SetConnector(connectors.ConnectorFactory(uri, self))
except Exception, msg:
self.logger.write_error(_("Exception while connecting %s!\n") % uri)
self.logger.write_error(traceback.format_exc())
# Did connection success ?
if self._connector is None:
# Oups.
self.logger.write_error(_("Connection failed to %s!\n") % uri)
self.StopSimulation()
return False
self.CurrentMode = SIMULATION_MODE
buildpath = self._getBuildPath()
# Eventually create build dir
if not os.path.exists(buildpath):
os.makedirs(buildpath)
# Generate SoftPLC IEC code
IECGenRes = self._Generate_SoftPLC()
# If IEC code gen fail, bail out.
if not IECGenRes:
self.logger.write_error(_("IEC-61131-3 code generation failed !\n"))
self.StopSimulation()
return False
# Reset variable and program list that are parsed from
# CSV file generated by IEC2C compiler.
self.ResetIECProgramsAndVariables()
gen_result = self.CTNGenerate_C(buildpath, self.PLCGeneratedLocatedVars)
CTNCFilesAndCFLAGS, CTNLDFLAGS, DoCalls = gen_result[:3]
# if some files have been generated put them in the list with their location
if CTNCFilesAndCFLAGS:
self.LocationCFilesAndCFLAGS = [(self.GetCurrentLocation(), CTNCFilesAndCFLAGS, DoCalls)]
else:
self.LocationCFilesAndCFLAGS = []
# confnode asks for some LDFLAGS
if CTNLDFLAGS:
# LDFLAGS can be either string
if type(CTNLDFLAGS) == type(str()):
self.LDFLAGS = [CTNLDFLAGS]
# or list of strings
elif type(CTNLDFLAGS) == type(list()):
self.LDFLAGS = CTNLDFLAGS[:]
else:
self.LDFLAGS = []
# Header file for extensions
open(os.path.join(buildpath, "beremiz.h"), "w").write(targets.GetHeader())
# Template based part of C code generation
# files are stacked at the beginning, as files of confnode tree root
for generator, filename, name in [
# debugger code
(self.Generate_plc_debugger, "plc_debugger.c", "Debugger"),
# init/cleanup/retrieve/publish, run and align code
(self.Generate_plc_main, "plc_main.c", "Common runtime"),
# declare located variables for simulate in a black box
(self.Generate_plc_declare_locations, "plc_declare_locations.c", "Declare Locations"),
# declare located variables for simulate in a black box
(self.Generate_lpc_retain_array_sim, "lpc_retain_array_sim.c", "Retain Array for Simulation")]:
try:
# Do generate
code = generator()
if code is None:
raise
code_path = os.path.join(buildpath, filename)
open(code_path, "w").write(code)
# Insert this file as first file to be compiled at root confnode
self.LocationCFilesAndCFLAGS[0][1].insert(0, (code_path, self.plcCFLAGS))
except Exception, exc:
self.logger.write_error(name + _(" generation failed !\n"))
self.logger.write_error(traceback.format_exc())
self.StopSimulation()
return False
# Get simulation builder
builder = self.GetBuilder()
if builder is None:
self.logger.write_error(_("Fatal : cannot get builder.\n"))
self.StopSimulation()
return False
# Build
try:
if not builder.build():
self.logger.write_error(_("C Build failed.\n"))
self.StopSimulation()
return False
except Exception, exc:
self.logger.write_error(_("C Build crashed !\n"))
self.logger.write_error(traceback.format_exc())
self.StopSimulation()
return False
data = builder.GetBinaryCode()
if data is not None:
if self._connector.NewPLC(builder.GetBinaryCodeMD5(), data, []):
self.UnsubscribeAllDebugIECVariable()
self.ProgramTransferred()
if self.AppFrame is not None:
self.AppFrame.CloseObsoleteDebugTabs()
self.AppFrame.RefreshPouInstanceVariablesPanel()
self.logger.write(_("Transfer completed successfully.\n"))
else:
self.logger.write_error(_("Transfer failed\n"))
self.StopSimulation()
return False
self._Run()
if not self.StatusTimer.IsRunning():
# Start the status Timer
self.StatusTimer.Start(milliseconds=500, oneShot=False)
def StopSimulation(self):
self.CurrentMode = None
self._SetConnector(None, False)
self.ApplyOnlineMode()
def _Stop(self):
ProjectController._Stop(self)
if self.CurrentMode == SIMULATION_MODE:
self.StopSimulation()
def CompareLocalAndRemotePLC(self):
# if self.LPCConnector is None:
# return
if self._connector is None:
return
# We are now connected. Update button status
MD5 = self.GetLastBuildMD5()
# Check remote target PLC correspondance to that md5
# if MD5 is not None and self.LPCConnector.MatchMD5(MD5):
if MD5 is not None and self._connector.MatchMD5(MD5):
# warns controller that program match
self.ProgramTransferred()
def ResetBuildMD5(self):
builder = self.GetBuilder()
if builder is not None:
builder.ResetBinaryCodeMD5(*([] if arch in PLC_module else [self.OnlineMode]))
def GetLastBuildMD5(self):
builder = self.GetBuilder()
if builder is not None:
return builder.GetBinaryCodeMD5(*([] if arch in PLC_module else [self.OnlineMode]))
else:
return None
def _Clean(self, building = False):
self._CloseView(self._IECCodeView)
runtime_list = fnmatch.filter(os.listdir(self._getBuildPath()), 'runtime_*')
if os.path.isdir(os.path.join(self._getBuildPath())) and os.path.isfile(
os.path.join(self._getBuildPath(), "hmi.py")) or runtime_list != []:
self.logger.write(_("Cleaning the build directory\n"))
# self.logger.write(_(str(os.path.join(self._getBuildPath(), "hmi.py"))))
for file in runtime_list:
os.remove(os.path.join(self._getBuildPath(), file))
if os.path.isfile(os.path.join(self._getBuildPath(), "hmi.py")):
os.remove(os.path.join(self._getBuildPath(), "hmi.py"))
else:
if not building:
self.logger.write_error(_("Build directory already clean\n"))
self.ShowMethod("_showIECcode", False)
self.EnableMethod("_Clean", False)
# kill the builder
self._builder = None
self.CompareLocalAndRemotePLC()
def _Transfer(self):
if self.OnlineMode == "NORMAL":
if self.IsPLCStarted():
dialog = wx.MessageDialog(self.AppFrame, "You must stop the PLC before transfer. Do you want to stop it now and transfer?", style=wx.YES_NO|wx.CENTRE)
if dialog.ShowModal() == wx.ID_YES:
self._Stop()
else:
return
ProjectController._Transfer(self)
return
if self.CurrentMode is None and self.OnlineMode != "OFF":
self.CurrentMode = TRANSFER_MODE
if ProjectController._Build(self):
ID_ABORTTRANSFERTIMER = wx.NewId()
self.AbortTransferTimer = wx.Timer(self.AppFrame, ID_ABORTTRANSFERTIMER)
self.AppFrame.Bind(wx.EVT_TIMER, self.AbortTransfer, self.AbortTransferTimer)
if self.OnlineMode == "BOOTLOADER":
self.BeginTransfer()
else:
self.logger.write(_("Resetting PLC\n"))
# self.StatusTimer.Stop()
self.LPCConnector.ResetPLC()
self.AbortTransferTimer.Start(milliseconds=5000, oneShot=True)
else:
self.CurrentMode = None
def BeginTransfer(self):
self.logger.write(_("Start PLC transfer\n"))
self.AbortTransferTimer.Stop()
ProjectController._Transfer(self)
self.AbortTransferTimer.Start(milliseconds=5000, oneShot=True)
def AbortTransfer(self, event):
self.logger.write_warning(_("Timeout waiting PLC to recover\n"))
self.CurrentMode = None
self.AbortTransferTimer.Stop()
self.AbortTransferTimer = None
event.Skip()
old_DebugThreadProc = ProjectController.DebugThreadProc
def LPCManager_DebugThreadProc(self, checking=False):
if checking:
state, ls = self._connector.GetTraceVariables()
while state == "Stopped":
state, ls = self._connector.GetTraceVariables()
time.sleep(0.1)
self.old_DebugThreadProc()
def _connect_debug(self, checking = False):
self.previous_plcstate = None
if self.AppFrame:
self.AppFrame.ResetGraphicViewers()
self.RegisterDebugVarToConnector()
if self.DispatchDebugValuesTimer is not None:
self.DispatchDebugValuesTimer.Start(
int(REFRESH_PERIOD * 1000), oneShot=True)
if self.DebugThread is None:
# self.DebugThread = Thread(target=self.DebugThreadProc)
self.DebugThread = Thread(target=self.LPCManager_DebugThreadProc, args=(checking, ))
self.DebugThread.setDaemon(True)
self.DebugThread.start()
def _Run(self):
"""
Start PLC
"""
if self.GetIECProgramsAndVariables():
self._connector.StartPLC()
self.logger.write(_("Starting PLC\n"))
self._connect_debug(True)
else:
self.logger.write_error(_("Couldn't start PLC !\n"))
self.UpdateMethodsFromPLCStatus()
def _Disconnect(self):
self._SetConnector(None)
self.KillDebugThread()
def _UpdateFw(self):
"""
Method called by user to flash the firmware of the PLC
"""
from FirmwareUpdateDialog import FirmwareUpdateDialog
from dialogs import DiscoveryDialog
from HostFirmwareUpdater import HostFirmwareUpdater
if self.firmawreUpadateIsRunning == True:
self.logger.write_error(_("Firmware update is already running!\n"))
return
self.firmawreUpadateIsRunning = True
self.logger.write(_("Firmware update started\n"))
# Launch the firmware selection dialog
dialog = FirmwareUpdateDialog(self.AppFrame)
answer = dialog.ShowModal()
imageFilePath = dialog.GetFirmwareImageFile()
updateType = dialog.GetFirmwareUpdateType()
chunksSize = dialog.GetChunksSize()
reboot = dialog.GetReboot()
dialog.Destroy()
if answer == wx.ID_CANCEL:
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
if imageFilePath == None or imageFilePath == "":
self.logger.write_error(_("No firmware image file selected!\n"))
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
else:
self.logger.write(_("Firmware image file: %s\n") % imageFilePath)
if updateType == 1:
self.logger.write(_("Firmware update type: Linux kernel image\n"))
elif updateType == 2:
self.logger.write(_("Firmware update type: Linux DTB image\n"))
elif updateType == 3:
self.logger.write(_("Firmware update type: Root file system image\n"))
else:
self.logger.write_error(_("Unknown firmware update type!\n"))
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
if chunksSize < 1024 or chunksSize > 1024 * 1024:
self.logger.write_error(_("Bad chunks size : %d KiB!\n") % chunksSize)
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
else:
self.logger.write(_("Chunks size : %d KiB\n") % chunksSize)
self.logger.write(_("Reboot after update : %s\n") % ("Yes" if reboot else "No"))
# Get the target PLC URI
# Check if an uri is already configured in the Beremiz project
uri = self.BeremizRoot.getURI_location()
if uri is not None and uri != "":
self.logger.write(_("PLC URI configured in the Beremiz project will be used: %s\n") % uri)
else:
# PLC URI is not configured in the Beremiz project
# Launch Service Discovery dialog
self.logger.write(_("PLC URI is not configured in the Beremiz project. Launching the Discover dialog\n"))
dialog = DiscoveryDialog(self.AppFrame)
answer = dialog.ShowModal()
uri = dialog.GetURI()
dialog.Destroy()
# Nothing choosed or cancel button
if uri is None or answer == wx.ID_CANCEL:
self.logger.write_error(_("Connection canceled!\n"))
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
# Get connector from uri
self._connector = None
try:
self._SetConnector(connectors.ConnectorFactory(uri, self))
except Exception:
self.logger.write_error(_("Exception while connecting %s!\n") % uri)
self.logger.write_error(traceback.format_exc())
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
# Did connection success ?
if self._connector is None:
# Oups.
self.logger.write_error(_("Connection failed to %s!\n") % uri)
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
else:
self.logger.write(_("Connected.\n"))
# Last confirmation before firmware update
answer = wx.MessageBox(_('Are you sure to launch the firmware update for the selected PLC?'),
_('Firmware Update'), wx.YES_NO | wx.CENTRE | wx.NO_DEFAULT,
self.AppFrame)
if answer != wx.YES:
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
return
# Some cosmetics
self.AppFrame.Refresh()
self.AppFrame.Update()
# Lauch the firmaware update on the remote PLC
updater = HostFirmwareUpdater(self._connector, imageFilePath, updateType, chunksSize, reboot, self.logger)
try:
updater.update()
except Exception as e:
self.logger.write_error(str(e) + "\n")
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmawreUpadateIsRunning = False
self._Disconnect()
return
self._Disconnect()
self.firmawreUpadateIsRunning = False
return True