根据率定输出文件,展示模拟过程和精度评价结果。

过程图下:

代码如下(水库名字已更改):

"""
2025/11/20 何坫鹏
模型输出txt文件转换并绘图 - 多场洪水过程子图和精度评价结果显示
"""


import pandas as pd
from io import StringIO

file_path = 'D:/xx水库/2率定/次模率定/场次洪水计算结果.txt'
##########################################################################################################读取场次洪水数据
# 读取所有行
with open(file_path, 'r', encoding='gbk') as f:
    lines = f.readlines()

print(f"原始文件行数: {len(lines)}")

# 查找数据结束的位置(遇到"场次洪水计算统计结果")
end_index = len(lines)
for i, line in enumerate(lines):
    # 检查是否包含"场次洪水计算统计结果"
    if '场次洪水计算统计结果' in line:
        end_index = i
        print(f"在第 {i+1} 行发现'场次洪水计算统计结果',停止读取")
        break

# 只读取"场次洪水计算统计结果"之前的内容
content_before_table = lines[:end_index]

print(f"实际读取行数: {len(content_before_table)}")

# 过滤掉包含多个减号的行(保持原有逻辑)
clean_lines = []
for line in content_before_table:
    # 检查是否包含多个连续减号(全角或半角)
    if '----' not in line and '————' not in line:
        clean_lines.append(line)

print(f"过滤后行数: {len(clean_lines)}")

# 将清洗后的数据写入临时字符串
clean_content = StringIO(''.join(clean_lines))

# 读取数据(保持原有的读取方式)
df = pd.read_csv(clean_content, sep='\s+', encoding='utf-8')

# 保存为CSV文件
output_path = 'D:/xx水库/2率定/次模率定/场次洪水计算结果.csv'
df.to_csv(output_path, index=False, encoding='utf-8')
print(f"转换完成! 保存到: {output_path}")
print(f"数据形状: {df.shape}")
print("列名:", df.columns.tolist())




##########################################################################################################画场次洪水过程图
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号

# 检查数据
print("数据列名:", df.columns.tolist())
print("\n数据概览:")
print(df.head())

# 识别洪水场次 - 当年月日时都为0时,表示新的洪水场次开始
# 创建一个场次标识列
df['场次'] = 0
flood_count = 0

# 先标记场次
for i in range(len(df)):
    # 如果年月日时都为0,则开始新的场次
    if df.iloc[i]['年'] == 0 and df.iloc[i]['月'] == 0 and df.iloc[i]['日'] == 0 and df.iloc[i]['时'] == 0:
        flood_count += 1
    df.loc[df.index[i], '场次'] = flood_count

print(f"识别到 {flood_count} 场洪水")

# 删除年月日时都为0的分隔行
df_clean = df[~((df['年'] == 0) & (df['月'] == 0) & (df['日'] == 0) & (df['时'] == 0))].copy()
print(f"删除分隔行后剩余行数: {len(df_clean)}")

# 创建时间序列(将年、月、日、时合并为datetime)
df_clean['时间'] = pd.to_datetime(
    df_clean['年'].astype(str) + '-' +
    df_clean['月'].astype(str) + '-' +
    df_clean['日'].astype(str) + ' ' +
    df_clean['时'].astype(str) + ':00:00'
)

# 获取实际有数据的场次
valid_floods = []
for i in range(flood_count + 1):  # 场次从0开始
    flood_data = df_clean[df_clean['场次'] == i]
    if len(flood_data) > 0:
        valid_floods.append(i)
valid_floods_number = [valid_floods + 1 for valid_floods in [0, 1, 2, 3]]
print(f"实际有数据的场次: {valid_floods_number}")
actual_flood_count = len(valid_floods)

