1

# opcua/client.py

     2

     3

     4

     5

import os

     6

     7

from editors.ConfTreeNodeEditor import ConfTreeNodeEditor

     8

from PLCControler import LOCATION_CONFNODE, LOCATION_VAR_INPUT, LOCATION_VAR_OUTPUT

     9

from .opcua_client_maker import OPCUAClientPanel, OPCUAClientModel, UA_IEC_types, authParams

    10

    11

import util.paths as paths

    12

    13

# Paths to open62541 assume that 

    14

# - open62541 directory is aside beremiz directory

    15

# - open62541 was just built (not installed)

    16

    17

Open62541Path = paths.ThirdPartyPath("open62541")

    18

Open62541LibraryPath = os.path.join(Open62541Path,"build","bin") 

    19

Open62541IncludePaths = [os.path.join(Open62541Path, *dirs) for dirs in [

    20

    ("plugins","include"),

    21

    ("build","src_generated"),

    22

    ("include",),

    23

    ("arch",)]]

    24

    25

# Tests need to use other default hosts

    26

OPCUA_DEFAULT_HOST = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1")

    27

    28

class OPCUAClientEditor(ConfTreeNodeEditor):

    29

    CONFNODEEDITOR_TABS = [

    30

        (_("OPC-UA Client"), "CreateOPCUAClient_UI")]

    31

    32

    def Log(self, msg):

    33

        self.Controler.GetCTRoot().logger.write(msg)

    34

    35

    def CreateOPCUAClient_UI(self, parent):

    36

        return OPCUAClientPanel(parent, self.Controler.GetModelData(), self.Log, self.Controler.GetConfig)

    37

    38

