AirTerminalDevice/tools/config_reader.py

from typing import Union

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='点位')
    
    @property
    def all_equp_names(self):
        return self.equp.loc[:,'编号'].to_list()
    
    def get_equp_info(self,equp_name) -> dict:
        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()}
        return equp_info
    
    def get_equp_point(self,equp_name,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)]
        point_map  = dict(zip(point_info.编号.to_list(),point_info.点位.to_list()))
        # 根据配置情况更新全局点位
        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 check_equp_is_exist(self,equp_name):
        if equp_name not in self.all_equp_names:
            raise Exception(f"设备{equp_name}不存在")