gRPC In Python Gebruiken Om Procesdata Op Een AXC F 3152 Te Lezen En Schrijven

Dit artikel beschrijft hoe je eenvoudige procesdata kunt benaderen en schrijven met Python met een AXC F 3152 door gebruik te maken van gRPC. (https://www.plcnext.help/te/Service_Components/gRPC_Introduction.htm)

Vereisten

Eerst moeten we de benodigde bestanden voorbereiden, buiten de PLC, bijv. op een Windows-machine.

1. Installeer Python 3.9 (3.10 kan fouten veroorzaken)
2. Installeer de vereiste Python-pakket om code te genereren uit de .proto-bestanden: pip install grpcio-tools==1.36.1
3. Download en pak de repository met de .proto-bestanden uit van https://github.com/PLCnext/gRPC

_pb2.py En _pb2_grpc.py Genereren Van .proto-bestanden

Vervolgens moeten we de vereiste python-bestanden genereren uit de meegeleverde .proto-bestanden. De laatste bevinden zich in de volgende map: gRPC-master/protobuf.

Gebruik deze code om een Python-script in de map gRPC-master aan te maken, bijv. generate_grpc.py. Het script

1. genereert de vereiste bestanden en plaatst ze in gRPC-master/pxc_grpc
2. past de importpaden aan

 import glob import os from pathlib import Path # create the output directory Path('pxc_grpc').mkdir(parents=True, exist_ok=True) grpc_command_base = 'python -m grpc_tools.protoc -I./protobuf --python_out=pxc_grpc --grpc_python_out=pxc_grpc ' import_paths = set() # generate the *_pb2.py and *_pb2_grpc.py files for filename in glob.iglob('./protobuf/**', recursive=True): if filename.endswith('.proto'): # store the import path path_parts = filename.split(os.sep) import_paths.add('.'.join(path_parts[1:-1])) grpc_command = ''.join([grpc_command_base, os.path.join('.', os.path.relpath(filename))]) stream = os.popen(grpc_command) output = stream.read() if output != '': print(''.join(['error/info for file ', os.path.relpath(filename), ' - ', output])) # get the python files in the base directory base_pys = set() for (dirpath, dirnames, filenames) in os.walk('./pxc_grpc'): for f in filenames: base_pys.add(f.split('.py')[0]) break # reformat the stored paths to adapt the import statements try: import_paths.remove('') except: pass import_paths = list(import_paths) import_paths.sort(key=len) import_paths.reverse() # adapt the imports for filename in glob.iglob('./pxc_grpc/**', recursive=True): if filename.endswith('.py'): new_lines = [] with open(filename, 'r') as file: lines = file.readlines() for line in lines: if line.startswith('from'): for import_path in import_paths: if import_path in line: line = line.replace(import_path, ''.join(['pxc_grpc.', import_path]), 1) break elif line.startswith('import'): parts = line.split() if parts[1] in base_pys: line = line.replace('import', 'from pxc_grpc import') new_lines.append(line) with open(filename, 'w') as file: file.write(''.join(new_lines)) 

Open een shell en voer het script uit: python generate_grpc.py

Een PLCnext Demo Project Aanmaken

Het getoonde project dient alleen ter demonstratie van hoe de gRPC-interface interacteert met de GDS. Voel je vrij om een bestaand project te gebruiken. Voor een individueel project moet je de poortnamen in het volgende Python-script dienovereenkomstig aanpassen, bijv. Arp.Plc.Eclr/MainInstance.strInput.

De PLC Voorbereiden

Installeer pip om je Python-pakketten te beheren:

1. Verbind de AXC F 3152-controller met het internet.
2. Voer de opdracht curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py uit.
3. Voer vervolgens de opdracht python3 get-pip.py uit.

Installeer de vereiste pakketten: pip install grpcio protobuf==3.20.0

Maak een map 'grpc_test' in de projectenmap (/opt/plcnext/projects/).

Kopieer de 'pxc_grpc'-map, met de aangemaakte Python-bestanden, naar 'grpc_test', bijv. met WinSCP.

Maak een Python-script in de 'grpc_test'-map met de naam 'grpc_test.py' en voeg de volgende code in:

 import grpc from pxc_grpc.Plc.Gds.IDataAccessService_pb2 import IDataAccessServiceReadSingleRequest, \ IDataAccessServiceReadRequest, IDataAccessServiceWriteSingleRequest, IDataAccessServiceWriteRequest from pxc_grpc.Plc.Gds.IDataAccessService_pb2_grpc import IDataAccessServiceStub from pxc_grpc.Plc.Gds.WriteItem_pb2 import WriteItem def write_single_string(stub, port_name, value): single_write_request = IDataAccessServiceWriteSingleRequest() single_write_request.data.PortName = port_name single_write_request.data.Value.TypeCode = 19 single_write_request.data.Value.StringValue = value return stub.WriteSingle(single_write_request) def write_single_int(stub, port_name, value): single_write_request = IDataAccessServiceWriteSingleRequest() single_write_request.data.PortName = port_name single_write_request.data.Value.TypeCode = 6 single_write_request.data.Value.Int16Value = value return stub.WriteSingle(single_write_request) def write_multiple_values(stub): write_request = IDataAccessServiceWriteRequest() wi1 = WriteItem() wi1.PortName = 'Arp.Plc.Eclr/MainInstance.strInput' wi1.Value.StringValue = "test1" wi1.Value.TypeCode = 19 wi2 = WriteItem() wi2.PortName = 'Arp.Plc.Eclr/MainInstance.strInput2' wi2.Value.StringValue = "test2" wi2.Value.TypeCode = 19 # add multiple WriteItems at once write_request.data.extend([wi1, wi2]) # add WriteItems separately # response1.data.append(wi1) # response1.data.append(wi2) return stub.Write(write_request) def read_single_value(stub, port_name): single_read_request = IDataAccessServiceReadSingleRequest() single_read_request.portName=port_name return stub.ReadSingle(single_read_request) def read_multiple_values(stub, port_names): read_request = IDataAccessServiceReadRequest() read_request.portNames.extend(port_names) return stub.Read(read_request) if __name__ == "__main__": # create channel and stub channel = grpc.insecure_channel('unix:/run/plcnext/grpc.sock') stub = IDataAccessServiceStub(channel) print(write_single_string(stub, 'Arp.Plc.Eclr/MainInstance.strInput', 'test123')) print(write_single_int(stub, 'Arp.Plc.Eclr/MainInstance.iInput', 18)) print(write_multiple_values(stub)) r = read_single_value(stub, 'Arp.Plc.Eclr/MainInstance.strInput') print(r) print(r._ReturnValue.Value.TypeCode) print(r._ReturnValue.Value.StringValue) r = read_multiple_values(stub, ['Arp.Plc.Eclr/MainInstance.iInput', 'Arp.Plc.Eclr/MainInstance.strInput']) for value in r._ReturnValue: print(value, value.Value.TypeCode) 

Verbind je PLC met PLCnext Engineer, download het project en start de live view.

Het Voorbeeld Uitvoeren

Start nu het voorbeeld. Log in op de PLC via ssh en navigeer naar 'grpc_test', start dan het Python-script:

1. cd projects/grpc_test/
2. python3 grpc_test.py

De gRPC stelt interactie met GDS-variabelen in staat.

Datatypes

Om variabelen te lezen en te schrijven, is het datatype vereist, bijv. wi1.Value.TypeCode = 19. De typen worden beschreven in het gegenereerde bestand gRPC-master/pxc_grpc/ArpTypes_pb2.py vanaf regel 242:

CT_None = 0 CT_End = 0 CT_Void = 1 CT_Boolean = 2 CT_Char = 3 CT_Int8 = 4 CT_Uint8 = 5 CT_Int16 = 6 CT_Uint16 = 7 CT_Int32 = 8 CT_Uint32 = 9 CT_Int64 = 10 CT_Uint64 = 11 CT_Real32 = 12 CT_Real64 = 13 CT_Struct = 18 CT_String = 19 CT_Utf8String = 19 CT_Array = 20 CT_DateTime = 23 CT_Version = 24 CT_Guid = 25 CT_AnsiString = 26 CT_Object = 28 CT_Utf16String = 30 CT_Stream = 34 CT_Enumerator = 35 CT_SecureString = 36 CT_Enum = 37 CT_Dictionary = 38 CT_SecurityToken = 39 CT_Exception = 40 CT_IecTime = 41 CT_IecTime64 = 42 CT_IecDate = 43 CT_IecDate64 = 44 CT_IecDateTime = 45 CT_IecDateTime64 = 46 CT_IecTimeOfDay = 47 CT_IecTimeOfDay64 = 48 

De overeenkomstige waardevariabelen, bijv. r._ReturnValue.Value.StringValue, kunnen in hetzelfde bestand gevonden worden, vanaf regel 365, bijv. BoolValue, Int8Value, StringValue.