class OPCUAClient(object):

    39

    XSD = """<?xml version="1.0" encoding="ISO-8859-1" ?>

    40

    <xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">

    41

      <xsd:element name="OPCUAClient">

    42

        <xsd:complexType>

    43

          <xsd:sequence>

    44

            <xsd:element name="AuthType" minOccurs="0">

    45

              <xsd:complexType>

    46

                <xsd:choice minOccurs="0">

    47

                  <xsd:element name="x509">

    48

                    <xsd:complexType>

    49

                      <xsd:sequence>

    50

                        <xsd:element name="Policy">

    51

                          <xsd:annotation>

    52

                            <xsd:documentation>Default to Basic256Sha256 if not specified</xsd:documentation>

    53

                          </xsd:annotation>

    54

                          <xsd:complexType>

    55

                            <xsd:choice minOccurs="0">

    56

                              <xsd:element name="Basic256Sha256"/>

    57

                              <xsd:element name="Basic128Rsa15"/>

    58

                              <xsd:element name="Basic256"/>

    59

                            </xsd:choice>

    60

                          </xsd:complexType>

    61

                        </xsd:element>

    62

                        <xsd:element name="Mode">

    63

                          <xsd:complexType>

    64

                            <xsd:choice minOccurs="0">

    65

                              <xsd:element name="SignAndEncrypt"/>

    66

                              <xsd:element name="Sign"/>

    67

                            </xsd:choice>

    68

                          </xsd:complexType>

    69

                        </xsd:element>

    70

                      </xsd:sequence>

    71

                      <xsd:attribute name="Certificate" type="xsd:string" use="optional" default="certificate.pem"/>

    72

                      <xsd:attribute name="PrivateKey" type="xsd:string" use="optional" default="private_key.pem"/>

    73

                    </xsd:complexType>

    74

                  </xsd:element>

    75

                  <xsd:element name="UserPassword">

    76

                    <xsd:complexType>

    77

                      <xsd:attribute name="User" type="xsd:string" use="optional"/>

    78

                      <xsd:attribute name="Password" type="xsd:string" use="optional"/>

    79

                    </xsd:complexType>

    80

                  </xsd:element>

    81

                </xsd:choice>

    82

              </xsd:complexType>

    83

            </xsd:element>

    84

          </xsd:sequence>

    85

          <xsd:attribute name="Server_URI" type="xsd:string" use="optional" default="opc.tcp://"""+OPCUA_DEFAULT_HOST+""":4840"/>

    86

        </xsd:complexType>

    87

      </xsd:element>

    88

    </xsd:schema>

    89

    """

    90

    91

    EditorType = OPCUAClientEditor

    92

    93

    def __init__(self):

    94

        self.modeldata = OPCUAClientModel(self.Log, self.CTNMarkModified)

    95

    96

        filepath = self.GetFileName()

    97

        if os.path.isfile(filepath):

    98

            self.modeldata.LoadCSV(filepath)

    99

   100

    def Log(self, msg):

   101

        self.GetCTRoot().logger.write(msg)

   102

   103

    def GetModelData(self):

   104

        return self.modeldata

   105

   106

    def GetConfig(self):

   107

        def cfg(path): 

   108

            try:

   109

                attr=self.GetParamsAttributes("OPCUAClient."+path)

   110

            except ValueError:

   111

                return None

   112

            return attr["value"]

   113

   114

        AuthType = cfg("AuthType")

   115

        res = dict(URI=cfg("Server_URI"), AuthType=AuthType)

   116

   117

        paramList = authParams.get(AuthType, None)

   118

        if paramList:

   119

            for name,default in paramList:

   120

                value = cfg("AuthType."+name)

   121

                if value == "" or value is None:

   122

                    value = default

   123

                # cryptomaterial is expected to be in project's user provide file directory

   124

                if name in ["Certificate","PrivateKey"]:

   125

                    value = os.path.join(self.GetCTRoot()._getProjectFilesPath(), value)

   126

                res[name] = value

   127

   128

        return res

   129

   130

    def GetFileName(self):

   131

        return os.path.join(self.CTNPath(), 'selected.csv')

   132

   133

    def OnCTNSave(self, from_project_path=None):

   134

        self.modeldata.SaveCSV(self.GetFileName())

   135

        return True

   136

   137

    def CTNGenerate_C(self, buildpath, locations):

   138

        current_location = self.GetCurrentLocation()

   139

        locstr = "_".join(map(str, current_location))

   140

        c_path = os.path.join(buildpath, "opcua_client__%s.c" % locstr)

   141

   142

        c_code = '#include "beremiz.h"\n'

   143

        c_code += self.modeldata.GenerateC(c_path, locstr, self.GetConfig())

   144

   145

        with open(c_path, 'w') as c_file:

   146

            c_file.write(c_code)

   147

   148

        LDFLAGS = ['"' + os.path.join(Open62541LibraryPath, "libopen62541.a") + '"', '-lcrypto']

   149

   150

        CFLAGS = ' '.join(['-I"' + path + '"' for path in Open62541IncludePaths])

   151

   152

        # Note: all user provided files are systematicaly copied, including cryptomaterial

   153

   154

        return [(c_path, CFLAGS)], LDFLAGS, True

   155

   156

    def GetVariableLocationTree(self):

   157

        current_location = self.GetCurrentLocation()

   158

        locstr = "_".join(map(str, current_location))

   159

        name = self.BaseParams.getName()

   160

        entries = []

   161

        for direction, data in self.modeldata.items():

   162

            iec_direction_prefix = {"input": "__I", "output": "__Q"}[direction]

   163

            for row in data:

   164

                dname, ua_nsidx, ua_nodeid_type, _ua_node_id, ua_type, iec_number = row

   165

                iec_type, C_type, iec_size_prefix, ua_type_enum, ua_type = UA_IEC_types[ua_type]

   166

                c_loc_name = iec_direction_prefix + iec_size_prefix + locstr + "_" + str(iec_number)

   167

                entries.append({

   168

                    "name": dname,

   169

                    "type": {"input": LOCATION_VAR_INPUT, "output": LOCATION_VAR_OUTPUT}[direction],

   170

                    "size": {"X":1, "B":8, "W":16, "D":32, "L":64}[iec_size_prefix],

   171

                    "IEC_type": iec_type,

   172

                    "var_name": c_loc_name,

   173

                    "location": iec_size_prefix + ".".join([str(i) for i in current_location]) + "." + str(iec_number),

   174

                    "description": "",

   175

                    "children": []})

   176

        return {"name": name,

   177

                "type": LOCATION_CONFNODE,

   178

                "location": ".".join([str(i) for i in current_location]) + ".x",

   179

                "children": entries}

   180