用Python通过gRPC在AXC F 3152上读写过程数据
本文描述如何利用gRPC在AXC F 3152上用Python访问和写入简单过程数据。(https://www.plcnext.help/te/Service_Components/gRPC_Introduction.htm)
前置条件
首先在PLC外部（如Windows机器）准备所需文件。
- 安装Python 3.9（3.10可能报错）
- 安装从.proto文件生成代码所需Python包：
pip install grpcio-tools==1.36.1 - 从https://github.com/PLCnext/gRPC下载并解压含.proto文件的仓库
从.proto文件生成_pb2.py和_pb2_grpc.py
接下来从提供的.proto文件生成所需Python文件。后者位于文件夹：gRPC-master/protobuf。
在gRPC-master文件夹创建Python脚本generate_grpc.py，脚本：
- 生成所需文件并放入gRPC-master/pxc_grpc
- 调整导入路径
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)) 打开shell执行脚本：pyton generate_grpc.py
创建PLCnext演示项目
所示项目仅演示gRPC接口如何与GDS交互。可随意用现有项目代替。对于个别项目，需按以下Python脚本相应编辑端口名，如Arp.Plc.Eclr/MainInstance.strInput。
准备PLC
安装pip管理Python包：
- 将AXC F 3152控制器连接互联网。
- 执行命令
curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py。 - 再执行
python3 get-pip.py。
安装所需包：pip install grpcio protobuf==3.20.0
在项目目录(/opt/plcnext/projects/)创建'grpc_test'文件夹。
将含生成Python文件的'pxc_grpc'文件夹复制到'grpc_test'，如用WinSCP。
在'grpc_test'文件夹创建Python脚本'grpc_test.py'并插入以下代码：
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) 将PLC连接PLCnext Engineer，下载项目并启动实时视图。
运行示例
现在运行示例。通过ssh登录PLC，导航到'grpc_test'，启动Python脚本：
cd projects/grpc_test/python3 grpc_test.py
gRPC实现与GDS变量交互。
数据类型
读写变量需数据类型，如wi1.Value.TypeCode = 19。类型在生成文件gRPC-master/pxc_grpc/ArpTypes_pb2.py第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 对应值变量，如r._ReturnValue.Value.StringValue，在同一文件第365行起，如BoolValue、Int8Value、StringValue。
工业技术