python_opc_api/jjq_support_1.py

#!/usr/bin/env python3
import redis

from asyncua import Client
from datetime import datetime

opc_server_url = "opc.tcp://192.168.10.133:49320"
redis_ip = '127.0.0.1'


def get_point_arr():
    for i in range(137):
        three_digit_numbers = [f"OPC.电液控.支架集.{num:03d}.立柱压力.前柱压力" for num in range(1, 138)]
    return three_digit_numbers


def get_opc_data():
    ns = 2
    opc_point_arr = get_point_arr()

    # redis
    pool = redis.ConnectionPool(host=redis_ip, password='')
    redis_conn = redis.Redis(connection_pool=pool)

    with Client(url=opc_server_url) as client:
        i = 0
        result_arr = {}
        print(datetime.now().strftime('Start_time:%Y-%m-%d %H:%M:%S.%f'))
        for i in range(len(opc_point_arr)):
            try:
                node = f"ns={ns};s={opc_point_arr[i]}"
                tag = client.get_node(node)
                value = tag.read_value()
                # redis值
                redis_conn.set(str(opc_point_arr[i]), str(value))
                arr_key = node.split("=")
                result_arr[arr_key[2]] = value
            finally:
                continue

        return result_arr


if __name__ == '__main__':
    print(get_opc_data())