zlwl-algos-new/USER/wangliming/task/offline_algo/test.py

from pywebio.input import *
from pywebio.output import *
from pywebio.session import *
import pywebio.pin as pin
from pywebio import start_server
import pandas as pd
from io import StringIO

from utils import judge_field, image

# field: v:[字段名， 单位， 数据类型， 格式, 默认空值填充规则， 默认无效值填充规则]
fields = {
            '设备码': ['sn', ['-'], '-', '-', '不处理', '不处理'], 
            '时间': ['time', ['-'], '-', ['2023-01-01 00:00:00', '-'],'删除整行', '删除整行'], 
            '电池包电流': ['pack_crnt', ['A', 'mA'], '-', '-', '向上临近填充', '向上临近填充'], 
            '电池包电压': ['pack_volt', ['V', 'mV'], '-', '-', '向上临近填充', '向上临近填充'], 
            '电池包soc': ['soc', ['%', '-'], '-', '-', '向上临近填充', '向上临近填充'],
            '电池包soh': ['soh', ['%', '-'], '-', '-', '向上临近填充', '向上临近填充'], 
            '单体电压列表': ['cell_voltage', ['V', 'mV'], '-', ["*,*,*,...", "[*,*,*,...]"], '向上临近填充', '向上临近填充'], 
            '单体温度列表': ['cell_temp', ['℃'], '-', ["*,*,*,...", "[*,*,*,...]"], '向上临近填充', '向上临近填充'],
            '绝缘电阻': ['ins_resis', ['Ω', 'kΩ'], '-', '-', '向上临近填充', '向上临近填充'], 
            # '速度': ['speed', ['m/s', 'Km/h'], '-', '-', '向上临近填充', '向上临近填充'], 
            # '海拔': ['altitude', ['Km', 'm'], '-', '-', '向上临近填充', '向上临近填充'],
            # '经度': ['longitude', ['-'], '-', '-', '向上临近填充', '向上临近填充'], 
            # '纬度': ['latitude', ['-'] ,'-', '-', '向上临近填充', '向上临近填充'], 
            # '累积行驶里程': ['mileage', ['Km', 'm'], '-', '-', '向上临近填充', '向上临近填充']
        }

def dataclean(a):
    print("数据清洗")
