edouard@3337: # opcua/client.py edouard@3337: kinsamanka@3750: edouard@3337: edouard@3337: import os edouard@3337: edouard@3337: from editors.ConfTreeNodeEditor import ConfTreeNodeEditor edouard@3339: from PLCControler import LOCATION_CONFNODE, LOCATION_VAR_INPUT, LOCATION_VAR_OUTPUT edouard@3589: from .opcua_client_maker import OPCUAClientPanel, OPCUAClientModel, UA_IEC_types, authParams edouard@3337: edouard@3337: import util.paths as paths edouard@3337: edouard@3337: # Paths to open62541 assume that edouard@3337: # - open62541 directory is aside beremiz directory edouard@3337: # - open62541 was just built (not installed) edouard@3337: edouard@3337: Open62541Path = paths.ThirdPartyPath("open62541") edouard@3337: Open62541LibraryPath = os.path.join(Open62541Path,"build","bin") edouard@3337: Open62541IncludePaths = [os.path.join(Open62541Path, *dirs) for dirs in [ edouard@3337: ("plugins","include"), edouard@3337: ("build","src_generated"), edouard@3337: ("include",), edouard@3337: ("arch",)]] edouard@3337: edouard@3718: # Tests need to use other default hosts edouard@3718: OPCUA_DEFAULT_HOST = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1") edouard@3718: edouard@3337: class OPCUAClientEditor(ConfTreeNodeEditor): edouard@3337: CONFNODEEDITOR_TABS = [ edouard@3337: (_("OPC-UA Client"), "CreateOPCUAClient_UI")] edouard@3337: edouard@3337: def Log(self, msg): edouard@3337: self.Controler.GetCTRoot().logger.write(msg) edouard@3337: edouard@3337: def CreateOPCUAClient_UI(self, parent): edouard@3589: return OPCUAClientPanel(parent, self.Controler.GetModelData(), self.Log, self.Controler.GetConfig) edouard@3337: edouard@3337: class OPCUAClient(object): edouard@3337: XSD = """ edouard@3337: edouard@3337: edouard@3337: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: Default to Basic256Sha256 if not specified edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3589: edouard@3718: edouard@3337: edouard@3337: edouard@3337: edouard@3337: """ edouard@3337: edouard@3337: EditorType = OPCUAClientEditor edouard@3337: edouard@3337: def __init__(self): edouard@3674: self.modeldata = OPCUAClientModel(self.Log, self.CTNMarkModified) edouard@3337: edouard@3337: filepath = self.GetFileName() edouard@3337: if os.path.isfile(filepath): edouard@3337: self.modeldata.LoadCSV(filepath) edouard@3337: edouard@3378: def Log(self, msg): edouard@3378: self.GetCTRoot().logger.write(msg) edouard@3378: edouard@3337: def GetModelData(self): edouard@3337: return self.modeldata edouard@3674: edouard@3589: def GetConfig(self): edouard@3678: def cfg(path): edouard@3678: try: edouard@3678: attr=self.GetParamsAttributes("OPCUAClient."+path) edouard@3678: except ValueError: edouard@3678: return None edouard@3678: return attr["value"] edouard@3678: edouard@3589: AuthType = cfg("AuthType") edouard@3589: res = dict(URI=cfg("Server_URI"), AuthType=AuthType) edouard@3589: edouard@3589: paramList = authParams.get(AuthType, None) edouard@3589: if paramList: edouard@3589: for name,default in paramList: edouard@3591: value = cfg("AuthType."+name) edouard@3591: if value == "" or value is None: edouard@3591: value = default edouard@3591: # cryptomaterial is expected to be in project's user provide file directory edouard@3591: if name in ["Certificate","PrivateKey"]: edouard@3591: value = os.path.join(self.GetCTRoot()._getProjectFilesPath(), value) edouard@3591: res[name] = value edouard@3589: edouard@3589: return res edouard@3337: edouard@3337: def GetFileName(self): edouard@3337: return os.path.join(self.CTNPath(), 'selected.csv') edouard@3337: edouard@3337: def OnCTNSave(self, from_project_path=None): edouard@3337: self.modeldata.SaveCSV(self.GetFileName()) edouard@3337: return True edouard@3337: edouard@3337: def CTNGenerate_C(self, buildpath, locations): edouard@3337: current_location = self.GetCurrentLocation() edouard@3337: locstr = "_".join(map(str, current_location)) edouard@3337: c_path = os.path.join(buildpath, "opcua_client__%s.c" % locstr) edouard@3337: edouard@3612: c_code = '#include "beremiz.h"\n' edouard@3612: c_code += self.modeldata.GenerateC(c_path, locstr, self.GetConfig()) edouard@3337: edouard@3820: with open(c_path, 'w') as c_file: edouard@3337: c_file.write(c_code) edouard@3337: edouard@3611: LDFLAGS = ['"' + os.path.join(Open62541LibraryPath, "libopen62541.a") + '"', '-lcrypto'] edouard@3337: edouard@3337: CFLAGS = ' '.join(['-I"' + path + '"' for path in Open62541IncludePaths]) edouard@3337: edouard@3591: # Note: all user provided files are systematicaly copied, including cryptomaterial edouard@3591: edouard@3337: return [(c_path, CFLAGS)], LDFLAGS, True edouard@3337: edouard@3339: def GetVariableLocationTree(self): edouard@3339: current_location = self.GetCurrentLocation() edouard@3339: locstr = "_".join(map(str, current_location)) edouard@3339: name = self.BaseParams.getName() edouard@3339: entries = [] kinsamanka@3750: for direction, data in self.modeldata.items(): edouard@3339: iec_direction_prefix = {"input": "__I", "output": "__Q"}[direction] edouard@3339: for row in data: edouard@3339: dname, ua_nsidx, ua_nodeid_type, _ua_node_id, ua_type, iec_number = row edouard@3339: iec_type, C_type, iec_size_prefix, ua_type_enum, ua_type = UA_IEC_types[ua_type] edouard@3339: c_loc_name = iec_direction_prefix + iec_size_prefix + locstr + "_" + str(iec_number) edouard@3339: entries.append({ edouard@3339: "name": dname, edouard@3339: "type": {"input": LOCATION_VAR_INPUT, "output": LOCATION_VAR_OUTPUT}[direction], edouard@3339: "size": {"X":1, "B":8, "W":16, "D":32, "L":64}[iec_size_prefix], edouard@3339: "IEC_type": iec_type, edouard@3339: "var_name": c_loc_name, edouard@3339: "location": iec_size_prefix + ".".join([str(i) for i in current_location]) + "." + str(iec_number), edouard@3339: "description": "", edouard@3339: "children": []}) edouard@3339: return {"name": name, edouard@3339: "type": LOCATION_CONFNODE, edouard@3339: "location": ".".join([str(i) for i in current_location]) + ".x", edouard@3339: "children": entries} edouard@3339: