gRPC Python – 프로세스 데이터 읽기 및 쓰기

이 문서에서는 gRPC를 사용하는 AXC F 3152를 사용하여 Python으로 간단한 프로세스 데이터에 액세스하고 쓰는 방법을 설명합니다. (https://www.plcnext.help/te/Service_Components/gRPC_Introduction.htm)

전제조건

먼저 PLC 외부(예:Windows 시스템)에서 필요한 파일을 준비해야 합니다.

1. Python 3.9 설치(3.10에서는 오류가 발생할 수 있음)
2. .proto 파일에서 코드를 생성하는 데 필요한 Python 패키지 설치:pip install grpcio-tools==1.36.1
3. https://github.com/PLCnext/gRPC에서 .proto 파일이 포함된 저장소를 다운로드하고 압축을 풉니다.

.proto 파일에서 _pb2.py 및 _pb2_grpc.py 생성

다음으로 제공된 .proto 파일에서 필요한 Python 파일을 생성해야 합니다. 후자는 gRPC-master/protobuf 폴더에 있습니다.

이 코드를 사용하여 폴더 gRPC-master에 Python 스크립트를 생성합니다(예:generate_grpc.py). 스크립트

1. 필요한 파일을 생성하여 gRPC-master/pxc_grpc에 배치
2. 가져오기 경로 조정

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

쉘을 열고 스크립트를 실행하십시오:pyton generate_grpc.py

PLCnext 데모 프로젝트 생성

표시된 프로젝트는 gRPC 인터페이스가 GDS와 상호 작용하는 방법만 보여야 합니다. 대신 기존 프로젝트를 자유롭게 사용하십시오. 개별 프로젝트의 경우 다음 Python 스크립트에서 포트 이름을 적절하게 편집해야 합니다(예:Arp.Plc.Eclr/MainInstance.strInput). .

PLC 준비

Python 패키지를 관리하려면 pip를 설치하세요.

1. AXC F 3152 컨트롤러를 인터넷에 연결합니다.
2. 컬 https://bootstrap.pypa.io/get-pip.py -o get-pip.py 명령을 입력합니다.
3. 그런 다음 python3 get-pip.py 명령을 입력합니다.

필요한 패키지 설치:pip install grpcio protobuf==3.20.0

프로젝트 디렉토리(/opt/plcnext/projects/)에 'grpc_test' 폴더를 생성합니다.

크레이트된 Python 파일이 포함된 'pxc_grpc' 폴더를 'grpc_test'(예:WinSCP 사용)에 복사합니다.

'grpc_test.py'라는 이름의 'grpc_test' 폴더에 Python 스크립트를 만들고 다음 코드를 삽입합니다.

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 스크립트를 시작합니다.

1. cd projects/grpc_test/
2. 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 .