def main():
        set_env(title='神目电池安全卫士', output_animation=False)
    # while True:
        # 上传文件要求: 需要有表头，单体信息格式为 “**，**，**”
        file = file_upload(accept=['.csv', '.xlsx','.xls'], placeholder="请上传文件，支持 .csv, .xlsx, .xls 文件")
        # filename = '001PB047000001B230100075_2101TS_20230328171402_20230328193302.csv'
        # df = pd.read_csv(filename)
        filename = file['filename']
        put_markdown(f"## 文件读取成功，展示前5条数据如下")
        if filename.endswith(".csv"):
            df = pd.read_csv(StringIO(file['content'].decode('utf-8')))
        elif filename.endswith(".xlsx"):
            df = pd.read_excel(file['content'])
        elif filename.endswith(".xls"):
            df = pd.read_excel(file['content'])
        with use_scope('show_data'):
            # 展示文件的数据
            columns = list(df.columns)
            put_table(tdata=df.to_dict(orient='records')[0:5],header=columns)
        
        columns.append('-')
        # 判断列映射
        column_map_res = judge_field.judge_by_field_name(columns)
        with use_scope('set_field'):
            # 对文件的列进行选择
            put_markdown(f"## 字段自动化识别结果如下，可人工进行修改")
            
            header = ['列名','字段','单位','格式']
            values = []
            i = 0
            for k,v in fields.items():
                values.append({
                    '列名': k,
                    '字段': pin.put_select(name=f"field_{i}", options=columns, value=column_map_res.get(k, '-')),
                    '单位': pin.put_select(name=f"unit_{i}", options=v[1]),
                    '格式': pin.put_select(name=f"format_{i}", options=v[3]),
                })
                i = i + 1
            put_table(values,header=header)
        
        # 数据清洗
        put_markdown(f"## 是否进行数据清洗？") 
        pin.put_radio(name='dataclean', options=['是', '否', '默认'], value='默认', label='')

        do_dataclean = pin.pin_wait_change('dataclean')['value']
        if do_dataclean == '是':
            with use_scope('data_claen'):
                # 对文件的列进行选择
                put_markdown(f"## 数据清洗规则")
                header = ['列名', '空值处理', '空值指定值', '无效值处理', '无效值判定值', '无效值指定值']
                clean_rule = ['不处理', '删除整行', '向上临近填充', '向下临近填充', '中位数填充', '平均值填充', '指定值填充']
                values = []
                i = 0
                for k,v in fields.items():
                    values.append({
                        '列名': k,
                        '空值处理': pin.put_select(name=f"none_{i}", options=clean_rule, value=v[4]),
                        '空值指定值': pin.put_input(name=f"none_assign_{i}", type='float'),
                        '无效值处理': pin.put_select(name=f"invalid_{i}", options=clean_rule, value=v[5]),
                        '无效值判定值': pin.put_input(name=f"invalid_judge_{i}", type='float'),
                        '无效值指定值': pin.put_input(name=f"invalie_assign_{i}", type='float'),
                    })
                    i = i + 1
                put_table(values,header=header)
            
            put_button("执行数据清洗", onclick=lambda:dataclean(df))
                
            put_markdown(f"## 导出清洗后的数据")
            df.to_csv("after_clean.csv")
            with open("./after_clean.csv", 'rb') as f:
                content = f.read()
            put_file('data.csv', content, '下载清洗后的数据文件')

            
        with use_scope('algo_invoke'):
            # 对文件的列进行选择
            put_markdown(f"## 算法功能模块")
            # put_markdown(f"**1. 基本信息统计模块**")
            put_button("执行所有算法模块", onclick=lambda:dataclean(df))
            header = ['数据包内电池包数', '数据起始时间', '数据结束时间', '数据时长（天）', '电池包最高活跃度', '电池包平均活跃度', 
                      '总数据量', '日最大数据量']
            values = [1, "2023/3/28 17:14:00", "2023/3/28 19:32:00", 1, 100, 100, 835, 835]
            put_collapse("1. 基本信息统计模块",content=[
                put_button("执行本算法模块", onclick=lambda:dataclean(df)),
                put_table([values],header=header)])
            
            # put_markdown(f"**2. 电池信息分析模块**")
            header = ['SN', '电池类型', '电池包电芯串联数', '温度采样个数']
            values = ["001PB047000001B230100075", "三元电池", 112, 12]
            put_collapse("2. 电池信息分析模块", content=[
                put_button("执行本算法模块", onclick=lambda:dataclean(df)),
                put_table([values],header=header)])
            
            # put_markdown(f"**3. 电池安全诊断模块**")
            header = ['诊断指标', '数值', '诊断结果', '推荐数值']
            values = ['最高电压', 3.68, '⬆', 3.65]
            put_collapse("3. 电池安全诊断模块", content=[
                            put_button("执行本算法模块", onclick=lambda:dataclean(df)),
                            put_table([values],header=header),
                            put_text('''
xxxxxxxx电池于xxxx年xx月xx日发生电芯过压故障 ，故障电芯为xx号电芯,故障发生时最高电芯电压为xxV
xxxxxxxx电池于xxxx年xx月xx日发生绝缘离群故障 ，故障发生时绝缘值为xxxxx

电池是否存在热失控特征
电池是否存在析锂特征
电池是否存在内短路特征''')]
                        )
            
            # put_markdown(f"**4. 电池性能诊断模块**")
            header = ['SOC一致性', '内阻一致性', 'SOH一致性', '温度一致性']
            values = ["0.9", 0.9, 0.9, 0.9]
            
            put_collapse("4. 电池性能诊断模块", content=[
                put_button("执行本算法模块", onclick=lambda:dataclean(df)),
                put_markdown(f"**4.1 SOH衰减信息**"),
                put_text('''***'''),
                put_markdown(f"**4.2 一致性信息**"),
                put_table([values],header=header),
                put_markdown(f"**4.3 一致性绘图**"),
                put_html(image.plot_example().render_notebook()),
            ])
           
            header = ['日均充电次数', '累计充电安时量', '累计充电次数']
            values = [1, 50, 3]
            header2 = ['日均放电次数', '累计放电安时量', '累计放电次数']
            values2 = [3, 30, 10]
            header3 = ['存储soc均值', '存储温度均值', '最大存储时长']
            values3 =  [60, 23, 8]
            put_collapse("5. 电池画像模块", content=[
                put_button("执行本算法模块", onclick=lambda:dataclean(df)),
                put_markdown(f"**5.1 电池充电习惯**"),
                put_table([values],header=header),
                put_markdown("**充电时间段分布图**"),
                put_html(image.plot_example2().render_notebook()),
                put_markdown("**充电起始soc分布图**"),
                put_markdown("**充电结束soc分布图**"),
                
                put_markdown(f"**5.2 电池放电习惯**"),
                put_table([values2],header=header2),
                put_markdown("**放电时间段分布图**"),
                put_markdown("**放电起始soc分布图**"),
                put_markdown("**放电结束soc分布图**"),
                
                put_markdown(f"**5.3 电池存储习惯**"),
                put_table([values3],header=header3),
                put_markdown("**存储温度分布**")])
            
            

    
# main()
start_server(main, port=8080, remote_access=False, debug=True, host="0.0.0.0")