Usa gRPC en Python para Leer y Escribir Datos de Proceso en un AXC F 3152

Este artículo describe cómo acceder y escribir datos de proceso simples con Python en un AXC F 3152 utilizando gRPC. (https://www.plcnext.help/te/Service_Components/gRPC_Introduction.htm)

Requisitos Previos

Primero tenemos que preparar los archivos requeridos, fuera del PLC, ej., en una máquina Windows.

1. Instala Python 3.9 (3.10 puede causar errores)
2. Instala el paquete Python requerido para generar código desde archivos .proto: pip install grpcio-tools==1.36.1
3. Descarga y descomprime el repositorio conteniendo los archivos .proto desde https://github.com/PLCnext/gRPC

Genera _pb2.py y _pb2_grpc.py desde Archivos .proto

Luego, tenemos que generar los archivos python requeridos desde los archivos .proto proporcionados. Estos últimos están ubicados en la siguiente carpeta: gRPC-master/protobuf.

Usa este código para crear un script Python en la Carpeta gRPC-master, ej., generate_grpc.py. El script

1. genera los archivos requeridos y los coloca en gRPC-master/pxc_grpc
2. adapta las rutas de importación

 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)) 

Abre un shell y ejecuta el script: python generate_grpc.py

Crea un Proyecto Demo PLCnext

El proyecto mostrado solo debería demostrar cómo la interfaz gRPC interactúa con el GDS. Siéntete libre de usar un proyecto existente en su lugar. Para un proyecto individual, tienes que editar los nombres de puerto en el siguiente script Python acorde, ej., Arp.Plc.Eclr/MainInstance.strInput.

Prepara el PLC

Instala pip para gestionar tus paquetes Python:

1. Conecta el controlador AXC F 3152 a Internet.
2. Ejecuta el comando curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py.
3. Luego ejecuta el comando python3 get-pip.py.

Instala los paquetes requeridos: pip install grpcio protobuf==3.20.0

Crea una carpeta 'grpc_test' en el directorio de proyectos (/opt/plcnext/projects/).

Copia la carpeta 'pxc_grpc', conteniendo los archivos Python creados a 'grpc_test', ej., con WinSCP.

Crea un script Python en la carpeta 'grpc_test' llamado 'grpc_test.py' e inserta el siguiente código:

 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) 

Conecta tu PLC a PLCnext Engineer, descarga el proyecto e inicia la vista en vivo.

Ejecuta el Ejemplo

Ahora inicia el ejemplo. Haz login en el PLC vía ssh y navega a 'grpc_test', luego inicia el script Python:

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

El gRPC permite interacción con variables GDS.

Tipos de Datos

Para leer y escribir variables, el tipo de dato es requerido, ej., wi1.Value.TypeCode = 19. Los tipos están descritos en el archivo generado gRPC-master/pxc_grpc/ArpTypes_pb2.py empezando en línea 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 

Las variables de valor correspondientes, ej., r._ReturnValue.Value.StringValue, pueden encontrarse en el mismo archivo, comenzando en línea 365, ej., BoolValue, Int8Value, StringValue.