RailWise KB已发布

沉降数据分析工具 — 累计沉降、差异沉降与沉降速率计算及可视化

自动化监测数据的沉降分析工具，支持累计沉降、差异沉降、沉降速率计算，提供趋势判断和预警分析

复核 2026-07-09入门公开可引用RailWise 技术团队

tool-guide

沉降数据分析工具

1. 工具概述

1.1 用途

本工具用于自动化监测数据的沉降分析，计算累计沉降、差异沉降、沉降速率等关键指标，提供趋势判断和预警分析功能，支持监测日报、周报的快速编制。

1.2 适用场景

自动化全站仪监测数据的批量处理
监测日报、周报、月报的沉降分析
预警触发条件的自动判断
沉降趋势的可视化分析

1.3 输入输出

类型	内容
输入	时序观测数据（时间、点号、高程/坐标）
输出	累计沉降、本次沉降、沉降速率、差异沉降、趋势判断、预警状态

2. 理论基础

2.1 累计沉降计算

$$S_{cum}(t_i) = H_0 - H(t_i)$$

其中：

$S_{cum}$：累计沉降量（mm）
$H_0$：初始高程（mm）
$H(t_i)$：第i次观测高程（mm）

2.2 本次沉降计算

$$S_{this}(t_i) = H(t_{i-1}) - H(t_i)$$

2.3 沉降速率计算

$$v(t_i) = \frac{S_{this}}{\Delta t} = \frac{H(t_{i-1}) - H(t_i)}{t_i - t_{i-1}}$$

单位：mm/d

2.4 差异沉降计算

对于相邻两点A和B：

$$\Delta S_{AB} = S_{cum,A} - S_{cum,B}$$

$$\theta = \frac{\Delta S_{AB}}{L_{AB}}$$

其中：

$\Delta S_{AB}$：差异沉降量（mm）
$L_{AB}$：两点间距离（m）
$\theta$：差异沉降率（mm/m）

2.5 预警等级判定

预警等级	累计沉降(mm)	沉降速率(mm/d)	差异沉降率(mm/m)
正常	< 10	< 0.5	< 0.5
黄色预警	10~20	0.5~2.0	0.5~1.0
橙色预警	20~30	2.0~5.0	1.0~2.0
红色预警	> 30	> 5.0	> 2.0

注：具体阈值应根据项目设计文件和结构类型确定，上表为通用参考值。

3. 使用方法

3.1 命令行调用

python tool_settlement_analysis.py \
    --input monitoring_data.csv \
    --output report.csv \
    --baseline-date 2026-01-01 \
    --adjacent-points pairs.json

3.2 Python API调用

from settlement_analysis import SettlementAnalyzer

# 初始化分析器
analyzer = SettlementAnalyzer(
    baseline_date='2026-01-01',
    alert_thresholds={
        'yellow': {'cum': 10, 'rate': 0.5, 'diff': 0.5},
        'orange': {'cum': 20, 'rate': 2.0, 'diff': 1.0},
        'red': {'cum': 30, 'rate': 5.0, 'diff': 2.0}
    }
)

# 加载数据
data = analyzer.load_data('monitoring_data.csv')

# 计算分析结果
results = analyzer.analyze(data)

# 生成报告
analyzer.generate_report(results, 'report.csv')

# 绘制趋势图
analyzer.plot_trend(results, point_id='P01', save_path='trend.png')

4. 参数说明

4.1 输入参数

参数名	含义	类型	取值范围	必填	说明
`input_file`	输入数据文件	str	-	是	CSV格式，含时间、点号、高程
`baseline_date`	基准日期	str	日期格式	是	累计沉降计算的初始日期
`alert_thresholds`	预警阈值	dict	-	否	各等级阈值
`adjacent_points`	相邻点对	list	-	否	差异沉降计算的点对列表
`smoothing_window`	平滑窗口	int	1~10	否	移动平均窗口大小

4.2 输入数据格式

CSV文件格式：

datetime,point_id,x,y,z,status
2026-01-01 08:00:00,P01,1000.000,2000.000,10.523,正常
2026-01-01 08:00:00,P02,1005.000,2000.000,10.518,正常
2026-01-02 08:00:00,P01,1000.000,2000.000,10.521,正常

4.3 输出数据格式

point_id,datetime,cum_settlement,this_settlement,rate,alert_level,trend
P01,2026-01-01,0.00,0.00,0.00,正常,稳定
P01,2026-01-02,2.00,2.00,2.00,黄色预警,下沉

