crbn4/Building
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: .gitignore

Browse Source 
modified:   README.md
	new file:   requirement.txt
	new file:   scripts_2/README.md
	new file:   scripts_2/__init__.py
	new file:   scripts_2/__pycache__/__init__.cpython-310.pyc
	new file:   scripts_2/__pycache__/signal_utils.cpython-310.pyc
	new file:   scripts_2/__pycache__/task3_identify.cpython-310.pyc
	new file:   scripts_2/signal_utils.py
	new file:   scripts_2/task3_identify.py
	new file:   scripts_3/README.md
	new file:   scripts_3/__init__.py
	new file:   scripts_3/__pycache__/__init__.cpython-310.pyc
	new file:   scripts_3/__pycache__/task4_predict_freq.cpython-310.pyc
	new file:   scripts_3/task4_predict_freq.py
This commit is contained in:
CrbnsCat10n
2026-05-06 16:16:57 +08:00
parent fc8fcc2746
commit 1b3e836d98
15 changed files with 883 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -2,3 +2,4 @@
.vscode/
scripts/__pycache__/
old/
evaluation_outputs/

267
README.md
View File

@@ -1,193 +1,164 @@
# 建筑结构响应预测项目
# 建筑结构课程评测代码总览 (Course Evaluation Toolkit)
## 项目概述
## 项目简介
本项目面向建筑结构振动课程评测场景，提供三类核心功能：
本项目使用时间卷积网络（Temporal Convolutional Network, TCN）模型来预测建筑结构在地震波和简谐波激励下的响应。模型基于底部传感器的振动数据预测多个传感器的响应，实现结构健康监测和振动控制。
1. 根据基底简谐激励频率，预测模型顶部加速度 Z 向响应的整体均方根 RMS 值。
2. 根据顶部自由衰减响应，识别结构一阶自振频率和阻尼比。
3. 根据顶部强迫振动响应，反推出输入简谐波的激振频率。
## 主要特性
项目包含三套脚本目录：`scripts`、`scripts_2`、`scripts_3`。三套脚本分别对应 RMS 预测、结构动力特性识别、激振频率逆向预测。
- **多输出预测**：基于单个输入传感器预测5个输出传感器的响应
- **时间序列建模**：使用TCN处理序列数据，支持长距离依赖
- **数据增强**：支持有/无TMD（调谐质量阻尼器）条件下的数据训练
- **自动化评估**：提供完整的训练和测试流程
## 课程评测流程
教师评测时可按以下流程使用代码：
1. **顶部响应 RMS 预测**
教师进行一组基底激励输入试验。对每组简谐波频率，使用 `scripts` 目录下的脚本预测顶部加速度 Z 向响应的整体 RMS 值。
2. **结构动力参数识别**
教师对模型略微做出改动后，采集顶部自由衰减响应，使用 `scripts_2` 目录下的脚本识别一阶自振频率和阻尼比。
3. **激振频率逆向预测**
教师再次进行一组基底激励输入试验，使用 `scripts_3` 目录下的脚本，根据顶部加速度响应反推输入简谐波的激振频率。
## 项目结构
```
```text
Building-main/
├── src/
│ ├── __init__.py
│ ├── config.py # 配置文件，包含模型和训练参数
│ ├── dataset.py # 数据集处理和数据加载器
│ ├── model.py # TCN模型定义
│ ├── train.py # 训练脚本
│ └── evaluate.py # 评估脚本
├── downloads/ # 数据目录
│ ├── Non_TMD/ # 无TMD数据
│ │ ├── train/ # 训练数据
│ │ ├── val/ # 验证数据
│ │ └── test/ # 测试数据
│ └── TMD/ # 有TMD数据
├── 数据链接.txt # 数据下载链接配置文件
├── download_from_data_links.py # 数据下载脚本
├── best_model.pth # 训练好的模型权重
└── README.md # 项目说明文档
├── downloads/ # 数据目录
│ └── Non_TMD/
│ ├── free_vib/ # 自由衰减数据
│ ├── train/ # 简谐波训练数据
│ ├── val/ # 简谐波验证数据
│ └── test/ # 其他测试数据
├── scripts/ # 任务 1：顶部响应 RMS 预测
├── scripts_2/ # 任务 2：自振频率与阻尼比识别
├── scripts_3/ # 任务 3：激振频率逆向预测
├── requirement.txt # Python 依赖清单
└── README.md # 总说明文档
```
## 环境要求
## 三套脚本说明
- Python 3.8+
- PyTorch 1.9+
- CUDA（可选，用于GPU加速）
- 其他依赖：numpy, pandas, scikit-learn, matplotlib
### `scripts`
该目录用于完成顶部响应 RMS 预测任务。
## 安装依赖
- 输入为基底传感器信号与文件频率标签。
- 输出为顶部加速度 Z 向响应的整体 RMS 预测值。
- 主要脚本包括 `train_final.py`、`evaluate.py`、`predict_single.py`。
1. 创建虚拟环境：
```bash
python -m venv .venv310
# Windows
.venv310\Scripts\activate
# Linux/Mac
source .venv310/bin/activate
```
### `scripts_2`
该目录用于完成结构动力特性识别任务。
2. 安装依赖：
```bash
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install numpy pandas scikit-learn matplotlib
```
- 数据来源为 `downloads/Non_TMD/free_vib/` 目录下的自由衰减 CSV 文件。
- 使用顶部响应传感器 `WSMS00007` 的 `value3` 列。
- 输出一阶自振频率、阻尼比与诊断图像。
## 数据准备
### `scripts_3`
该目录用于完成激振频率逆向预测任务。
### 下载数据
- 使用强迫振动数据，如 `downloads/Non_TMD/val/` 或 `downloads/Non_TMD/test/`。
- 使用顶部响应传感器 `WSMS00007` 的 `value3` 列识别主频。
- 主要脚本为 `task4_predict_freq.py`。
运行数据下载脚本从配置文件下载数据：
## 传感器约束
- **顶部响应传感器**
任务 2 和任务 3 统一使用 `WSMS00007` 的 `value3` 列。
- **基底输入传感器**
`WSMS00012` 的 `value1` 列用于参考与对照。
- **任务 3 频率逆推**
激振频率预测直接基于顶部响应频谱结果。
## 环境准备
项目建议使用 **Python 3.10** 创建新的虚拟环境，再安装根目录的 `requirement.txt`。
**1. 创建虚拟环境**
```bash
python download_from_data_links.py
py -3.10 -m venv .venv
```
脚本会从 `数据链接.txt` 中读取下载链接，并将数据保存到 `downloads/` 目录。
### 数据格式
数据为CSV格式的长表结构，包含以下字段：
- `code`: 传感器编码
- `type`: 数据类型
- `time`: 时间戳
- `value1`, `value2`, `value3`: 三个轴的振动值
### 数据集说明
- **Non_TMD**: 无调谐质量阻尼器条件下的数据
- **TMD**: 安装调谐质量阻尼器条件下的数据
- 包含谐波激励（不同频率和振幅）和地震波激励数据
## 配置说明
主要配置参数在 `src/config.py` 中：
- **数据配置**：
- `SEQ_LEN`: 序列长度（512）
- `STEP_SIZE`: 滑动窗口步长（20）
- `BATCH_SIZE`: 批大小（256）
- **传感器配置**：
- `INPUT_SENSOR`: 输入传感器编码（'WSMS00012'）
- `OUTPUT_SENSORS`: 输出传感器编码列表
- `INPUT_AXIS`: 输入轴（'value1' - X轴）
- `OUTPUT_AXIS`: 输出轴（'value3' - Z轴）
- **模型配置**：
- `CHANNELS`: TCN各层通道数 [64, 64, 128, 128, 256, 256]
- `KERNEL_SIZE`: 卷积核大小（5）
- `DROPOUT`: Dropout率（0.2）
- **训练配置**：
- `LEARNING_RATE`: 学习率（1e-4）
- `EPOCHS`: 训练轮数（50）
- `WEIGHT_DECAY`: 权重衰减（1e-3）
## 训练模型
### 基本训练
运行训练脚本：
**2. 激活虚拟环境**
```bash
cd src
python train.py
.\.venv\Scripts\activate
```
训练过程会：
1. 加载Non_TMD条件下的训练和验证数据
2. 初始化TCN模型
3. 使用AdamW优化器和学习率调度器训练
4. 保存最佳模型到 `best_model.pth`
### 高级训练选项
修改 `src/config.py` 中的参数来自定义训练：
- 调整学习率、批大小等超参数
- 启用/禁用早停（`ENABLE_EARLY_STOP`）
- 修改模型架构参数
## 模型评估
运行评估脚本：
**3. 安装依赖**
```bash
cd src
python evaluate.py
python -m pip install -r requirement.txt
```
评估过程会：
1. 加载测试数据
2. 加载训练好的模型权重
3. 对测试数据进行预测
4. 生成可视化结果保存为 `test_results.png`
`requirement.txt` 已包含三套脚本使用到的核心依赖：
## 模型架构
- `numpy`
- `pandas`
- `scipy`
- `matplotlib`
- `scikit-learn`
### BuildingTCN
## 使用方法
- **输入**: (batch_size, seq_len, 1) - 底部传感器X轴数据
- **输出**: (batch_size, seq_len, 5) - 5个传感器的Z轴预测
- **架构**: 多层TCN + 线性输出层
### 1. 顶部响应 RMS 预测
对单个简谐波文件预测顶部响应 RMS：
### TemporalConvNet
```bash
python scripts\predict_single.py --file downloads\Non_TMD\val\harmonic_5mm_0.75Hz.csv
```
- 使用膨胀卷积（dilation）实现长距离依赖
- 残差连接保证梯度传播
- 权重归一化提高训练稳定性
对全量样本进行评估：
## 使用说明
```bash
python scripts\evaluate.py
```
### 自定义数据集
如需重新训练模型：
1. 将新数据放入 `downloads/Non_TMD/` 或 `downloads/TMD/` 相应子目录
2. 确保数据格式符合要求
3. 修改 `src/config.py` 中的传感器配置（如需要）
4. 重新运行训练脚本
```bash
python scripts\train_final.py
```
### 预测新数据
### 2. 自由衰减法识别自振频率和阻尼比
对自由衰减文件执行识别：
1. 准备输入数据（CSV格式）
2. 修改 `src/dataset.py` 中的数据加载逻辑
3. 使用加载的模型进行推理
```bash
python scripts_2\task3_identify.py --file downloads\Non_TMD\free_vib\free_decay.csv
```
## 注意事项
脚本输出内容包括：
- 确保数据目录结构正确
- GPU内存不足时可适当减小 `BATCH_SIZE`
- 模型收敛可能需要调整学习率
- 测试数据可视化仅显示第一个batch的第一条序列
- 一阶自振频率 `f_n`
- 对数衰减率 `delta`
- 阻尼比 `zeta`
- 原始波形、滤波结果与包络线图像
## 贡献
### 3. 根据顶部响应反推激振频率
对单个强迫振动文件执行频率逆推：
欢迎提交Issue和Pull Request来改进项目。
```bash
python scripts_3\task4_predict_freq.py --file downloads\Non_TMD\val\harmonic_5mm_0.75Hz.csv
```
## 许可证
脚本输出内容包括：
本项目仅供学习和研究使用。
- 稳态段采样范围
- 预测激振频率
- 文件名中的频率真值
- 绝对误差
- 时域与频域诊断图像
## 结果输出
运行脚本后，图像与评测结果默认保存到 `evaluation_outputs/` 目录下，对应子目录如下：
- `evaluation_outputs/task1_final/`
- `evaluation_outputs/task3_identify/`
- `evaluation_outputs/task4_predict_freq/`
## 说明
- `scripts` 目录负责 RMS 预测。
- `scripts_2` 目录负责自由衰减识别。
- `scripts_3` 目录负责激振频率逆推。
- 数据文件命名中包含频率标签时，脚本会自动提取真值用于误差统计。

6
requirement.txt Normal file
View File

@@ -0,0 +1,6 @@
# Python 3.10
numpy>=1.24,<2.0
pandas>=2.0,<3.0
scipy>=1.10,<2.0
matplotlib>=3.7,<4.0
scikit-learn>=1.3,<2.0

43
scripts_2/README.md Normal file
View File

@@ -0,0 +1,43 @@
# 结构动力特性识别脚本 (Free-Vibration Identification)
## 项目简介
本目录用于完成建筑结构动力特性识别任务。脚本基于专门的自由衰减数据文件，提取顶部响应传感器 `WSMS00007` 的 Z 向信号，识别结构一阶自振频率与阻尼比，并输出可视化诊断结果。
## 技术路线
本目录的分析流程围绕自由衰减信号展开，包含信号读取、主频识别、滤波降噪与阻尼估计四个步骤：
1. **顶部响应读取**
从 `free_vib` 目录下的 CSV 文件中读取 `WSMS00007` 的 `value3` 列，作为结构顶部响应信号。
2. **频域主频识别**
对原始自由衰减信号执行 FFT，在 `0.1Hz ~ 5.0Hz` 范围内寻找主峰，识别一阶自振频率。
3. **带通滤波**
以识别出的主频为中心，构造巴特沃斯带通滤波器，对原始信号进行滤波，得到平滑的衰减振动波形。
4. **阻尼比计算**
从滤波信号中提取波峰，自动选择包络线质量较高的一段，通过对数衰减率计算平均衰减量与阻尼比。
## 核心模块说明
- **`signal_utils.py`**
共享信号处理工具。负责传感器常量定义、采样率估计、稳态段截取、FFT 主频识别、抛物线插值、带通滤波与波峰筛选。
- **`task3_identify.py`**
任务主入口。负责校验自由衰减数据来源、执行一阶频率与阻尼比识别，并生成包含原始波形、滤波波形和包络线的诊断图。
## 使用指南
**1. 运行任务脚本**
对单个自由衰减 CSV 文件执行结构动力特性识别：
```bash
python scripts_2\task3_identify.py --file downloads\Non_TMD\free_vib\free_decay.csv
```
**2. 指定输出目录**
将图像与诊断结果输出到自定义目录：
```bash
python scripts_2\task3_identify.py --file downloads\Non_TMD\free_vib\free_decay.csv --output-dir evaluation_outputs\task3_custom
```
**3. 调整滤波带宽**
通过命令行参数修改主频两侧的带宽范围：
```bash
python scripts_2\task3_identify.py --file downloads\Non_TMD\free_vib\free_decay.csv --bandwidth 0.25
```

1
scripts_2/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Signal-processing scripts for free-vibration identification."""

BIN
scripts_2/__pycache__/__init__.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
scripts_2/__pycache__/signal_utils.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
scripts_2/__pycache__/task3_identify.cpython-310.pyc Normal file
View File

Binary file not shown.

273
scripts_2/signal_utils.py Normal file
View File

@@ -0,0 +1,273 @@
from __future__ import annotations
from pathlib import Path
import re
import numpy as np
import pandas as pd
from scipy.signal import butter, filtfilt, find_peaks
TOP_RESPONSE_SENSOR = "WSMS00007"
TOP_RESPONSE_AXIS = "value3"
BASE_REFERENCE_SENSOR = "WSMS00012"
BASE_REFERENCE_AXIS = "value1"
TIME_COLUMN = "time"
CODE_COLUMN = "code"
def ensure_parent_dir(path: Path) -> Path:
path = Path(path).resolve()
path.mkdir(parents=True, exist_ok=True)
return path
def parse_frequency_from_filename(file_name: str) -> float | None:
match = re.search(r"(\d+(?:\.\d+)?)Hz", file_name, flags=re.IGNORECASE)
if match is None:
return None
return float(match.group(1))
def estimate_sampling_rate(time_values: np.ndarray) -> float:
time_values = np.asarray(time_values, dtype=np.float64).reshape(-1)
if time_values.size < 2:
raise ValueError("At least two time samples are required to estimate sampling rate.")
dt = np.diff(time_values)
dt = dt[np.isfinite(dt)]
dt = dt[dt > 0.0]
if dt.size == 0:
raise ValueError("Failed to infer a positive sampling interval from the CSV time column.")
return float(1.0 / np.median(dt))
def calculate_rms(signal: np.ndarray) -> float:
signal = np.asarray(signal, dtype=np.float64).reshape(-1)
return float(np.sqrt(np.mean(np.square(signal))))
def read_sensor_signal(
file_path: Path,
sensor_code: str = TOP_RESPONSE_SENSOR,
value_column: str = TOP_RESPONSE_AXIS,
) -> tuple[np.ndarray, np.ndarray]:
df = pd.read_csv(file_path)
sensor_df = df.loc[df[CODE_COLUMN] == sensor_code, [TIME_COLUMN, value_column]].copy()
if sensor_df.empty:
raise ValueError(f"Sensor {sensor_code}/{value_column} was not found in {file_path.name}.")
sensor_df = sensor_df.sort_values(TIME_COLUMN)
sensor_df = sensor_df.drop_duplicates(subset=TIME_COLUMN, keep="first")
sensor_df[TIME_COLUMN] = sensor_df[TIME_COLUMN].astype("float64")
sensor_df[value_column] = sensor_df[value_column].astype("float64")
time_values = sensor_df[TIME_COLUMN].to_numpy(dtype=np.float64)
signal = sensor_df[value_column].to_numpy(dtype=np.float64)
if time_values.size < 8:
raise ValueError(f"Too few usable samples in {file_path.name}.")
return time_values, signal
def get_middle_segment(
time_values: np.ndarray,
signal: np.ndarray,
start_ratio: float = 0.20,
end_ratio: float = 0.80,
steady_window_ratio: float = 0.25,
stride_ratio: float = 0.05,
min_segment_length: int = 512,
subwindow_count: int = 4,
) -> tuple[np.ndarray, np.ndarray]:
signal = np.asarray(signal, dtype=np.float64).reshape(-1)
time_values = np.asarray(time_values, dtype=np.float64).reshape(-1)
if signal.size != time_values.size:
raise ValueError("Time and signal arrays must have the same length.")
length = signal.size
search_start = int(length * start_ratio)
search_end = int(length * end_ratio)
search_start = max(0, min(search_start, length - 1))
search_end = max(search_start + 1, min(search_end, length))
if search_end - search_start < min_segment_length:
center = length // 2
half = min_segment_length // 2
search_start = max(0, center - half)
search_end = min(length, search_start + min_segment_length)
search_start = max(0, search_end - min_segment_length)
search_length = search_end - search_start
window_length = max(min_segment_length, int(length * steady_window_ratio))
window_length = min(window_length, search_length)
if search_length <= window_length:
return time_values[search_start:search_end], signal[search_start:search_end]
stride = max(1, int(length * stride_ratio))
candidate_signal = signal[search_start:search_end]
candidate_rms = calculate_rms(candidate_signal)
best_score = float("inf")
best_slice = slice(search_start, search_start + window_length)
for window_start in range(search_start, search_end - window_length + 1, stride):
window_end = window_start + window_length
window_signal = signal[window_start:window_end]
split_windows = np.array_split(window_signal, subwindow_count)
split_rms = np.asarray([calculate_rms(chunk) for chunk in split_windows], dtype=np.float64)
split_peaks = np.asarray([float(np.max(np.abs(chunk))) for chunk in split_windows], dtype=np.float64)
rms_cv = float(split_rms.std() / max(split_rms.mean(), 1e-8))
peak_cv = float(split_peaks.std() / max(split_peaks.mean(), 1e-8))
window_rms = calculate_rms(window_signal)
score = rms_cv + 0.35 * peak_cv - 0.05 * (window_rms / max(candidate_rms, 1e-8))
if score < best_score:
best_score = score
best_slice = slice(window_start, window_end)
return time_values[best_slice], signal[best_slice]
def compute_windowed_spectrum(
signal: np.ndarray,
sampling_rate: float,
zero_padding_factor: int = 8,
) -> tuple[np.ndarray, np.ndarray]:
signal = np.asarray(signal, dtype=np.float64).reshape(-1)
if signal.size < 4:
raise ValueError("Signal is too short for FFT analysis.")
centered = signal - np.mean(signal)
window = np.hanning(centered.size)
fft_size = int(2 ** np.ceil(np.log2(max(centered.size * zero_padding_factor, centered.size))))
fft_values = np.fft.rfft(centered * window, n=fft_size)
freqs = np.fft.rfftfreq(fft_size, d=1.0 / sampling_rate)
magnitudes = np.abs(fft_values)
return freqs, magnitudes
def parabolic_peak_frequency(freqs: np.ndarray, magnitudes: np.ndarray, peak_index: int) -> float:
if peak_index <= 0 or peak_index >= magnitudes.size - 1:
return float(freqs[peak_index])
alpha = float(magnitudes[peak_index - 1])
beta = float(magnitudes[peak_index])
gamma = float(magnitudes[peak_index + 1])
denominator = alpha - 2.0 * beta + gamma
if abs(denominator) < 1e-12:
return float(freqs[peak_index])
offset = 0.5 * (alpha - gamma) / denominator
bin_width = float(freqs[1] - freqs[0])
return float(freqs[peak_index] + offset * bin_width)
def dominant_frequency_in_band(
signal: np.ndarray,
sampling_rate: float,
min_hz: float = 0.1,
max_hz: float = 5.0,
zero_padding_factor: int = 8,
) -> tuple[float, np.ndarray, np.ndarray, int]:
freqs, magnitudes = compute_windowed_spectrum(
signal=signal,
sampling_rate=sampling_rate,
zero_padding_factor=zero_padding_factor,
)
band_mask = (freqs >= min_hz) & (freqs <= max_hz)
if not np.any(band_mask):
raise ValueError(f"No FFT bins fall inside {min_hz:.2f} Hz to {max_hz:.2f} Hz.")
band_indices = np.flatnonzero(band_mask)
peak_index = int(band_indices[np.argmax(magnitudes[band_mask])])
peak_frequency = parabolic_peak_frequency(freqs, magnitudes, peak_index)
return peak_frequency, freqs, magnitudes, peak_index
def bandpass_filter(
signal: np.ndarray,
sampling_rate: float,
center_hz: float,
bandwidth_hz: float = 0.3,
order: int = 4,
) -> np.ndarray:
nyquist = 0.5 * sampling_rate
low_hz = max(0.05, center_hz - bandwidth_hz)
high_hz = min(center_hz + bandwidth_hz, nyquist * 0.98)
if not low_hz < high_hz:
raise ValueError("Invalid band-pass range. Check the estimated natural frequency and sampling rate.")
b, a = butter(order, [low_hz / nyquist, high_hz / nyquist], btype="bandpass")
return filtfilt(b, a, np.asarray(signal, dtype=np.float64))
def detect_decay_peaks(filtered_signal: np.ndarray, sampling_rate: float, frequency_hz: float) -> np.ndarray:
filtered_signal = np.asarray(filtered_signal, dtype=np.float64).reshape(-1)
samples_per_cycle = max(int(round(sampling_rate / max(frequency_hz, 1e-6))), 3)
min_peak_distance = max(1, int(0.65 * samples_per_cycle))
peaks, _ = find_peaks(filtered_signal, distance=min_peak_distance)
if peaks.size < 4:
peaks, _ = find_peaks(np.abs(filtered_signal), distance=max(1, min_peak_distance // 2))
return peaks.astype(int)
def select_decay_peak_window(
filtered_signal: np.ndarray,
peak_indices: np.ndarray,
sampling_rate: float,
min_peaks: int = 5,
max_peaks: int = 12,
) -> dict[str, np.ndarray | float | int]:
filtered_signal = np.asarray(filtered_signal, dtype=np.float64).reshape(-1)
peak_indices = np.asarray(peak_indices, dtype=int).reshape(-1)
if peak_indices.size < min_peaks:
raise ValueError("Too few peaks were detected for log-decrement damping estimation.")
amplitudes = np.abs(filtered_signal[peak_indices])
tail_start = int(filtered_signal.size * 0.85)
noise_slice = filtered_signal[tail_start:] if tail_start < filtered_signal.size else filtered_signal
noise_floor = max(float(np.median(np.abs(noise_slice))), 1e-8)
useful_mask = amplitudes > max(3.0 * noise_floor, 0.05 * float(amplitudes.max()))
useful_indices = peak_indices[useful_mask]
useful_amplitudes = amplitudes[useful_mask]
if useful_indices.size < min_peaks:
useful_indices = peak_indices
useful_amplitudes = amplitudes
best_window: dict[str, np.ndarray | float | int] | None = None
best_score = -float("inf")
max_window_size = min(max_peaks, useful_indices.size)
for window_size in range(min_peaks, max_window_size + 1):
for start in range(1, useful_indices.size - window_size + 1):
stop = start + window_size
idx_window = useful_indices[start:stop]
amp_window = useful_amplitudes[start:stop]
if np.any(amp_window <= 0.0):
continue
time_window = idx_window.astype(np.float64) / sampling_rate
log_amp = np.log(amp_window)
slope, intercept = np.polyfit(time_window, log_amp, 1)
if slope >= 0.0:
continue
fit_values = slope * time_window + intercept
ss_res = float(np.sum((log_amp - fit_values) ** 2))
ss_tot = float(np.sum((log_amp - np.mean(log_amp)) ** 2))
r_squared = 1.0 - ss_res / max(ss_tot, 1e-12)
amplitude_ratio = float(amp_window[0] / max(amp_window[-1], 1e-12))
score = r_squared + 0.02 * np.log(max(amplitude_ratio, 1.0))
if score > best_score:
best_score = score
best_window = {
"peak_indices": idx_window,
"peak_amplitudes": amp_window,
"time_window": time_window,
"fit_log_amplitudes": fit_values,
"fit_amplitudes": np.exp(fit_values),
"r_squared": r_squared,
"noise_floor": noise_floor,
}
if best_window is None:
raise ValueError("Failed to isolate a clean decay envelope segment for damping estimation.")
return best_window

226
scripts_2/task3_identify.py Normal file
View File

@@ -0,0 +1,226 @@
from __future__ import annotations
import argparse
from pathlib import Path
import sys
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
if __package__ is None or __package__ == "":
sys.path.append(str(Path(__file__).resolve().parents[1]))
from scripts_2.signal_utils import (
TOP_RESPONSE_AXIS,
TOP_RESPONSE_SENSOR,
bandpass_filter,
detect_decay_peaks,
dominant_frequency_in_band,
ensure_parent_dir,
estimate_sampling_rate,
read_sensor_signal,
select_decay_peak_window,
)
def parse_args() -> argparse.Namespace:
project_root = Path(__file__).resolve().parents[1]
parser = argparse.ArgumentParser(description="Identify the first natural frequency and damping ratio from free-decay data.")
parser.add_argument(
"--file",
type=str,
default=str(project_root / "downloads" / "Non_TMD" / "free_vib" / "free_decay.csv"),
help="Path to a free-vibration CSV inside the dedicated free_vib folder.",
)
parser.add_argument(
"--output-dir",
type=str,
default=str(project_root / "evaluation_outputs" / "task3_identify"),
help="Directory for plots and optional exported diagnostics.",
)
parser.add_argument(
"--bandwidth",
type=float,
default=0.3,
help="Half-bandwidth (Hz) of the Butterworth band-pass filter around the identified natural frequency.",
)
parser.add_argument(
"--show",
action="store_true",
help="Display the generated figure in addition to saving it.",
)
return parser.parse_args()
def validate_free_vibration_source(file_path: Path) -> None:
if "free_vib" not in {part.lower() for part in file_path.parts}:
raise ValueError(
f"{file_path.name} is not inside a dedicated free_vib folder. Forced-vibration tails are forbidden for task 3."
)
def build_envelope_curve(
time_values: np.ndarray,
peak_times: np.ndarray,
fit_amplitudes: np.ndarray,
) -> tuple[np.ndarray, np.ndarray]:
if peak_times.size < 2:
return time_values, np.zeros_like(time_values)
envelope = np.interp(time_values, peak_times, fit_amplitudes, left=fit_amplitudes[0], right=fit_amplitudes[-1])
return time_values, envelope
def save_task3_figure(
file_path: Path,
output_dir: Path,
time_values: np.ndarray,
raw_signal: np.ndarray,
filtered_signal: np.ndarray,
peak_indices: np.ndarray,
selected_peak_indices: np.ndarray,
selected_peak_amplitudes: np.ndarray,
fit_amplitudes: np.ndarray,
natural_frequency_hz: float,
damping_ratio: float,
) -> Path:
output_dir = ensure_parent_dir(output_dir)
peak_times = time_values[selected_peak_indices]
envelope_time, envelope = build_envelope_curve(time_values, peak_times, fit_amplitudes)
fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=False)
fig.suptitle(f"Task 3 | Free-Vibration Identification | {file_path.name}")
axes[0].plot(time_values, raw_signal, color="tab:gray", linewidth=1.0, label="Raw top response")
axes[0].set_title(f"Top Response ({TOP_RESPONSE_SENSOR} / {TOP_RESPONSE_AXIS})")
axes[0].set_ylabel("Acceleration")
axes[0].grid(True, alpha=0.25)
axes[0].legend()
axes[1].plot(time_values, filtered_signal, color="tab:blue", linewidth=1.0, label="Band-pass filtered")
axes[1].scatter(time_values[peak_indices], filtered_signal[peak_indices], color="tab:orange", s=14, label="Detected peaks")
axes[1].set_title(f"Filtered around f_n = {natural_frequency_hz:.4f} Hz")
axes[1].set_ylabel("Acceleration")
axes[1].grid(True, alpha=0.25)
axes[1].legend()
axes[2].plot(time_values, filtered_signal, color="tab:blue", linewidth=1.0, alpha=0.65, label="Filtered signal")
axes[2].scatter(peak_times, selected_peak_amplitudes, color="tab:red", s=28, label="Selected envelope peaks")
axes[2].plot(envelope_time, envelope, color="tab:green", linewidth=1.8, label="Fitted positive envelope")
axes[2].plot(envelope_time, -envelope, color="tab:green", linewidth=1.2, linestyle="--", label="Fitted negative envelope")
axes[2].set_title(f"Selected decay segment | damping ratio zeta = {damping_ratio:.5f}")
axes[2].set_xlabel("Time (s)")
axes[2].set_ylabel("Acceleration")
axes[2].grid(True, alpha=0.25)
axes[2].legend()
plt.tight_layout()
figure_path = output_dir / f"{file_path.stem}_task3_identify.png"
plt.savefig(figure_path, dpi=180, bbox_inches="tight")
return figure_path
def analyze_free_vibration(
file_path: Path,
output_dir: Path,
bandwidth_hz: float = 0.3,
show: bool = False,
) -> dict[str, float | int | str | Path]:
file_path = Path(file_path).resolve()
validate_free_vibration_source(file_path)
time_values, raw_signal = read_sensor_signal(file_path, sensor_code=TOP_RESPONSE_SENSOR, value_column=TOP_RESPONSE_AXIS)
sampling_rate = estimate_sampling_rate(time_values)
natural_frequency_hz, freqs, magnitudes, peak_index = dominant_frequency_in_band(
raw_signal,
sampling_rate=sampling_rate,
min_hz=0.1,
max_hz=5.0,
zero_padding_factor=16,
)
filtered_signal = bandpass_filter(
raw_signal,
sampling_rate=sampling_rate,
center_hz=natural_frequency_hz,
bandwidth_hz=bandwidth_hz,
order=4,
)
peak_indices = detect_decay_peaks(filtered_signal, sampling_rate=sampling_rate, frequency_hz=natural_frequency_hz)
peak_window = select_decay_peak_window(filtered_signal, peak_indices, sampling_rate=sampling_rate)
selected_peak_indices = np.asarray(peak_window["peak_indices"], dtype=int)
selected_peak_amplitudes = np.asarray(peak_window["peak_amplitudes"], dtype=np.float64)
fit_amplitudes = np.asarray(peak_window["fit_amplitudes"], dtype=np.float64)
log_decrements = np.log(selected_peak_amplitudes[:-1] / selected_peak_amplitudes[1:])
mean_log_decrement = float(np.mean(log_decrements))
damping_ratio = float(mean_log_decrement / (2.0 * np.pi))
damping_ratio_exact = float(
mean_log_decrement / np.sqrt((2.0 * np.pi) ** 2 + mean_log_decrement**2)
)
figure_path = save_task3_figure(
file_path=file_path,
output_dir=output_dir,
time_values=time_values,
raw_signal=raw_signal,
filtered_signal=filtered_signal,
peak_indices=peak_indices,
selected_peak_indices=selected_peak_indices,
selected_peak_amplitudes=selected_peak_amplitudes,
fit_amplitudes=fit_amplitudes,
natural_frequency_hz=natural_frequency_hz,
damping_ratio=damping_ratio,
)
if show:
plt.show()
plt.close("all")
return {
"file_path": str(file_path),
"sensor_code": TOP_RESPONSE_SENSOR,
"axis": TOP_RESPONSE_AXIS,
"sampling_rate_hz": float(sampling_rate),
"natural_frequency_hz": float(natural_frequency_hz),
"fft_peak_bin_hz": float(freqs[peak_index]),
"fft_peak_magnitude": float(magnitudes[peak_index]),
"detected_peak_count": int(len(peak_indices)),
"selected_peak_count": int(len(selected_peak_indices)),
"mean_log_decrement": mean_log_decrement,
"damping_ratio": damping_ratio,
"damping_ratio_exact": damping_ratio_exact,
"envelope_r2": float(peak_window["r_squared"]),
"figure_path": str(figure_path),
}
def print_report(result: dict[str, float | int | str | Path]) -> None:
print("=== Task 3: Structural Dynamic Identification ===")
print(f"Input file: {result['file_path']}")
print(f"Top response sensor: {result['sensor_code']} / {result['axis']}")
print(f"Sampling rate: {result['sampling_rate_hz']:.4f} Hz")
print(f"Natural frequency f_n: {result['natural_frequency_hz']:.6f} Hz")
print(f"Mean log decrement delta: {result['mean_log_decrement']:.6f}")
print(f"Damping ratio zeta (delta / 2pi): {result['damping_ratio']:.6f}")
print(f"Damping ratio exact: {result['damping_ratio_exact']:.6f}")
print(f"Detected peaks: {result['detected_peak_count']}")
print(f"Selected peaks for envelope: {result['selected_peak_count']}")
print(f"Envelope linearity R^2: {result['envelope_r2']:.6f}")
print(f"Figure saved to: {result['figure_path']}")
def main() -> None:
args = parse_args()
result = analyze_free_vibration(
file_path=Path(args.file),
output_dir=Path(args.output_dir),
bandwidth_hz=float(args.bandwidth),
show=bool(args.show),
)
print_report(result)
if __name__ == "__main__":
main()

33
scripts_3/README.md Normal file
View File

@@ -0,0 +1,33 @@
# 激振频率逆向预测 (Forced-Vibration Frequency Prediction)
## 项目简介
本目录用于完成强迫振动频率逆向预测任务。脚本基于建筑顶部响应传感器 `WSMS00007` 的 Z 向数据，从稳态振动段中识别主频，反推出地震台输入的简谐波频率，并输出频率误差与可视化结果。
## 技术路线
本目录的分析流程覆盖稳态段提取、主频精细识别与结果校验三个部分：
1. **稳态中段截取**
对强迫振动时程执行滑动窗口搜索，从中间区域选出波动最稳定的响应片段，用于频谱分析。
2. **顶部响应主频识别**
读取 `WSMS00007` 的 `value3` 列，对稳态段进行加窗 FFT 与补零计算，并结合抛物线插值输出高精度激振频率。
3. **文件名真值校验**
从样本文件名中提取频率标签，计算预测值与真值之间的绝对误差。
## 核心模块说明
- **`task4_predict_freq.py`**
任务主脚本。负责稳态段截取、频谱分析、抛物线插值、误差计算与频域图像生成。
## 使用指南
**1. 运行任务脚本**
对单个强迫振动样本预测激振频率：
```bash
python scripts_3\task4_predict_freq.py --file downloads\Non_TMD\val\harmonic_5mm_0.75Hz.csv
```
**2. 指定输出目录**
将图像保存到自定义目录：
```bash
python scripts_3\task4_predict_freq.py --file downloads\Non_TMD\val\harmonic_5mm_0.75Hz.csv --output-dir evaluation_outputs\task4_custom
```

1
scripts_3/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Signal-processing scripts for forced-vibration frequency prediction."""

BIN
scripts_3/__pycache__/__init__.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
scripts_3/__pycache__/task4_predict_freq.cpython-310.pyc Normal file
View File

Binary file not shown.

180
scripts_3/task4_predict_freq.py Normal file
View File

@@ -0,0 +1,180 @@
from __future__ import annotations
import argparse
from pathlib import Path
import sys
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
if __package__ is None or __package__ == "":
sys.path.append(str(Path(__file__).resolve().parents[1]))
from scripts_2.signal_utils import (
TOP_RESPONSE_AXIS,
TOP_RESPONSE_SENSOR,
dominant_frequency_in_band,
ensure_parent_dir,
estimate_sampling_rate,
get_middle_segment,
parse_frequency_from_filename,
read_sensor_signal,
)
def parse_args() -> argparse.Namespace:
project_root = Path(__file__).resolve().parents[1]
parser = argparse.ArgumentParser(description="Predict the harmonic input frequency from top-response steady-state data only.")
parser.add_argument(
"--file",
type=str,
default=str(project_root / "downloads" / "Non_TMD" / "val" / "harmonic_5mm_0.75Hz.csv"),
help="Path to a forced-vibration CSV such as data from Non_TMD val/test.",
)
parser.add_argument(
"--output-dir",
type=str,
default=str(project_root / "evaluation_outputs" / "task4_predict_freq"),
help="Directory for plots and optional exported diagnostics.",
)
parser.add_argument(
"--show",
action="store_true",
help="Display the generated figure in addition to saving it.",
)
return parser.parse_args()
def save_task4_figure(
file_path: Path,
output_dir: Path,
time_segment: np.ndarray,
signal_segment: np.ndarray,
freqs: np.ndarray,
magnitudes: np.ndarray,
predicted_frequency_hz: float,
true_frequency_hz: float | None,
) -> Path:
output_dir = ensure_parent_dir(output_dir)
fig, axes = plt.subplots(2, 1, figsize=(12, 8))
fig.suptitle(f"Task 4 | Forced-Vibration Frequency Prediction | {file_path.name}")
axes[0].plot(time_segment, signal_segment, color="tab:blue", linewidth=1.0)
axes[0].set_title(f"Steady middle segment from {TOP_RESPONSE_SENSOR} / {TOP_RESPONSE_AXIS}")
axes[0].set_xlabel("Time (s)")
axes[0].set_ylabel("Acceleration")
axes[0].grid(True, alpha=0.25)
band_mask = (freqs >= 0.1) & (freqs <= 5.0)
axes[1].plot(freqs[band_mask], magnitudes[band_mask], color="tab:purple", linewidth=1.1, label="Windowed FFT")
axes[1].axvline(
predicted_frequency_hz,
color="tab:red",
linestyle="--",
linewidth=1.6,
label=f"Predicted = {predicted_frequency_hz:.5f} Hz",
)
if true_frequency_hz is not None:
axes[1].axvline(
true_frequency_hz,
color="tab:green",
linestyle=":",
linewidth=1.6,
label=f"Filename truth = {true_frequency_hz:.5f} Hz",
)
axes[1].set_title("High-resolution spectrum from top steady-state response")
axes[1].set_xlabel("Frequency (Hz)")
axes[1].set_ylabel("Magnitude")
axes[1].grid(True, alpha=0.25)
axes[1].legend()
plt.tight_layout()
figure_path = output_dir / f"{file_path.stem}_task4_predict_freq.png"
plt.savefig(figure_path, dpi=180, bbox_inches="tight")
return figure_path
def analyze_forced_vibration(
file_path: Path,
output_dir: Path,
show: bool = False,
) -> dict[str, float | int | str]:
file_path = Path(file_path).resolve()
time_values, raw_signal = read_sensor_signal(file_path, sensor_code=TOP_RESPONSE_SENSOR, value_column=TOP_RESPONSE_AXIS)
time_segment, signal_segment = get_middle_segment(time_values, raw_signal)
sampling_rate = estimate_sampling_rate(time_segment)
predicted_frequency_hz, freqs, magnitudes, peak_index = dominant_frequency_in_band(
signal_segment,
sampling_rate=sampling_rate,
min_hz=0.1,
max_hz=5.0,
zero_padding_factor=32,
)
true_frequency_hz = parse_frequency_from_filename(file_path.name)
absolute_error_hz = None if true_frequency_hz is None else abs(predicted_frequency_hz - true_frequency_hz)
figure_path = save_task4_figure(
file_path=file_path,
output_dir=output_dir,
time_segment=time_segment,
signal_segment=signal_segment,
freqs=freqs,
magnitudes=magnitudes,
predicted_frequency_hz=predicted_frequency_hz,
true_frequency_hz=true_frequency_hz,
)
if show:
plt.show()
plt.close("all")
return {
"file_path": str(file_path),
"sensor_code": TOP_RESPONSE_SENSOR,
"axis": TOP_RESPONSE_AXIS,
"sampling_rate_hz": float(sampling_rate),
"segment_sample_count": int(signal_segment.size),
"segment_start_time": float(time_segment[0]),
"segment_end_time": float(time_segment[-1]),
"predicted_frequency_hz": float(predicted_frequency_hz),
"fft_peak_bin_hz": float(freqs[peak_index]),
"fft_peak_magnitude": float(magnitudes[peak_index]),
"true_frequency_hz": None if true_frequency_hz is None else float(true_frequency_hz),
"absolute_error_hz": None if absolute_error_hz is None else float(absolute_error_hz),
"figure_path": str(figure_path),
"note": "Prediction uses only the top response sensor WSMS00007/value3. Base sensor WSMS00012/value1 is not used to infer the answer.",
}
def print_report(result: dict[str, float | int | str]) -> None:
print("=== Task 4: Excitation Frequency Reverse Prediction ===")
print(f"Input file: {result['file_path']}")
print(f"Top response sensor: {result['sensor_code']} / {result['axis']}")
print(f"Sampling rate: {result['sampling_rate_hz']:.4f} Hz")
print(f"Steady segment samples: {result['segment_sample_count']}")
print(f"Steady segment time range: {result['segment_start_time']:.4f} s to {result['segment_end_time']:.4f} s")
print(f"Predicted excitation frequency: {result['predicted_frequency_hz']:.6f} Hz")
if result["true_frequency_hz"] is not None:
print(f"Filename frequency truth: {result['true_frequency_hz']:.6f} Hz")
print(f"Absolute error: {result['absolute_error_hz']:.6f} Hz")
else:
print("Filename frequency truth: unavailable")
print(f"Figure saved to: {result['figure_path']}")
print(result["note"])
def main() -> None:
args = parse_args()
result = analyze_forced_vibration(
file_path=Path(args.file),
output_dir=Path(args.output_dir),
show=bool(args.show),
)
print_report(result)
if __name__ == "__main__":
main()