# 为每场洪水创建子图
if actual_flood_count > 0:
    # 计算子图的行列数
    n_cols = 2  # 每行2个子图
    n_rows = (actual_flood_count + n_cols - 1) // n_cols  # 向上取整

    fig, axes = plt.subplots(n_rows, n_cols, figsize=(15, 5 * n_rows))

    # 如果只有一行,确保axes是二维数组
    if n_rows == 1:
        axes = axes.reshape(1, -1)
    elif actual_flood_count == 1:
        axes = np.array([[axes]])

    # 将axes展平以便迭代
    axes_flat = axes.flatten()

    for idx, flood_id in enumerate(valid_floods):
        # 提取当前场次的洪水数据
        flood_data = df_clean[df_clean['场次'] == flood_id]

        if len(flood_data) == 0:
            continue

        # 设置时间为索引
        flood_data = flood_data.set_index('时间')

        # 在当前子图上绘制
        ax = axes_flat[idx]
        ax.plot(flood_data.index, flood_data['实测流量'], label='实测流量', linewidth=1.5, color='blue')
        ax.plot(flood_data.index, flood_data['计算流量'], label='计算流量', linewidth=1.5, color='red', linestyle='--')

        # 添加标题和标签
        start_time = flood_data.index[0].strftime('%Y-%m-%d %H:%M')
        end_time = flood_data.index[-1].strftime('%Y-%m-%d %H:%M')
        ax.set_title(f'洪水场次 {idx + 1}\n{start_time} 至 {end_time}', fontsize=12)
        ax.set_xlabel('时间', fontsize=10)
        ax.set_ylabel('流量', fontsize=10)
        ax.legend(fontsize=9)
        ax.grid(True, alpha=0.3)

        # 自动调整x轴日期显示格式
        ax.tick_params(axis='x', rotation=45)

    # 隐藏多余的子图
    for i in range(actual_flood_count, len(axes_flat)):
        axes_flat[i].set_visible(False)

    # 调整子图间距
    plt.tight_layout()

    # 保存图形
    plt.savefig('D:/xx水库/2率定/次模率定/多场洪水对比图.png', dpi=300, bbox_inches='tight')

    # 显示图形
    plt.show()


print("绘图完成!")


########################################################################################################展示洪水精度评价结果





import pandas as pd
import re
import matplotlib.pyplot as plt

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号


def parse_flood_data_from_file(file_path):
    """从文件中解析洪水数据,自动处理编码问题"""
    # 尝试不同的编码方式
    encodings = ['utf-8', 'gbk', 'gb2312', 'gb18030', 'latin-1']

    for encoding in encodings:
        try:
            with open(file_path, 'r', encoding=encoding) as file:
                lines = file.readlines()

            print(f"使用编码 {encoding} 成功读取文件")
            print(f"文件总行数: {len(lines)}")

            # 找到表格开始的位置
            start_line = -1
            for i, line in enumerate(lines):
                if ("洪水场次" in line or "场次" in line) and ("降雨量" in line or "径流深" in line):
                    start_line = i
                    print(f"找到表格开始行: {i}, 内容: {line.strip()}")
                    break

            if start_line == -1:
                print("未找到表格开始行")
                continue  # 尝试下一种编码

            # 提取表格数据
            data_lines = []
            for i, line in enumerate(lines[start_line:], start_line):
                line = line.strip()
                # 检查是否包含数字(是数据行)
                if re.search(r'\d', line) and len(line) > 10:
                    data_lines.append((i, line))
                    print(f"数据行 {i}: {line}")
                # 如果遇到空行或明显不是数据的行,停止读取
                elif not line or "---" in line or "====" in line or "————" in line:
                    print(f"跳过行 {i}: {line}")
                    continue

            print(f"找到 {len(data_lines)} 行数据")

            # 解析数据 - 使用更灵活的方法
            data = []
            for i, line in data_lines:  # 跳过表头
                print(f"解析行 {i}: {line}")

                # 方法1: 尝试按竖线分割
                parts = line.split('|')
                print(f"按竖线分割结果: {parts}")

                if len(parts) >= 4:
                    # 第一部分:洪号和参数优选
                    part1_numbers = re.findall(r'-?\d+\.?\d*', parts[0])
                    # 第二部分:径流深相关数据
                    part2_numbers = re.findall(r'-?\d+\.?\d*', parts[1])
                    # 第三部分:洪峰相关数据
                    part3_numbers = re.findall(r'-?\d+\.?\d*', parts[2])
                    # 第四部分:确定性系数
                    part4_numbers = re.findall(r'-?\d+\.?\d*', parts[3])

                    # 合并所有数字
                    all_numbers = part1_numbers + part2_numbers + part3_numbers + part4_numbers
                    print(f"方法1提取的数字: {all_numbers}")

                    if len(all_numbers) >= 12:
                        row_data = {
                            '洪号': all_numbers[0],
                            '参数优选/人工调参': all_numbers[1],
                            '降雨量': float(all_numbers[2]),
                            '实测径流深': float(all_numbers[3]),
                            '计算径流深': float(all_numbers[4]),
                            '绝对误差': float(all_numbers[5]),
                            '相对误差': float(all_numbers[6]),
                            '实测洪峰': float(all_numbers[7]),
                            '计算洪峰': float(all_numbers[8]),
                            '相对误差_洪峰': float(all_numbers[9]),
                            '峰现时差': float(all_numbers[10]),
                            '确定性系数': float(all_numbers[11])
                        }
                        data.append(row_data)
                        print(f"成功解析数据: {row_data}")
                        continue

                # 方法2: 如果方法1失败,尝试直接从整行提取所有数字
                all_numbers = re.findall(r'-?\d+\.?\d*', line)
                print(f"方法2提取的数字: {all_numbers}")

                if len(all_numbers) >= 12:
                    row_data = {
                        '洪号': all_numbers[0],
                        '参数优选/人工调参': all_numbers[1],
                        '降雨量': float(all_numbers[2]),
                        '实测径流深': float(all_numbers[3]),
                        '计算径流深': float(all_numbers[4]),
                        '绝对误差': float(all_numbers[5]),
                        '相对误差': float(all_numbers[6]),
                        '实测洪峰': float(all_numbers[7]),
                        '计算洪峰': float(all_numbers[8]),
                        '相对误差_洪峰': float(all_numbers[9]),
                        '峰现时差': float(all_numbers[10]),
                        '确定性系数': float(all_numbers[11])
                    }
                    data.append(row_data)
                    print(f"成功解析数据: {row_data}")
                else:
                    print(f"警告: 行 {i} 数据不完整,跳过。提取到的数字: {all_numbers}")

            if data:  # 如果成功解析到数据
                print(f"成功解析 {len(data)} 条洪水数据")
                return pd.DataFrame(data)

        except UnicodeDecodeError:
            print(f"编码 {encoding} 失败")
            continue  # 尝试下一种编码
        except Exception as e:
            print(f"解析数据时出错: {e}")
            continue

    print("无法解析文件,尝试了所有编码格式")
    return None