5. 示例计算

5.1 示例1：单点沉降分析

场景：监测点P01的连续观测数据。

观测数据：

日期	高程(m)
2026-01-01	10.523
2026-01-02	10.521
2026-01-03	10.518
2026-01-04	10.514
2026-01-05	10.509

计算结果：

日期	累计沉降(mm)	本次沉降(mm)	速率(mm/d)	趋势
01-01	0.00	0.00	0.00	稳定
01-02	2.00	2.00	2.00	下沉
01-03	5.00	3.00	3.00	加速下沉
01-04	9.00	4.00	4.00	加速下沉
01-05	14.00	5.00	5.00	红色预警

5.2 示例2：差异沉降分析

场景：相邻桥墩P01和P02，间距20m。

点号	累计沉降(mm)
P01	15.0
P02	8.0

差异沉降：ΔS = 15.0 - 8.0 = 7.0 mm 差异沉降率：θ = 7.0 / 20 = 0.35 mm/m → 正常

6. 结果解读

6.1 趋势判断规则

趋势类型	判断条件	工程含义
稳定	速率 < 0.2 mm/d	结构处于稳定状态
缓慢下沉	0.2~0.5 mm/d	正常施工影响
下沉	0.5~2.0 mm/d	需关注
加速下沉	连续3天速率增大	风险增大，需分析原因
急剧下沉	> 5.0 mm/d	立即预警，采取应急措施
回弹	速率为负	可能为注浆或卸载效应

6.2 预警响应建议

预警等级	响应措施
黄色预警	加密监测频率，分析原因，准备应急措施
橙色预警	高频监测，启动应急预案，通知相关方
红色预警	连续监测，立即停止施工，启动应急响应

7. 代码实现

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
沉降数据分析工具
支持累计沉降、差异沉降、沉降速率计算及可视化

Author: RailWise技术团队
Version: 1.0.0
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime


# 默认预警阈值
DEFAULT_THRESHOLDS = {
    'yellow': {'cum': 10.0, 'rate': 0.5, 'diff': 0.5},
    'orange': {'cum': 20.0, 'rate': 2.0, 'diff': 1.0},
    'red': {'cum': 30.0, 'rate': 5.0, 'diff': 2.0}
}


@dataclass
class SettlementRecord:
    """单点沉降记录"""
    point_id: str
    datetime: datetime
    elevation: float
    cum_settlement: float = 0.0
    this_settlement: float = 0.0
    rate: float = 0.0
    alert_level: str = '正常'
    trend: str = '稳定'


