AirTerminalDevice/apps/DHU/config_reader.py

from typing import Union
from datetime import datetime

import numpy as np
import pandas as pd

class ConfigReader:
    
    def __init__(self,path):
        self.path  = path
        self.equp  = pd.read_excel(self.path,sheet_name='设备')
        self.point = pd.read_excel(self.path,sheet_name='点位')
        self.app   = pd.read_excel(self.path,sheet_name='程序')
        # self.meta  = pd.read_excel(self.path,sheet_name='元数据')
        # self.meta  = {self.meta.loc[:,'Key'].iat[i]:self.meta.loc[:,'Value'].iat[i] for i in range(len(self.meta))}
    
    @property
    def all_equp_names(self):
        return self.equp.loc[:,'编号'].to_list()
    
    def get_equp_info(self,equp_name,key:str,info_type):
        self.check_equp_is_exist(equp_name)
        equp_info = self.equp.loc[self.equp.loc[:,'编号']==equp_name,'名称':].to_dict(orient='list')
        equp_info = {k:v[0] for k,v in equp_info.items()}
        info_value = equp_info[key]
        info_value = convert_info_type(info_value,info_type)
        return info_value
    
    def get_equp_point(self,equp_name,equp_type,equp_class:Union[str,list]=None) -> dict:
        # 点位类型
        # A: 模型输入
        # B: 模型输出
        self.check_equp_is_exist(equp_name)
        equp_class = self.point.类型.to_list() if equp_class is None else equp_class
        equp_class = [equp_class] if isinstance(equp_class,str) else equp_class
        point_info = self.point.loc[self.point.类型.isin(equp_class)].loc[lambda dt:dt.设备类型.str.contains(equp_type)]
        point_map  = dict(zip(point_info.编号.to_list(),point_info.点位.to_list()))
        
        for point_name,point_id in point_map.items():
            if not isinstance(point_id,str) and np.isnan(point_id):
                raise Exception(f'点位{point_name}的点位ID缺失')
        
        # 根据配置情况更新全局点位
        if equp_name in self.point.columns:
            for point_name in point_map.keys():
                new_point_id = self.point.loc[self.point.编号==point_name,equp_name].iat[0]
                if isinstance(new_point_id,str):
                    point_map[point_name] = new_point_id
        return point_map
    
    def get_app_info(
        self,
        equp_name: str,
        app_type : str,
        key      : str,
        info_type: str = None
    ) -> str:
        value = self.app.loc[lambda dt:(dt.程序类型==app_type) & (dt.项目==key)]
        if value.shape[0] != 1:
            raise Exception(f'程序类型({app_type})下不存在唯一的的项目({key})')
        info_value = value.loc[:,'全局配置'].iat[0]
        # 更新全局配置
        if equp_name in value.columns:
            equp_value = value.loc[:,equp_name].iat[0]
            if isinstance(equp_value,(str,float)) and not np.isnan(equp_value):
                info_value = equp_value
        # 调整数据类型
        info_value = convert_info_type(info_value,info_type)
        
        return info_value
        
    def check_equp_is_exist(self,equp_name):
        if equp_name not in self.all_equp_names:
            raise Exception(f"设备{equp_name}不存在")


def convert_info_type(info_value,info_type):
    if info_type == 'int':
        info_value = int(info_value)
    elif info_type == 'float':
        info_value = float(info_value)
    elif info_type == 'bool':
        info_value = bool(info_value)
    elif info_type == 'datetime':
        if info_value == 'NOW':
            info_value = datetime.now()
        elif not isinstance(info_value,datetime):
            info_value = datetime.strptime(info_value,'%Y/%m/%d %H:%M:%S')
    elif info_type == 'str':
        info_value = str(info_value)
    elif info_type is None:
        pass
    else:
        raise ValueError(f"{info_type} type error!")
    return info_value