lpcmanager

Moved command evaluator code to LPCCommand.py.
revamp
2018-02-05, Edouard Tisserant
2faf6a91e2d8
Moved command evaluator code to LPCCommand.py.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import cmd
from threading import Thread, Semaphore, Lock, Timer
from types import StringType, UnicodeType
import wx
from LPCBus import *
build_lock = Lock()
wx_eval_lock = Semaphore(0)
eval_res = None
# Command log for debug, for viewing from wxInspector
# __builtins__.cmdlog = []
def wx_evaluator(callable, *args, **kwargs):
global eval_res
eval_res = None
try:
eval_res = callable(*args, **kwargs)
finally:
wx_eval_lock.release()
def evaluator(callable, *args, **kwargs):
global eval_res
wx.CallAfter(wx_evaluator, callable, *args, **kwargs)
wx_eval_lock.acquire()
return eval_res
def location(loc):
return tuple(map(int, loc.split(".")))
class LPCCommand(cmd.Cmd):
prompt = ""
RefreshTimer = None
def __init__(self, CTR, Log):
cmd.Cmd.__init__(self, stdin=Log, stdout=Log)
self.use_rawinput = False
self.restore_last_state = False
self.Log = Log
self.CTR = CTR
for function, (arg_types, opt) in {
"Exit": ([], 0),
"Show": ([bool], 1),
"Refresh": ([], 0),
"Close": ([], 0),
"Compile": ([], 0),
"SetProjectProperties": ([str, str, str, str], 0),
"SetOnlineMode": ([str, str], 1),
"AddBus": ([int, str, str], 1),
"RenameBus": ([int, str], 0),
"ChangeBusIECChannel": ([int, int], 0),
"RemoveBus": ([int], 0),
"AddModule": ([location, int, str, str, str], 1),
"RenameModule": ([location, str], 0),
"ChangeModuleIECChannel": ([location, int], 0),
"ChangeModuleInitCode": ([location, str], 0),
"RemoveModule": ([location, int], 0),
"StartGroup": ([location, str, str], 1),
"AddVariable": ([location, location, str, str, str, str, str, str], 1),
"ChangeVariableParams": (
[location, location, str, str, str, str, str, str], 1),
"RemoveVariable": ([location, location], 0)
}.iteritems():
arg_number = len(arg_types)
def CmdFunction(self, line):
args_toks = line.split('"')
if len(args_toks) % 2 == 0:
self.Log.write("Error: Invalid command\n")
sys.stdout.flush()
return
args = []
for num, arg in enumerate(args_toks):
if num % 2 == 0:
stripped = arg.strip()
if stripped:
args.extend(stripped.split(" "))
else:
args.append(arg)
number = None
extra = ""
if opt == 0 and len(args) != arg_number:
number = arg_number
elif len(args) > arg_number:
number = arg_number
extra = " at most"
elif len(args) < arg_number - opt:
number = arg_number - opt
extra = " at least"
if number is not None:
if number == 0:
self.Log.write("Error: No argument%s expected\n" % extra)
elif number == 1:
self.Log.write("Error: 1 argument%s expected\n" % extra)
else:
self.Log.write("Error: %d arguments%s expected\n" % (number, extra))
sys.stdout.flush()
return
for num, arg in enumerate(args):
try:
args[num] = arg_types[num](arg)
except:
self.Log.write("Error: Invalid value for argument %d\n" % (num + 1))
sys.stdout.flush()
return
func = getattr(self, function)
res = evaluator(func, *args)
# cmdlog.append((function, line, res))
# if len(cmdlog) > 100: # prevent debug log to grow too much
# cmdlog.pop(0)
if isinstance(res, (StringType, UnicodeType)):
self.Log.write(res)
return False
else:
return res
setattr(self, "do_%s" % function, CmdFunction)
def RestartTimer(self):
if self.RefreshTimer is not None:
self.RefreshTimer.cancel()
self.RefreshTimer = Timer(0.1, wx.CallAfter, args=[self.Refresh])
self.RefreshTimer.start()
def Exit(self):
global frame, app
self.Close()
app.ExitMainLoop()
return True
def do_EOF(self, line):
return self.Exit()
def Show(self, loading=False):
global frame
if frame is not None:
self.CTR.SetAppFrame(frame, frame.Log)
self.CTR.UpdateMethodsFromPLCStatus()
frame.Show(loading)
frame.Raise()
self.restore_last_state = True
#self.RestartTimer()
def Refresh(self):
global frame
if frame is not None:
if self.restore_last_state:
self.restore_last_state = False
frame.RestoreLastState()
else:
frame.RefreshEditor()
frame._Refresh(TITLE, PROJECTTREE, POUINSTANCEVARIABLESPANEL, FILEMENU, EDITMENU)
frame.RefreshAll()
def Close(self):
global frame
self.CTR.ResetAppFrame(self.Log)
if frame is not None:
frame.Hide()
def Compile(self):
self.CTR._Build()
def SetProjectProperties(self, projectname, productname, productversion, companyname):
new_properties = {
"projectName": projectname,
"productName": productname,
"productVersion": productversion,
"companyName": companyname}
self.CTR.SetProjectProperties(properties=new_properties, buffer=False)
self.RestartTimer()
def SetOnlineMode(self, mode, path=None):
self.CTR.SetOnlineMode(mode, path)
self.RestartTimer()
def AddBus(self, iec_channel, name, icon=None):
for child in self.CTR.IterChildren():
if child.BaseParams.getName() == name:
return "Error: A bus named %s already exists\n" % name
elif child.BaseParams.getIEC_Channel() == iec_channel:
return "Error: A bus with IEC_channel %d already exists\n" % iec_channel
bus = self.CTR.CTNAddChild(name, "LPCBus", iec_channel)
if bus is None:
return "Error: Unable to create bus\n"
bus.SetIcon(icon)
self.RestartTimer()
def RenameBus(self, iec_channel, name):
bus = self.CTR.GetChildByIECLocation((iec_channel,))
if bus is None:
return "Error: No bus found\n"
for child in self.CTR.IterChildren():
if child != bus and child.BaseParams.getName() == name:
return "Error: A bus named %s already exists\n" % name
bus.BaseParams.setName(name)
self.RestartTimer()
def ChangeBusIECChannel(self, old_iec_channel, new_iec_channel):
bus = self.CTR.GetChildByIECLocation((old_iec_channel,))
if bus is None:
return "Error: No bus found\n"
for child in self.CTR.IterChildren():
if child != bus and child.BaseParams.getIEC_Channel() == new_iec_channel:
return "Error: A bus with IEC_channel %d already exists\n" % new_iec_channel
self.CTR.UpdateProjectVariableLocation(str(old_iec_channel),
str(new_iec_channel))
bus.BaseParams.setIEC_Channel(new_iec_channel)
self.RestartTimer()
def RemoveBus(self, iec_channel):
bus = self.CTR.GetChildByIECLocation((iec_channel,))
if bus is None:
return "Error: No bus found\n"
self.CTR.RemoveProjectVariableByFilter(str(iec_channel))
self.CTR.Children["LPCBus"].remove(bus)
self.RestartTimer()
def AddModule(self, parent, iec_channel, name, icode, icon=None):
module = self.CTR.GetChildByIECLocation(parent)
if module is None:
return "Error: No parent found\n"
for child in GetModuleChildren(module):
if child["name"] == name:
return "Error: A module named %s already exists\n" % name
elif child["IEC_Channel"] == iec_channel:
return "Error: A module with IEC_channel %d already exists\n" % iec_channel
GetLastModuleGroup(module).append({"name": name,
"type": LOCATION_MODULE,
"IEC_Channel": iec_channel,
"icon": icon,
"init": icode,
"children": []})
self.RestartTimer()
def RenameModule(self, iec_location, name):
module = self.CTR.GetChildByIECLocation(iec_location)
if module is None:
return "Error: No module found\n"
parent = self.CTR.GetChildByIECLocation(iec_location[:-1])
if parent is self.CTR:
return "Error: No module found\n"
if module["name"] != name:
for child in GetModuleChildren(parent):
if child["name"] == name:
return "Error: A module named %s already exists\n" % name
module["name"] = name
self.RestartTimer()
def ChangeModuleIECChannel(self, old_iec_location, new_iec_channel):
module = self.CTR.GetChildByIECLocation(old_iec_location)
if module is None:
return "Error: No module found\n"
parent = self.CTR.GetChildByIECLocation(old_iec_location[:-1])
if parent is self.CTR:
return "Error: No module found\n"
if module["IEC_Channel"] != new_iec_channel:
for child in GetModuleChildren(parent):
if child["IEC_Channel"] == new_iec_channel:
return "Error: A module with IEC_channel %d already exists\n" % new_iec_channel
self.CTR.UpdateProjectVariableLocation(".".join(map(str, old_iec_location)),
".".join(map(str, old_iec_location[:1] + (new_iec_channel,))))
module["IEC_Channel"] = new_iec_channel
self.RestartTimer()
def ChangeModuleInitCode(self, iec_location, icode):
module = self.CTR.GetChildByIECLocation(iec_location)
if module is None:
return "Error: No module found\n"
module["init"] = icode
def RemoveModule(self, parent, iec_channel):
module = self.CTR.GetChildByIECLocation(parent)
if module is None:
return "Error: No parent found\n"
child = GetModuleBySomething(module, "IEC_Channel", (iec_channel,))
if child is None:
return "Error: No module found\n"
self.CTR.RemoveProjectVariableByFilter(".".join(map(str, parent + (iec_channel,))))
RemoveModuleChild(module, child)
self.RestartTimer()
def StartGroup(self, parent, name, icon=None):
module = self.CTR.GetChildByIECLocation(parent)
if module is None:
return "Error: No parent found\n"
for child in module["children"]:
if child["type"] == LOCATION_GROUP and child["name"] == name:
return "Error: A group named %s already exists\n" % name
module["children"].append({"name": name,
"type": LOCATION_GROUP,
"icon": icon,
"children": []})
self.RestartTimer()
def AddVariable(self, parent, location, name, direction, type, rcode, pcode, description=""):
module = self.CTR.GetChildByIECLocation(parent)
if module is None:
return "Error: No parent found\n"
for child in GetModuleChildren(module):
if child["name"] == name:
return "Error: A variable named %s already exists\n" % name
if child["location"] == location and child["type"] == LOCATION_TYPES[direction]:
return "Error: A variable with location %s already exists\n" % ".".join(map(str, location))
GetLastModuleGroup(module).append({"name": name,
"location": location,
"type": LOCATION_TYPES[direction],
"IEC_type": type,
"description": description,
"retrieve": rcode,
"publish": pcode})
self.RestartTimer()
def ChangeVariableParams(self, parent, location, new_name, new_direction, new_type, new_rcode, new_pcode,
new_description=None):
module = self.CTR.GetChildByIECLocation(parent)
if module is None:
return "Error: No parent found\n"
variable = None
for child in GetModuleChildren(module):
if child["location"] == location and child["type"] == LOCATION_TYPES[new_direction]:
variable = child
elif child["name"] == new_name:
return "Error: A variable named %s already exists\n" % new_name
if variable is None:
return "Error: No variable found\n"
if variable["name"] != new_name:
self.CTR.UpdateProjectVariableName(variable["name"], new_name)
variable["name"] = new_name
variable["type"] = LOCATION_TYPES[new_direction]
variable["IEC_type"] = new_type
variable["retrieve"] = new_rcode
variable["publish"] = new_pcode
if new_description is not None:
variable["description"] = new_description
self.RestartTimer()
def RemoveVariable(self, parent, location, direction):
module = self.CTR.GetChildByIECLocation(parent)
if module is None:
return "Error: No parent found\n"
child = GetModuleVariable(module, location, direction)
if child is None:
return "Error: No variable found\n"
size = LOCATION_SIZES[self.CTR.GetBaseType(child["IEC_type"])]
address = "%" + LOCATION_DIRS[child["type"]] + size + ".".join(map(str, parent + location))
self.CTR.RemoveProjectVariableByAddress(address)
RemoveModuleChild(module, child)
self.RestartTimer()