class SettlementAnalyzer:
    """沉降数据分析器"""

    def __init__(
        self,
        baseline_date: str,
        alert_thresholds: Optional[Dict] = None,
        smoothing_window: int = 3
    ):
        """
        初始化分析器

        Args:
            baseline_date: 基准日期 'YYYY-MM-DD'
            alert_thresholds: 预警阈值字典
            smoothing_window: 平滑窗口大小
        """
        self.baseline_date = pd.to_datetime(baseline_date)
        self.thresholds = alert_thresholds or DEFAULT_THRESHOLDS
        self.smoothing_window = smoothing_window

    def load_data(self, filepath: str) -> pd.DataFrame:
        """
        加载监测数据

        Args:
            filepath: CSV文件路径

        Returns:
            DataFrame: 监测数据
        """
        df = pd.read_csv(filepath)
        df['datetime'] = pd.to_datetime(df['datetime'])
        df = df.sort_values(['point_id', 'datetime'])
        return df

    def calculate_settlement(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        计算沉降指标

        Args:
            df: 监测数据DataFrame

        Returns:
            DataFrame: 含沉降分析结果
        """
        results = []

        for point_id, group in df.groupby('point_id'):
            group = group.sort_values('datetime').reset_index(drop=True)

            # 基准高程
            baseline = group.loc[0, 'z']

            for i in range(len(group)):
                row = group.loc[i]

                # 累计沉降
                cum_settlement = (baseline - row['z']) * 1000  # mm

                # 本次沉降和速率
                if i > 0:
                    prev_row = group.loc[i - 1]
                    this_settlement = (prev_row['z'] - row['z']) * 1000
                    delta_days = (row['datetime'] - prev_row['datetime']).total_seconds() / 86400
                    rate = this_settlement / delta_days if delta_days > 0 else 0
                else:
                    this_settlement = 0.0
                    rate = 0.0

                # 预警等级
                alert_level = self._determine_alert(cum_settlement, rate)

                # 趋势判断
                trend = self._determine_trend(rate, i, group, this_settlement)

                results.append({
                    'point_id': point_id,
                    'datetime': row['datetime'],
                    'elevation': row['z'],
                    'cum_settlement': cum_settlement,
                    'this_settlement': this_settlement,
                    'rate': rate,
                    'alert_level': alert_level,
                    'trend': trend
                })

        return pd.DataFrame(results)

    def _determine_alert(self, cum: float, rate: float) -> str:
        """判定预警等级"""
        red = self.thresholds['red']
        orange = self.thresholds['orange']
        yellow = self.thresholds['yellow']

        if cum >= red['cum'] or rate >= red['rate']:
            return '红色预警'
        elif cum >= orange['cum'] or rate >= orange['rate']:
            return '橙色预警'
        elif cum >= yellow['cum'] or rate >= yellow['rate']:
            return '黄色预警'
        else:
            return '正常'

    def _determine_trend(
        self,
        rate: float,
        index: int,
        group: pd.DataFrame,
        this_settlement: float
    ) -> str:
        """判断沉降趋势"""
        if abs(rate) < 0.2:
            return '稳定'
        elif rate < 0:
            return '回弹'
        elif rate < 0.5:
            return '缓慢下沉'
        elif rate < 2.0:
            return '下沉'
        elif rate < 5.0:
            # 检查是否加速
            if index >= 2:
                rates = []
                for j in range(max(0, index-2), index+1):
                    if j > 0:
                        prev = group.loc[j-1]
                        curr = group.loc[j]
                        ds = (prev['z'] - curr['z']) * 1000
                        dt = (curr['datetime'] - prev['datetime']).total_seconds() / 86400
                        rates.append(ds/dt if dt > 0 else 0)
                if len(rates) >= 3 and rates[-1] > rates[-2] > rates[-3]:
                    return '加速下沉'
            return '下沉'
        else:
            return '急剧下沉'

    def calculate_differential(
        self,
        df: pd.DataFrame,
        point_pairs: List[Tuple[str, str, float]]
    ) -> pd.DataFrame:
        """
        计算差异沉降

        Args:
            df: 沉降分析结果
            point_pairs: 点对列表 [(点A, 点B, 距离m), ...]

        Returns:
            DataFrame: 差异沉降结果
        """
        diff_results = []

        for p1, p2, distance in point_pairs:
            data1 = df[df['point_id'] == p1].sort_values('datetime')
            data2 = df[df['point_id'] == p2].sort_values('datetime')

            if len(data1) == 0 or len(data2) == 0:
                continue

            # 按日期对齐
            merged = pd.merge(
                data1[['datetime', 'cum_settlement']],
                data2[['datetime', 'cum_settlement']],
                on='datetime',
                suffixes=('_1', '_2')
            )

            merged['diff_settlement'] = merged['cum_settlement_1'] - merged['cum_settlement_2']
            merged['diff_rate'] = merged['diff_settlement'] / distance
            merged['point_pair'] = f"{p1}-{p2}"
            merged['distance'] = distance

            diff_results.append(merged)

        if diff_results:
            return pd.concat(diff_results, ignore_index=True)
        return pd.DataFrame()

    def plot_trend(
        self,
        df: pd.DataFrame,
        point_id: str,
        save_path: Optional[str] = None,
        show_plot: bool = True
    ) -> None:
        """
        绘制沉降趋势图

        Args:
            df: 沉降分析结果
            point_id: 监测点号
            save_path: 保存路径
            show_plot: 是否显示
        """
        data = df[df['point_id'] == point_id].sort_values('datetime')

        if len(data) == 0:
            print(f"未找到监测点: {point_id}")
            return

        fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True)

        # 累计沉降曲线
        ax1.plot(data['datetime'], data['cum_settlement'], 'b-o', linewidth=2, markersize=4)
        ax1.axhline(y=0, color='k', linewidth=0.5)
        ax1.set_ylabel('累计沉降 (mm)', fontsize=11)
        ax1.set_title(f'监测点 {point_id} 沉降趋势图', fontsize=13)
        ax1.grid(True, alpha=0.3)
        ax1.invert_yaxis()

        # 预警线
        yellow = self.thresholds['yellow']['cum']
        orange = self.thresholds['orange']['cum']
        red = self.thresholds['red']['cum']
        ax1.axhline(y=yellow, color='yellow', linestyle='--', alpha=0.7, label='黄色预警')
        ax1.axhline(y=orange, color='orange', linestyle='--', alpha=0.7, label='橙色预警')
        ax1.axhline(y=red, color='red', linestyle='--', alpha=0.7, label='红色预警')
        ax1.legend(loc='lower right')

        # 沉降速率柱状图
        colors = ['green' if r < 0.5 else 'yellow' if r < 2.0 else 'orange' if r < 5.0 else 'red'
                  for r in data['rate']]
        ax2.bar(data['datetime'], data['rate'], color=colors, alpha=0.7, width=0.5)
        ax2.set_ylabel('沉降速率 (mm/d)', fontsize=11)
        ax2.set_xlabel('日期', fontsize=11)
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()

        if save_path:
            plt.savefig(save_path, dpi=150, bbox_inches='tight')
        if show_plot:
            plt.show()

    def generate_report(self, df: pd.DataFrame, filepath: str) -> None:
        """生成分析报告"""
        df.to_csv(filepath, index=False, encoding='utf-8-sig')
        print(f"报告已保存至: {filepath}")

    def get_summary(self, df: pd.DataFrame) -> Dict:
        """获取汇总统计"""
        latest = df.sort_values('datetime').groupby('point_id').last()

        summary = {
            'total_points': df['point_id'].nunique(),
            'max_cum_settlement': latest['cum_settlement'].max(),
            'max_rate': latest['rate'].max(),
            'alert_count': {
                '正常': len(latest[latest['alert_level'] == '正常']),
                '黄色预警': len(latest[latest['alert_level'] == '黄色预警']),
                '橙色预警': len(latest[latest['alert_level'] == '橙色预警']),
                '红色预警': len(latest[latest['alert_level'] == '红色预警'])
            }
        }

        return summary


# CLI接口
if __name__ == '__main__':
    import argparse
    import json

    parser = argparse.ArgumentParser(description='沉降数据分析工具')
    parser.add_argument('--input', type=str, required=True, help='输入数据文件')
    parser.add_argument('--output', type=str, default='settlement_report.csv', help='输出报告')
    parser.add_argument('--baseline', type=str, required=True, help='基准日期 YYYY-MM-DD')
    parser.add_argument('--pairs', type=str, default=None, help='相邻点对JSON文件')
    parser.add_argument('--plot', type=str, default=None, help='绘制趋势图的点号')

    args = parser.parse_args()

    # 初始化分析器
    analyzer = SettlementAnalyzer(baseline_date=args.baseline)

    # 加载并分析数据
    data = analyzer.load_data(args.input)
    results = analyzer.calculate_settlement(data)

    # 差异沉降分析
    if args.pairs:
        with open(args.pairs, 'r') as f:
            pairs = json.load(f)
        diff_results = analyzer.calculate_differential(results, pairs)
        if not diff_results.empty:
            diff_results.to_csv('differential_settlement.csv', index=False)

    # 生成报告
    analyzer.generate_report(results, args.output)

    # 打印汇总
    summary = analyzer.get_summary(results)
    print("\n沉降分析汇总:")
    print(f"  监测点总数: {summary['total_points']}")
    print(f"  最大累计沉降: {summary['max_cum_settlement']:.2f} mm")
    print(f"  最大沉降速率: {summary['max_rate']:.2f} mm/d")
    print("  预警统计:")
    for level, count in summary['alert_count'].items():
        print(f"    {level}: {count} 点")

    # 绘制趋势图
    if args.plot:
        analyzer.plot_trend(results, args.plot, save_path=f'{args.plot}_trend.png')

8. 注意事项

8.1 精度限制

累计沉降精度取决于初始高程的稳定性，应确保基准点稳定
自动化监测数据需进行粗差剔除后再分析
差异沉降分析需确保两点观测时间同步

8.2 适用条件

适用于自动化全站仪、水准仪等设备的时序数据
适用于常规沉降监测项目
特殊结构（如大跨度桥梁）需单独设定阈值

8.3 常见问题

Q: 基准点发生位移怎么办？ A: 应定期检核基准点，发现位移后重新建立基准或进行基准修正。

Q: 数据缺失如何处理？ A: 单次缺失可插值，连续缺失超过3次应标注并分析原因。

Q: 回弹是否意味着安全？ A: 不一定。回弹可能是注浆压力或卸载效应，需结合施工工况分析。

元数据标签

#沉降分析 #累计沉降 #差异沉降 #沉降速率 #趋势判断 #预警分析 #自动化监测 #Python工具 #RailWise

文档版本: 1.0.0 | 最后更新: 2026-06-10 | 作者: RailWise技术团队