lpcmanager

Enable IEC60870 server data loading and starting on PLC
iec60870
2 months ago, Dino Kosic
d8b12828e6aa
Enable IEC60870 server data loading and starting on PLC
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
import os
import zipfile
import wx
from ProjectController import ProjectController
from FirmwareUpdateDialog import FirmwareUpdateDialog
from HostFirmwareUpdater import HostFirmwareUpdater
from plcopen.types_enums import ComputeConfigurationResourceName
from LPCArch import GetLPCProductDesc,GetLPCProduct
LPCStatusMethods = [
{"bitmap": "UpdateFw",
"name": _("UpdateFw"),
"shown": True,
"tooltip": _("Update the PLC firmware"),
"method": "_UpdateFw"}
]
LPCDefaultMethods = ProjectController.DefaultMethods.copy()
LPCDefaultMethods["_UpdateFw"]=False
LPCMethodsFromStatus = ProjectController.MethodsFromStatus.copy()
LPCMethodsFromStatus["Started"]["_UpdateFw"] = True
LPCMethodsFromStatus["Stopped"]["_UpdateFw"] = True
LPCMethodsFromStatus["Empty"]["_UpdateFw"] = True
ProjectController.ConfNodeMethods += [
{
"bitmap": "ALARM",
"name": _("Manage alarms"),
"tooltip": _("Describe and translate alarms"),
"method": "_ManageAlarms"
},
]
class LPCProjectController(ProjectController):
StatusMethods = ProjectController.StatusMethods + LPCStatusMethods
DefaultMethods = LPCDefaultMethods
MethodsFromStatus = LPCMethodsFromStatus
def __init__(self, frame, logger, buildpath, arch):
self.arch = arch
self.OrigBuildPath = buildpath
ProjectController.__init__(self, frame, logger)
# Firmware update running status
self.firmwareUpadateIsRunning = False
def GetProjectName(self):
"""
Beremiz uses project directory name as project name
In LPCManager, project name is given as PLCOpen project name
and is passed at startup by SetProjectProperties command
"""
return self.Project.getname()
def xEyeExport(self):
"""
Zip project files for use in xEye
"""
opj = os.path.join
ops = os.path.split
output_file_path = opj(ops(self._getBuildPath())[0],
self.GetProjectName() + ".xEye")
zf = zipfile.ZipFile(output_file_path, mode='w',
compression=zipfile.ZIP_DEFLATED)
for extrafilespath in [self._getExtraFilesPath(),
self._getProjectFilesPath()]:
dirname = ops(extrafilespath)[-1]
for name in os.listdir(extrafilespath):
zf.write(opj(extrafilespath, str(name)), opj(dirname, str(name)), zipfile.ZIP_DEFLATED)
zf.write(self._builder.bin_path, self._builder.bin)
zf.write(opj(self._getBuildPath(), 'lastbuildPLC.md5'), 'lastbuildPLC.md5')
zf.close()
self.logger.write(_("xEye export file created at: %s\n") % output_file_path)
def _Build(self):
"""
AutoSave before regular build, and zip for xEye after
"""
if self.ProjectTestModified():
self.SaveProject()
self.AppFrame.RefreshAfterSave()
result = ProjectController._Build(self)
if result:
self.xEyeExport()
return result
def _Transfer(self):
product_name = GetLPCProduct().lower().replace("_","")
if not self._connector.CheckProductID(product_name):
message = (_("Product ID doesn't match selected target type.\n"))
dialog = wx.MessageDialog(
self.AppFrame, message, _("Error"), wx.OK | wx.ICON_ERROR)
dialog.ShowModal()
return False
return ProjectController._Transfer(self)
def SetProjectName(self, name):
return self.Project.setname(name)
def SetOnlineMode(self, mode, path=None):
# SetOnlineMode is only for MC8
return None
# 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(
ComputeConfigurationResourceName(config_name, res_name),
[{"Name": "main_task",
"Triggering": "Cyclic",
"Interval": "T#50ms",
"Priority": 0}],
[])
def LoadProject(self, ProjectPath, BuildPath=None):
"""
Create plcopen project if plc.xml not found,
so that original LoadProject doesn't make errors.
After regular load, add CanOpen nodes if not already there,
and eventualy save project again.
TargetType is forced to MC9.
"""
plc_file = os.path.join(ProjectPath, "plc.xml")
if not os.path.isfile(plc_file):
self.NewProject(ProjectPath, BuildPath)
else :
ProjectController.LoadProject(self, ProjectPath, BuildPath)
if GetLPCProductDesc().get("CAN", True):
canopen_child = self.GetChildByName("CanOpen")
if canopen_child is None:
canopen = self.CTNAddChild("CanOpen", "CanOpen", 0)
canopen.LoadChildren()
canopen.CTNRequestSave()
self.SetParamsAttribute('BeremizRoot.TargetType', self.arch)
if self.CTNTestModified():
self.SaveProject()
self.RefreshConfNodesBlockLists()
return None
def _ManageAlarms(self):
dlg = wx.SingleChoiceDialog(
self.AppFrame,
'Alarms translations', 'Choose a language to edit:',
['TODO'],
wx.CHOICEDLG_STYLE)
if dlg.ShowModal() == wx.ID_OK:
print('You selected: %s\n' % dlg.GetStringSelection())
# TODO XXX
dlg.Destroy()
def _UpdateFw(self):
"""
Method called by user to flash the firmware of the PLC
"""
if self.firmwareUpadateIsRunning == True:
self.logger.write_error(_("Firmware update is already running!\n"))
return
self.firmwareUpadateIsRunning = True
self.logger.write(_("Firmware update started\n"))
version_string = self._connector.GetVersions()
# Launch the firmware selection dialog
dialog = FirmwareUpdateDialog(self.AppFrame, version_string)
answer = dialog.ShowModal()
imageFilePath = dialog.GetFirmwareImageFile()
updateType = dialog.GetFirmwareUpdateType()
chunksSize = dialog.GetChunksSize()
dialog.Destroy()
if answer == wx.ID_CANCEL:
self.logger.write_error(_("Firmware update canceled!\n"))
self.firmwareUpadateIsRunning = 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.firmwareUpadateIsRunning = 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.firmwareUpadateIsRunning = 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.firmwareUpadateIsRunning = False
return
else:
self.logger.write(_("Chunks size : %d KiB\n") % chunksSize)
# 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.firmwareUpadateIsRunning = False
return
# do we still have a connection ?
if self._connector is None:
# Oups.
self.logger.write_error(_("Firmware update canceled because connection lost!\n"))
self.firmwareUpadateIsRunning = False
return
# Some cosmetics
self.AppFrame.Refresh()
self.AppFrame.Update()
dlg = wx.ProgressDialog(_("Firmware update"),
_("Updating firmware, please wait."),
maximum = 100,
parent=self.AppFrame,
style = 0
| wx.PD_APP_MODAL
| wx.PD_CAN_ABORT
| wx.PD_ESTIMATED_TIME
| wx.PD_REMAINING_TIME
)
# Lauch the firmaware update on the remote PLC
updater = HostFirmwareUpdater(self, imageFilePath, updateType, chunksSize, dlg)
try:
updater.update()
except Exception as e:
self.logger.write_error(str(e) + "\n")
self.logger.write_error(_("Firmware update failed!\n"))
if not dlg.WasCancelled():
dlg.Update(100, _("Firmware update failed!\n(%s)"%str(e)))
return
finally:
self.firmwareUpadateIsRunning = False
dlg.Destroy()
# Propose reboot
answer = wx.MessageBox(_('Firmware update requires rebooting and reconnecting to complete. Reboot PLC now?'),
_('Firmware Update'), wx.YES_NO | wx.CENTRE | wx.NO_DEFAULT,
self.AppFrame)
if answer == wx.YES:
self.logger.write(_("Rebooting PLC...\n"))
updater.RebootTarget()
self._Disconnect()
wx.MessageBox(_('Do not poweroff the device while it is rebooting. Wait until you can connect again to ensure update succeeded.'),
_('WARNING'), wx.OK | wx.CENTRE | wx.NO_DEFAULT,
self.AppFrame)
return
return True