ed@448: #!/usr/bin/env python ed@448: # -*- coding: utf-8 -*- ed@448: ed@448: #This file is part of Beremiz, a Integrated Development Environment for ed@448: #programming IEC 61131-3 automates supporting plcopen standard and CanFestival. ed@448: # ed@448: #Copyright (C) 2007: Edouard TISSERANT and Laurent BESSARD ed@448: # ed@448: #See COPYING file for copyrights details. ed@448: # ed@448: #This library is free software; you can redistribute it and/or ed@448: #modify it under the terms of the GNU General Public ed@448: #License as published by the Free Software Foundation; either ed@448: #version 2.1 of the License, or (at your option) any later version. ed@448: # ed@448: #This library is distributed in the hope that it will be useful, ed@448: #but WITHOUT ANY WARRANTY; without even the implied warranty of ed@448: #MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU ed@448: #General Public License for more details. ed@448: # ed@448: #You should have received a copy of the GNU General Public ed@448: #License along with this library; if not, write to the Free Software ed@448: #Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ed@448: ed@448: import ctypes, os, commands, types, sys edouard@453: from LPCProto import * ed@448: ed@448: class LPCObject(): edouard@508: def __init__(self,pluginsroot, location): edouard@508: self.PLCStatus = "Disconnected" ed@448: self.pluginsroot = pluginsroot ed@448: self.PLCprint = pluginsroot.logger.write edouard@453: self._Idxs = [] edouard@508: self.UpdateLocation(location) edouard@508: edouard@508: def UpdateLocation(self, location): edouard@508: # Is that a comport ? edouard@508: if len(location) == 5 and\ edouard@508: location.startswith("COM") and \ edouard@508: location[3].isdigit() and \ edouard@508: location[4]==":" : edouard@508: self.StorageConnection = None edouard@453: try: edouard@508: comport = int(location[3]) - 1 edouard@508: self.SerialConnection = LPCProto(comport,#number edouard@508: 115200, #speed edouard@508: 2) #timeout edouard@508: # This will update status edouard@508: self.HandleSerialTransaction(IDLETransaction()) edouard@453: except Exception,e: edouard@453: self.pluginsroot.logger.write_error(str(e)+"\n") edouard@453: self.SerialConnection = None edouard@508: self.PLCStatus = "Disconnected" edouard@508: # or a drive unit ? edouard@508: elif len(location)==2 and \ edouard@508: location[0].isalpha() and \ edouard@508: location[1] == ':' : edouard@453: self.SerialConnection = None edouard@520: if os.path.exists(location): edouard@508: self.StorageConnection = location edouard@508: self.PLCStatus = "Stopped" edouard@508: else: edouard@508: self.pluginsroot.logger.write_error("Drive "+ edouard@508: location+ edouard@508: " do not exist !\n") edouard@508: self.StorageConnection = None edouard@508: self.PLCStatus = "Disconnected" edouard@508: edouard@508: def HandleSerialTransaction(self, transaction): edouard@508: if self.SerialConnection is not None: edouard@508: try: edouard@508: self.PLCStatus, res = self.SerialConnection.HandleTransaction(transaction) edouard@508: return res edouard@508: except LPCError,e: edouard@508: self.pluginsroot.logger.write_error(str(e)+"\n") edouard@508: self.SerialConnection = None edouard@508: self.PLCStatus = "Disconnected" edouard@508: return None edouard@453: ed@448: def StartPLC(self, debug=False): ed@448: PLCprint("StartPLC") edouard@453: self.HandleSerialTransaction(STARTTransaction()) ed@448: ed@448: def StopPLC(self): ed@448: PLCprint("StopPLC") edouard@453: self.HandleSerialTransaction(STOPTransaction()) ed@448: ed@448: def ForceReload(self): edouard@453: pass ed@448: ed@448: def GetPLCstatus(self): edouard@508: self.HandleSerialTransaction(IDLETransaction()) edouard@508: return self.PLCStatus ed@448: ed@448: def NewPLC(self, md5sum, data, extrafiles): edouard@521: if os.path.exists(self.StorageConnection): edouard@521: firmwarepath = os.path.join( edouard@521: self.StorageConnection, edouard@521: "firmware.bin") edouard@521: try: edouard@521: if os.path.exists(firmwarepath ): edouard@521: os.unlink(firmwarepath) edouard@521: f = open(firmwarepath, "wb") edouard@521: f.write(data) edouard@521: f.close() edouard@521: return True edouard@521: except LPCError,e: edouard@521: self.StorageConnection = None edouard@521: self.PLCStatus = "Disconnected" edouard@521: self.pluginsroot.logger.write_error( edouard@521: "LPC transfer error : "+ edouard@521: str(e)+"\n") ed@448: ed@448: def MatchMD5(self, MD5): edouard@521: data = self.HandleSerialTransaction(GET_PLCIDTransaction()) edouard@525: print "PLCINFO",data[32:] edouard@521: return data[:32] == MD5 edouard@521: ed@448: class IEC_STRING(ctypes.Structure): ed@448: """ ed@448: Must be changed according to changes in iec_types.h ed@448: """ ed@448: _fields_ = [("len", ctypes.c_uint8), edouard@502: ("body", ctypes.c_char * 126)] ed@448: edouard@502: TypeTranslator = {"BOOL" : (ctypes.c_uint8, lambda x:x.value!=0, lambda t,x:t(x)), edouard@502: "STEP" : (ctypes.c_uint8, lambda x:x.value, lambda t,x:t(x)), edouard@502: "TRANSITION" : (ctypes.c_uint8, lambda x:x.value, lambda t,x:t(x)), edouard@502: "ACTION" : (ctypes.c_uint8, lambda x:x.value, lambda t,x:t(x)), edouard@502: "SINT" : (ctypes.c_int8, lambda x:x.value, lambda t,x:t(x)), edouard@502: "USINT" : (ctypes.c_uint8, lambda x:x.value, lambda t,x:t(x)), edouard@502: "BYTE" : (ctypes.c_uint8, lambda x:x.value, lambda t,x:t(x)), edouard@502: "STRING" : (IEC_STRING, lambda x:x.body[:x.len], lambda t,x:t(len(x),x)), edouard@502: "INT" : (ctypes.c_int16, lambda x:x.value, lambda t,x:t(x)), edouard@502: "UINT" : (ctypes.c_uint16, lambda x:x.value, lambda t,x:t(x)), edouard@502: "WORD" : (ctypes.c_uint16, lambda x:x.value, lambda t,x:t(x)), edouard@502: "WSTRING" : (None, None, None),#TODO edouard@502: "DINT" : (ctypes.c_int32, lambda x:x.value, lambda t,x:t(x)), edouard@502: "UDINT" : (ctypes.c_uint32, lambda x:x.value, lambda t,x:t(x)), edouard@502: "DWORD" : (ctypes.c_uint32, lambda x:x.value, lambda t,x:t(x)), edouard@502: "LINT" : (ctypes.c_int64, lambda x:x.value, lambda t,x:t(x)), edouard@502: "ULINT" : (ctypes.c_uint64, lambda x:x.value, lambda t,x:t(x)), edouard@502: "LWORD" : (ctypes.c_uint64, lambda x:x.value, lambda t,x:t(x)), edouard@502: "REAL" : (ctypes.c_float, lambda x:x.value, lambda t,x:t(x)), edouard@502: "LREAL" : (ctypes.c_double, lambda x:x.value, lambda t,x:t(x)), ed@448: } edouard@502: edouard@502: def SetTraceVariablesList(self, idxs): edouard@502: """ edouard@502: Call ctype imported function to append edouard@502: these indexes to registred variables in PLC debugger edouard@502: """ edouard@502: if idxs: edouard@502: buff = "" edouard@502: # keep a copy of requested idx edouard@502: self._Idxs = idxs[:] edouard@502: for idx,iectype,force in idxs: edouard@502: idxstr = ctypes.string_at( edouard@502: ctypes.pointer( edouard@502: ctypes.c_uint32(length)),4) edouard@502: if force !=None: edouard@502: c_type,unpack_func, pack_func = self.TypeTranslator.get(iectype, (None,None,None)) edouard@502: forcedsizestr = chr(ctypes.sizeof(c_type)) edouard@502: forcestr = ctypes.string_at( edouard@502: ctypes.pointer( edouard@502: pack_func(c_type,force)), edouard@502: forced_type_size) edouard@502: buff += idxstr + forced_type_size_str + forcestr edouard@502: else: edouard@502: buff += idxstr + chr(0) edouard@508: data = self.HandleSerialTransaction( edouard@502: SET_TRACE_VARIABLETransaction(buff)) edouard@502: else: edouard@502: self._Idxs = [] edouard@502: ed@448: def GetTraceVariables(self): ed@448: """ ed@448: Return a list of variables, corresponding to the list of required idx ed@448: """ edouard@508: offset = 0 edouard@508: strbuf = self.HandleSerialTransaction( edouard@508: GET_TRACE_VARIABLETransaction()) edouard@508: size = len(strbuf) - 4 edouard@508: if size > 0 and self.PLCStatus == "Started": edouard@508: tick = ctypes.cast( edouard@508: ctypes.c_char_p(strbuf[:4]), edouard@508: ctypes.POINTER(ctypes.c_int)).contents edouard@508: buffer = ctypes.cast( edouard@508: ctypes.c_char_p(strbuf[4:]), edouard@508: ctypes.c_void_p) edouard@508: for idx, iectype, forced in self._Idxs: edouard@508: cursor = ctypes.c_void_p(buffer.value + offset) edouard@508: c_type,unpack_func, pack_func = self.TypeTranslator.get(iectype, (None,None,None)) edouard@508: if c_type is not None and offset < size: edouard@508: res.append(unpack_func(ctypes.cast(cursor, edouard@508: ctypes.POINTER(c_type)).contents)) edouard@508: offset += ctypes.sizeof(c_type) edouard@508: else: edouard@508: if c_type is None: edouard@508: PLCprint("Debug error - " + iectype + " not supported !") edouard@508: if offset >= size: edouard@508: PLCprint("Debug error - buffer too small !") edouard@508: break edouard@508: if offset and offset == size: edouard@502: return self.PLCStatus, tick.value, res edouard@508: PLCprint("Debug error - wrong buffer unpack !") edouard@453: return self.PLCStatus, None, None ed@448: