python_opc_api/main.py

import os
import re
import json
import redis
import logging
import asyncio

from flask import Flask, request, render_template
from asyncua import Client
from datetime import datetime


class SubHandler(object):
    r = None
    pool = None
    config = None

    def __init__(self):
        try:
            self.config = get_conf()
            redis_conf = self.config['redis_conf']
            ip = redis_conf['ip']
            self.pool = redis.ConnectionPool(host=ip, password='')
            self.r = redis.Redis(connection_pool=self.pool)
        except Exception as e:
            raise Exception("redis连接错误！")

    async def datachange_notification(self, node, val, data):
        print("Python: New data change event", node, val)

    def event_notification(self, event):
        print("Python: New event", event)


app = Flask(__name__)


# route
@app.route("/get/", methods=['GET', 'POST'])
def get():
    # 矿名key
    sys_key = request.values.get('sys_key')
    # 自动化系统key
    sys_name = request.values.get('sys_name')
    # 参数判断
    if sys_key is None or len(sys_key) == 0 or sys_name is None or len(sys_name) == 0:
        return "参数错误"
    # 获取对应配置
    base_conf = get_conf()
    # 获取opc服务地址
    sys_url = base_conf['sys_conf'][sys_key][sys_name]['opc_server_url']

    # 获取配置文件名
    sys_file_path = base_conf['sys_conf'][sys_key][sys_name]['path']
    sys_file_name = base_conf['sys_conf'][sys_key][sys_name]['file_name']
    # 获取点位数组
    point_conf = get_conf(sys_file_name, sys_file_path)
    try:
        # 取到配置
        point_base_dic = point_conf['sys_point']
        point_arr = []
        # 取点位数据
        # print(point_base_dic)
        for group in point_base_dic:
            for item in point_base_dic[group]:
                point_arr.append(item['key'])
    except Exception as e:
        print ("点位获取错误")
        raise Exception("点位获取错误！")
        
    # 根据服务地址获取Opc数据
    data = asyncio.run(get_opc_data(sys_url, point_arr, 2))
    # 显示请求地址IP
    # get_request_ip()
    
    for group in point_base_dic:
        for item in point_base_dic[group]:
            try:
                key = item['key']
                point_val = data[key]
                item['val'] = point_val
            finally:
                continue

    return point_base_dic


def get_conf(file_name="config.json", file_path="./config"):
    path = f"{file_path}/{file_name}"
    with open(path, "r", encoding="utf-8") as f:
        content = json.load(f)
        f.close()
        return content


async def get_opc_data(sys_url, point_arr, ns=2):
    if sys_url is None or point_arr is None:
        return
    url = sys_url
    async with Client(url=url) as client:
        handler = SubHandler()
        r = handler.r
        i = 0
        result_arr = {}
        print(datetime.now().strftime('Start_time:%Y-%m-%d %H:%M:%S.%f'))
        while i < len(point_arr):
            node = f"ns={ns};s={point_arr[i]}"
            tag = client.get_node(node)
            try:
                sub = await client.create_subscription(500, handler)
                print ("没毛病")
            except Exception as e:
                print(e.args)
                print ("设置监听错误")
                raise Exception("设置监听错误！")
            value = None
            print("获取value")
            try:
                value = await tag.read_value()
                print(value)
                # redis值
                # r.set(str(node), str(value))
                # 获取点位key
                arr_key = node.split("=")
                result_arr[arr_key[2]] = value
                # print(f"tag1 is: {tag} with value {value} ")
                i += 1
                if i == len(point_arr):
                    print("Get Data Success")
            finally:
                print(f"node={node} 点位值获取错误")
                i += 1
                continue
        print(datetime.now().strftime('End_time:%Y-%m-%d %H:%M:%S.%f'))
        return result_arr


def get_request_ip():
    ip = request.remote_addr
    logging.debug(ip)


def api_get_error(msg):
    return '{"code":-1, "msg":"' + msg + '"}'


if __name__ == '__main__':
    config = get_conf()
    port = config['server_port']
    app.run(
        host='0.0.0.0',
        port=config['server_port']
    )