def display_flood_table(df):
    """以表格图形式展示洪水计算统计结果"""
    # 创建图形和轴
    fig, ax = plt.subplots(figsize=(20, 8))
    ax.axis('tight')
    ax.axis('off')

    # 准备表格数据
    table_data = []

    # 添加表头 - 按照您要求的顺序
    headers = [
        '洪号', '参数优选/人工调参', '降雨量(mm)', '实测径流深(mm)',
        '计算径流深(mm)', '绝对误差(mm)', '相对误差',
        '实测洪峰', '计算洪峰', '相对误差', '峰现时差(小时)', '确定性系数'
    ]
    table_data.append(headers)

    # 添加数据行
    for _, row in df.iterrows():
        table_row = [
            row['洪号'],
            row['参数优选/人工调参'],
            f"{row['降雨量']:.1f}",
            f"{row['实测径流深']:.1f}",
            f"{row['计算径流深']:.1f}",
            f"{row['绝对误差']:.2f}",
            f"{row['相对误差']:.2f}",
            f"{row['实测洪峰']:.0f}",
            f"{row['计算洪峰']:.0f}",
            f"{row['相对误差_洪峰']:.2f}",
            f"{row['峰现时差']:.0f}",
            f"{row['确定性系数']:.2f}"
        ]
        table_data.append(table_row)

    # 创建表格
    table = ax.table(
        cellText=table_data,
        loc='center',
        cellLoc='center'
    )

    # 设置表格样式
    table.auto_set_font_size(False)
    table.set_fontsize(9)  # 稍微减小字体以适应更多列
    table.scale(1, 1.8)  # 调整表格大小

    # 设置表头样式
    for i in range(len(headers)):
        table[(0, i)].set_facecolor('#4CAF50')
        table[(0, i)].set_text_props(weight='bold', color='white')

    # 设置数据行交替颜色
    for i in range(1, len(table_data)):
        color = '#F5F5F5' if i % 2 == 1 else '#FFFFFF'
        for j in range(len(headers)):
            table[(i, j)].set_facecolor(color)

    # 设置标题
    plt.title('洪水计算统计结果', fontsize=16, fontweight='bold', pad=20)

    # 调整布局并显示
    plt.tight_layout()
    plt.show()


# 使用示例
if __name__ == "__main__":
    # 替换为你的文件路径
    file_path = 'D:/xx水库/2率定/次模率定/场次洪水计算结果.txt'

    # 解析数据
    df = parse_flood_data_from_file(file_path)

    if df is not None:
        print("\n最终解析的数据:")
        print(df)
        # 只显示洪水计算统计结果表格图
        display_flood_table(df)
    else:

        if df is not None:
            flood_count = len(df)
            print(f"成功解析 {flood_count} 场洪水数据")

            # 显示所有洪水场次
            print("洪水场次列表:", df['洪号'].tolist())

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