结构健康监测仿真-主题046-结构健康监测中的区块链技术
·
结构健康监测仿真 - 主题046：结构健康监测中的区块链技术
目录
引言
1.1 背景与动机
随着结构健康监测（SHM）系统的规模不断扩大，监测数据的安全性、完整性和可追溯性成为了关键挑战。传统的集中式数据管理系统面临着数据篡改、单点故障、信任缺失等问题。区块链技术作为一种分布式账本技术，为解决这些问题提供了创新的解决方案。
传统SHM数据管理面临的挑战：
- 数据篡改风险：集中式数据库容易被恶意修改
- 数据孤岛：不同机构间的数据难以共享和互认
- 信任问题：数据提供方和使用方之间缺乏信任机制
- 追溯困难：数据变更历史难以追溯和审计
- 隐私泄露：敏感的结构数据面临泄露风险
区块链技术的优势：
- 不可篡改：数据一旦写入区块链就无法修改
- 去中心化：无单点故障，系统更加健壮
- 透明可信：所有参与者共享同一账本
- 全程可追溯：完整记录数据的历史变更
- 隐私保护：通过加密技术保护敏感数据
1.2 区块链的定义
区块链（Blockchain）是一种分布式数据库技术，通过密码学原理将数据区块按时间顺序相连，形成链式结构。每个区块包含一批交易记录，并通过哈希值与前一个区块链接，确保数据的完整性和不可篡改性。
核心特征：
- 分布式存储：数据存储在网络中的多个节点上
- 共识机制：通过算法达成全网一致
- 密码学安全：使用哈希函数和数字签名保护数据
- 智能合约：自动执行的程序代码
- 透明匿名：交易透明但参与者身份可匿名
1.3 发展历程
区块链技术的发展经历了三个阶段：
第一阶段：区块链1.0 - 数字货币（2009-2013）
- 比特币的诞生
- focus on价值转移
- 简单的交易记录
第二阶段：区块链2.0 - 智能合约（2013-2017）
- 以太坊平台的出现
- 可编程的区块链
- 去中心化应用（DApp）
第三阶段：区块链3.0 - 行业应用（2017至今）
- 跨行业应用拓展
- 联盟链和私有链的发展
- 与物联网、AI等技术融合
区块链基础概念
2.1 区块链数据结构
2.1.1 区块结构
区块的组成：
┌─────────────────────────────────────────────────────────────┐
│ 区块头（Block Header） │
├─────────────────────────────────────────────────────────────┤
│ 版本号（Version） │ 4字节 │ 协议版本信息 │
├─────────────────────────────────────────────────────────────┤
│ 前一区块哈希（Prev Hash） │ 32字节 │ 前一个区块的哈希值 │
├─────────────────────────────────────────────────────────────┤
│ Merkle根（Merkle Root） │ 32字节 │ 交易树的根哈希 │
├─────────────────────────────────────────────────────────────┤
│ 时间戳（Timestamp） │ 4字节 │ 区块创建时间 │
├─────────────────────────────────────────────────────────────┤
│ 难度目标（Bits） │ 4字节 │ 挖矿难度 │
├─────────────────────────────────────────────────────────────┤
│ 随机数（Nonce） │ 4字节 │ 工作量证明 │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 区块体（Block Body） │
├─────────────────────────────────────────────────────────────┤
│ 交易计数（Tx Count） │ 变长 │ 交易数量 │
├─────────────────────────────────────────────────────────────┤
│ 交易列表（Transactions） │ 变长 │ 具体的交易数据 │
└─────────────────────────────────────────────────────────────┘
2.1.2 链式结构
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ 区块N │ ←── │ 区块2 │ ←── │ 区块1 │ ←── │ 区块0 │
│ (最新) │ │ │ │ │ │ (创世) │
├──────────┤ ├──────────┤ ├──────────┤ ├──────────┤
│ 区块头 │ │ 区块头 │ │ 区块头 │ │ 区块头 │
│ 区块体 │ │ 区块体 │ │ 区块体 │ │ 区块体 │
│ Prev Hash│ ───→│ Prev Hash│ ───→│ Prev Hash│ ───→│ 0000 │
└──────────┘ └──────────┘ └──────────┘ └──────────┘
2.2 区块链类型
2.2.1 公有链（Public Blockchain）
特点：
- 完全开放，任何人都可以参与
- 去中心化程度最高
- 匿名性强
- 交易确认时间较长
- 能耗较高
代表：
- 比特币（Bitcoin）
- 以太坊（Ethereum）
- 莱特币（Litecoin）
适用场景：
- 数字货币
- 去中心化金融（DeFi）
- 公开透明的数据存证
2.2.2 联盟链（Consortium Blockchain）
特点：
- 由多个组织共同管理
- 部分去中心化
- 身份可验证
- 交易速度快
- 能耗较低
代表：
- Hyperledger Fabric
- R3 Corda
- 蚂蚁链
- 腾讯区块链
适用场景：
- 跨机构数据共享
- 供应链管理
- 金融清算
- SHM数据管理
2.2.3 私有链（Private Blockchain）
特点：
- 单一组织控制
- 中心化程度较高
- 权限严格管理
- 效率最高
- 隐私性最好
适用场景：
- 企业内部数据管理
- 敏感数据存储
- 内部审计追踪
三种区块链对比：
| 特性 | 公有链 | 联盟链 | 私有链 |
|---|---|---|---|
| 访问权限 | 公开 | 授权 | 私有 |
| 去中心化 | 完全 | 部分 | 有限 |
| 交易速度 | 慢 | 快 | 最快 |
| 能耗 | 高 | 中 | 低 |
| 隐私性 | 低 | 中 | 高 |
| 适用场景 | 数字货币 | 跨机构合作 | 企业内部 |
区块链核心技术
3.1 密码学基础
3.1.1 哈希函数
概念：
哈希函数是将任意长度的输入转换为固定长度输出的函数。区块链中常用的哈希函数是SHA-256。
特性：
- 确定性：相同输入产生相同输出
- 快速计算：输入到输出的计算速度快
- 抗碰撞性：难以找到两个不同输入产生相同输出
- 雪崩效应：输入微小变化导致输出巨大变化
- 不可逆性：无法从输出反推输入
Python实现：
import hashlib
def sha256_hash(data):
"""计算SHA-256哈希值"""
if isinstance(data, str):
data = data.encode('utf-8')
return hashlib.sha256(data).hexdigest()
# 示例
data = "结构健康监测数据"
hash_value = sha256_hash(data)
print(f"原始数据: {data}")
print(f"哈希值: {hash_value}")
3.1.2 数字签名
概念：
数字签名用于验证消息的真实性和完整性，同时确保不可否认性。
流程：
发送方： 接收方：
┌─────────┐ ┌─────────┐
│ 原始数据 │ │ 原始数据 │
└────┬────┘ └────▲────┘
│ │
↓ │
┌─────────┐ ┌─────────┐
│ 哈希运算 │ │ 哈希运算 │
└────┬────┘ └────▲────┘
│ │
↓ │
┌─────────┐ ┌─────────┐
│ 私钥签名 │──────────────→│ 公钥验证 │
└─────────┘ 签名+数据 └─────────┘
Python实现：
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
def generate_key_pair():
"""生成RSA密钥对"""
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048
)
public_key = private_key.public_key()
return private_key, public_key
def sign_data(private_key, data):
"""使用私钥签名数据"""
if isinstance(data, str):
data = data.encode('utf-8')
signature = private_key.sign(
data,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return signature
def verify_signature(public_key, data, signature):
"""使用公钥验证签名"""
if isinstance(data, str):
data = data.encode('utf-8')
try:
public_key.verify(
signature,
data,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return True
except:
return False
3.1.3 Merkle树
概念：
Merkle树是一种哈希树，用于高效验证大量数据的完整性。
结构：
┌──────────────┐
│ Merkle Root │ ← 根哈希
└──────┬───────┘
│
┌───────────────┼───────────────┐
│ │
┌────┴────┐ ┌────┴────┐
│ Hash 12 │ │ Hash 34 │
└────┬────┘ └────┬────┘
│ │
┌─────┴─────┐ ┌──────┴──────┐
│ │ │ │
┌────┴────┐ ┌────┴────┐ ┌────┴────┐ ┌────┴────┐
│ Tx1 │ │ Tx2 │ │ Tx3 │ │ Tx4 │ ← 交易数据
└─────────┘ └─────────┘ └─────────┘ └─────────┘
Python实现：
import hashlib
class MerkleTree:
"""Merkle树实现"""
def __init__(self, transactions):
self.transactions = transactions
self.tree = self.build_tree()
def hash_pair(self, a, b):
"""哈希两个值的组合"""
combined = (a + b).encode('utf-8') if b else a.encode('utf-8')
return hashlib.sha256(combined).hexdigest()
def build_tree(self):
"""构建Merkle树"""
if not self.transactions:
return []
# 计算叶子节点哈希
tree_level = [self.hash_pair(tx, '') for tx in self.transactions]
# 逐层构建
tree = [tree_level]
while len(tree_level) > 1:
next_level = []
for i in range(0, len(tree_level), 2):
left = tree_level[i]
right = tree_level[i+1] if i+1 < len(tree_level) else left
next_level.append(self.hash_pair(left, right))
tree_level = next_level
tree.append(tree_level)
return tree
def get_root(self):
"""获取Merkle根"""
if self.tree:
return self.tree[-1][0]
return None
def get_proof(self, index):
"""获取特定交易的证明路径"""
proof = []
for level in self.tree[:-1]:
if index % 2 == 0:
sibling_index = index + 1 if index + 1 < len(level) else index
else:
sibling_index = index - 1
proof.append(level[sibling_index])
index //= 2
return proof
3.2 共识机制
3.2.1 工作量证明（PoW）
概念：
通过计算难题来竞争记账权，计算量越大，获得记账权的概率越高。
流程：
1. 收集交易，构建候选区块
2. 计算区块头的哈希值
3. 检查哈希值是否小于目标值
4. 如果不符合，改变Nonce重新计算
5. 找到符合条件的哈希后，广播区块
6. 其他节点验证后接受区块
Python实现：
import hashlib
import time
def proof_of_work(block_data, difficulty=4):
"""
工作量证明算法
Parameters:
-----------
block_data : str
区块数据
difficulty : int
难度值（前导零的个数）
Returns:
--------
nonce : int
找到的随机数
hash_value : str
符合条件的哈希值
"""
target = '0' * difficulty
nonce = 0
start_time = time.time()
while True:
# 组合数据和随机数
data = f"{block_data}{nonce}"
hash_value = hashlib.sha256(data.encode()).hexdigest()
# 检查是否满足条件
if hash_value.startswith(target):
elapsed_time = time.time() - start_time
print(f"找到Nonce: {nonce}")
print(f"哈希值: {hash_value}")
print(f"耗时: {elapsed_time:.2f}秒")
return nonce, hash_value
nonce += 1
3.2.2 权益证明（PoS）
概念：
根据持有的代币数量和持有时间来分配记账权，持有越多，获得记账权的概率越高。
优点：
- 能耗低
- 效率高
- 安全性好
代表：
- 以太坊2.0
- Cardano
- Polkadot
3.2.3 实用拜占庭容错（PBFT）
概念：
通过多轮投票达成共识，适用于联盟链和私有链。
流程：
1. Request: 客户端发送请求
2. Pre-prepare: 主节点广播预准备消息
3. Prepare: 节点广播准备消息
4. Commit: 节点广播提交消息
5. Reply: 节点回复客户端
特点：
- 交易确认快
- 无分叉
- 适合小规模网络
- 节点数量受限
3.3 智能合约
3.3.1 智能合约概念
定义：
智能合约是运行在区块链上的自动执行的程序代码，当预设条件被满足时，合约自动执行相应的操作。
特性：
- 自动执行：无需人工干预
- 不可篡改：代码部署后无法修改
- 透明公开：所有人都可以查看代码
- 确定性：相同输入产生相同输出
3.3.2 Solidity智能合约示例
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;
contract SHMDataContract {
// 数据结构
struct SensorData {
string sensorId;
uint256 timestamp;
string dataHash;
address provider;
bool isValid;
}
// 状态变量
mapping(string => SensorData) public dataRecords;
mapping(address => bool) public authorizedProviders;
string[] public dataIds;
// 事件
event DataRecorded(string dataId, string sensorId, uint256 timestamp);
event ProviderAuthorized(address provider);
// 修饰器
modifier onlyAuthorized() {
require(authorizedProviders[msg.sender], "Not authorized");
_;
}
// 授权数据提供者
function authorizeProvider(address provider) public {
authorizedProviders[provider] = true;
emit ProviderAuthorized(provider);
}
// 记录传感器数据
function recordData(
string memory dataId,
string memory sensorId,
string memory dataHash
) public onlyAuthorized {
require(bytes(dataRecords[dataId].sensorId).length == 0, "Data already exists");
dataRecords[dataId] = SensorData({
sensorId: sensorId,
timestamp: block.timestamp,
dataHash: dataHash,
provider: msg.sender,
isValid: true
});
dataIds.push(dataId);
emit DataRecorded(dataId, sensorId, block.timestamp);
}
// 验证数据
function verifyData(string memory dataId, string memory dataHash)
public view returns (bool) {
SensorData memory data = dataRecords[dataId];
if (!data.isValid) return false;
return keccak256(abi.encodePacked(data.dataHash)) ==
keccak256(abi.encodePacked(dataHash));
}
// 获取数据记录数量
function getDataCount() public view returns (uint256) {
return dataIds.length;
}
}
区块链在SHM中的应用场景
4.1 数据存证与溯源
4.1.1 数据完整性保护
应用场景：
- 传感器原始数据存证
- 分析报告存证
- 维护记录存证
实现方案：
原始数据 → 哈希计算 → 区块链存储
↓
哈希值 + 时间戳 + 签名
优势：
- 防止数据篡改
- 提供篡改证据
- 支持法律举证
4.1.2 数据溯源
应用场景：
- 追踪数据来源
- 验证数据流转
- 责任追溯
实现方案：
数据创建 → 数据传输 → 数据处理 → 数据使用
↓ ↓ ↓ ↓
区块1 区块2 区块3 区块4
4.2 跨机构数据共享
4.2.1 多方协作场景
参与方：
- 结构业主
- 监测服务商
- 政府部门
- 保险公司
- 研究机构
数据共享流程：
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 业主 │ ←→ │ 区块链 │ ←→ │ 监测方 │
└──────────┘ └──────────┘ └──────────┘
↑↓
┌──────────┐
│ 政府 │
└──────────┘
4.2.2 权限管理
分级权限：
- Level 1: 查看公开数据
- Level 2: 查看详细数据
- Level 3: 提交数据
- Level 4: 管理权限
4.3 设备身份认证
4.3.1 传感器身份管理
DID（去中心化身份）：
传感器DID = did:shm:sensor:uuid
↓
区块链身份注册
↓
公钥证书绑定
4.3.2 设备认证流程
1. 传感器生成密钥对
2. 在区块链上注册DID
3. 监测时签名数据
4. 验证方检查签名
5. 确认数据来源可信
区块链架构设计
5.1 联盟链架构
5.1.1 Hyperledger Fabric架构
核心组件：
┌─────────────────────────────────────────────────────────────┐
│ 应用层（Application Layer） │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ SDK/API │ │ CLI工具 │ │ 前端应用 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 区块链网络（Network Layer） │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Peer节点 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 背书节点 │ │ 提交节点 │ │ 锚节点 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────────────────────────────┘ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Orderer节点 │ │
│ │ （排序服务，共识机制） │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 数据层（Data Layer） │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 账本存储 │ │ 状态数据库 │ │ 历史索引 │ │
│ │ (LevelDB) │ │ (CouchDB) │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
5.1.2 通道机制
概念：
通道是Fabric中的私有数据通道，只有加入通道的节点才能看到该通道的数据。
应用场景：
通道A：业主-监测方-政府（公开数据）
通道B：业主-保险公司（保险数据）
通道C：监测方-研究机构（研究数据）
5.2 节点类型
5.2.1 客户端节点
功能：
- 提交交易提案
- 接收交易结果
- 查询区块链数据
5.2.2 Peer节点
类型：
- 背书节点（Endorser）：模拟执行交易
- 提交节点（Committer）：验证并写入区块
- 锚节点（Anchor）：跨组织通信
5.2.3 Orderer节点
功能：
- 接收交易
- 排序交易
- 打包区块
- 分发区块
智能合约与数据管理
6.1 数据上链策略
6.1.1 完整数据上链
适用场景：
- 数据量小
- 需要完全透明
- 无隐私要求
优点：
- 数据完整可验证
- 无需外部存储
缺点：
- 存储成本高
- 性能受限
6.1.2 哈希上链
适用场景：
- 数据量大
- 需要隐私保护
- 成本敏感
方案：
原始数据 → IPFS/云存储 → 返回哈希 → 区块链存储哈希
优点：
- 成本低
- 性能好
- 隐私保护
6.2 数据隐私保护
6.2.1 同态加密
概念：
允许在加密数据上进行计算，结果解密后与明文计算结果相同。
应用：
加密数据A + 加密数据B = 加密结果
解密(加密结果) = 明文A + 明文B
6.2.2 零知识证明
概念：
证明者向验证者证明某个陈述为真，而不透露任何有用信息。
应用：
- 证明数据完整性而不透露数据内容
- 验证身份而不透露身份信息
6.3 数据查询优化
6.3.1 索引设计
链上索引：
mapping(string => uint256) public dataIndex;
mapping(uint256 => string) public indexToData;
链下索引：
- 使用The Graph等索引服务
- 建立关系型数据库索引
Python仿真实现
7.1 区块链基础实现
import hashlib
import json
import time
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle, FancyBboxPatch, Circle, FancyArrowPatch
import matplotlib.animation as animation
import warnings
warnings.filterwarnings('ignore')
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# 使用Agg后端，不弹出窗口
plt.switch_backend('Agg')
class Transaction:
"""交易类"""
def __init__(self, sender, receiver, data, timestamp=None):
self.sender = sender
self.receiver = receiver
self.data = data
self.timestamp = timestamp or time.time()
self.tx_hash = self.calculate_hash()
def calculate_hash(self):
"""计算交易哈希"""
tx_string = f"{self.sender}{self.receiver}{self.data}{self.timestamp}"
return hashlib.sha256(tx_string.encode()).hexdigest()
def to_dict(self):
"""转换为字典"""
return {
'sender': self.sender,
'receiver': self.receiver,
'data': self.data,
'timestamp': self.timestamp,
'hash': self.tx_hash
}
class Block:
"""区块类"""
def __init__(self, index, transactions, timestamp, previous_hash, nonce=0):
self.index = index
self.transactions = transactions
self.timestamp = timestamp
self.previous_hash = previous_hash
self.nonce = nonce
self.hash = self.calculate_hash()
self.merkle_root = self.calculate_merkle_root()
def calculate_hash(self):
"""计算区块哈希"""
block_string = json.dumps({
'index': self.index,
'transactions': [tx.to_dict() for tx in self.transactions],
'timestamp': self.timestamp,
'previous_hash': self.previous_hash,
'nonce': self.nonce
}, sort_keys=True)
return hashlib.sha256(block_string.encode()).hexdigest()
def calculate_merkle_root(self):
"""计算Merkle根"""
if not self.transactions:
return hashlib.sha256(b'').hexdigest()
hashes = [tx.tx_hash for tx in self.transactions]
while len(hashes) > 1:
if len(hashes) % 2 == 1:
hashes.append(hashes[-1])
new_hashes = []
for i in range(0, len(hashes), 2):
combined = hashes[i] + hashes[i+1]
new_hashes.append(hashlib.sha256(combined.encode()).hexdigest())
hashes = new_hashes
return hashes[0]
def mine_block(self, difficulty=4):
"""挖矿"""
target = '0' * difficulty
while not self.hash.startswith(target):
self.nonce += 1
self.hash = self.calculate_hash()
print(f"区块 {self.index} 挖矿成功，Nonce: {self.nonce}")
def to_dict(self):
"""转换为字典"""
return {
'index': self.index,
'transactions': [tx.to_dict() for tx in self.transactions],
'timestamp': self.timestamp,
'previous_hash': self.previous_hash,
'hash': self.hash,
'merkle_root': self.merkle_root,
'nonce': self.nonce
}
class Blockchain:
"""区块链类"""
def __init__(self, difficulty=4):
self.chain = [self.create_genesis_block()]
self.difficulty = difficulty
self.pending_transactions = []
self.mining_reward = 1
self.nodes = set()
def create_genesis_block(self):
"""创建创世区块"""
return Block(0, [], time.time(), "0")
def get_latest_block(self):
"""获取最新区块"""
return self.chain[-1]
def add_transaction(self, transaction):
"""添加交易到待处理列表"""
self.pending_transactions.append(transaction)
def mine_pending_transactions(self, mining_reward_address):
"""挖矿待处理交易"""
# 创建奖励交易
reward_tx = Transaction("system", mining_reward_address, f"Reward {self.mining_reward}")
self.pending_transactions.append(reward_tx)
# 创建新区块
new_block = Block(
len(self.chain),
self.pending_transactions,
time.time(),
self.get_latest_block().hash
)
# 挖矿
new_block.mine_block(self.difficulty)
# 添加到链
self.chain.append(new_block)
self.pending_transactions = []
return new_block
def is_chain_valid(self):
"""验证区块链完整性"""
for i in range(1, len(self.chain)):
current_block = self.chain[i]
previous_block = self.chain[i-1]
# 验证当前区块哈希
if current_block.hash != current_block.calculate_hash():
return False
# 验证前一区块哈希链接
if current_block.previous_hash != previous_block.hash:
return False
return True
def get_chain_stats(self):
"""获取区块链统计信息"""
total_transactions = sum(len(block.transactions) for block in self.chain)
return {
'block_count': len(self.chain),
'total_transactions': total_transactions,
'pending_transactions': len(self.pending_transactions),
'chain_valid': self.is_chain_valid(),
'difficulty': self.difficulty
}
class SHMBlockchainSystem:
"""SHM区块链系统"""
def __init__(self, n_nodes=5):
self.n_nodes = n_nodes
self.blockchain = Blockchain(difficulty=3)
self.nodes = [f"Node_{i}" for i in range(n_nodes)]
self.sensor_data = []
def simulate_sensor_data(self, n_samples=100):
"""模拟传感器数据"""
for i in range(n_samples):
sensor_id = f"sensor_{i % 20}"
data = {
'timestamp': time.time() + i,
'value': np.random.normal(0, 1),
'location': f"location_{i % 10}",
'type': np.random.choice(['acceleration', 'strain', 'temperature'])
}
# 创建交易
tx = Transaction(
sender=sensor_id,
receiver="blockchain",
data=json.dumps(data)
)
self.blockchain.add_transaction(tx)
self.sensor_data.append({
'sensor_id': sensor_id,
'data': data,
'tx_hash': tx.tx_hash
})
# 每10个交易挖矿一次
if (i + 1) % 10 == 0:
miner = np.random.choice(self.nodes)
self.blockchain.mine_pending_transactions(miner)
return self.sensor_data
def verify_data_integrity(self, data_index):
"""验证数据完整性"""
if data_index >= len(self.sensor_data):
return False
data_record = self.sensor_data[data_index]
tx_hash = data_record['tx_hash']
# 在区块链中查找交易
for block in self.blockchain.chain:
for tx in block.transactions:
if tx.tx_hash == tx_hash:
return True
return False
def get_system_stats(self):
"""获取系统统计"""
chain_stats = self.blockchain.get_chain_stats()
return {
'n_nodes': self.n_nodes,
'total_sensor_data': len(self.sensor_data),
'chain_stats': chain_stats
}
def simulate_blockchain_shm():
"""区块链SHM仿真"""
print("="*60)
print("区块链技术仿真 - 结构健康监测")
print("="*60)
# 创建SHM区块链系统
print("\n1. 初始化SHM区块链系统...")
shm_system = SHMBlockchainSystem(n_nodes=5)
print(f" 创建了 {shm_system.n_nodes} 个区块链节点")
# 模拟传感器数据
print("\n2. 模拟传感器数据上链...")
sensor_data = shm_system.simulate_sensor_data(n_samples=100)
print(f" 生成了 {len(sensor_data)} 条传感器数据记录")
# 获取区块链统计
print("\n3. 区块链统计信息...")
stats = shm_system.get_system_stats()
print(f" 区块数量: {stats['chain_stats']['block_count']}")
print(f" 总交易数: {stats['chain_stats']['total_transactions']}")
print(f" 链有效性: {stats['chain_stats']['chain_valid']}")
# 验证数据完整性
print("\n4. 验证数据完整性...")
verified_count = sum(1 for i in range(len(sensor_data))
if shm_system.verify_data_integrity(i))
print(f" 验证通过: {verified_count}/{len(sensor_data)}")
return {
'shm_system': shm_system,
'sensor_data': sensor_data,
'stats': stats
}
def create_visualization(results):
"""创建可视化"""
print("\n生成可视化...")
fig = plt.figure(figsize=(20, 16))
shm_system = results['shm_system']
sensor_data = results['sensor_data']
stats = results['stats']
blockchain = shm_system.blockchain
# 1. 区块链结构示意图
ax1 = plt.subplot(4, 4, 1)
ax1.set_xlim(0, 10)
ax1.set_ylim(0, 10)
ax1.set_aspect('equal')
ax1.axis('off')
ax1.set_title('Blockchain Structure', fontsize=12, fontweight='bold')
# 绘制区块
block_colors = ['lightblue', 'lightgreen', 'lightyellow', 'lightcoral', 'lightpink']
for i, block in enumerate(blockchain.chain[:5]):
x = 0.5 + i * 1.8
y = 5
# 区块
block_rect = FancyBboxPatch((x, y), 1.5, 2, boxstyle="round,pad=0.05",
facecolor=block_colors[i % len(block_colors)],
edgecolor='black', linewidth=1.5)
ax1.add_patch(block_rect)
ax1.text(x+0.75, y+1.5, f'Block {block.index}', ha='center', fontsize=8, fontweight='bold')
ax1.text(x+0.75, y+1, f'Txs: {len(block.transactions)}', ha='center', fontsize=7)
ax1.text(x+0.75, y+0.5, f'Nonce: {block.nonce}', ha='center', fontsize=6)
# 连接箭头
if i < len(blockchain.chain[:5]) - 1:
ax1.arrow(x+1.5, y+1, 0.3, 0, head_width=0.2, head_length=0.1, fc='red', ec='red')
# 2. 区块增长趋势
ax2 = plt.subplot(4, 4, 2)
block_indices = list(range(len(blockchain.chain)))
tx_counts = [len(block.transactions) for block in blockchain.chain]
ax2.bar(block_indices, tx_counts, color='skyblue', edgecolor='black', alpha=0.7)
ax2.set_xlabel('Block Index')
ax2.set_ylabel('Transaction Count')
ax2.set_title('Transactions per Block')
ax2.grid(True, alpha=0.3)
# 3. 交易类型分布
ax3 = plt.subplot(4, 4, 3)
tx_types = {}
for data in sensor_data:
tx_type = data['data']['type']
tx_types[tx_type] = tx_types.get(tx_type, 0) + 1
colors = ['#ff9999', '#66b3ff', '#99ff99']
ax3.pie(tx_types.values(), labels=tx_types.keys(), autopct='%1.1f%%',
colors=colors, startangle=90)
ax3.set_title('Transaction Type Distribution')
# 4. 传感器活跃度
ax4 = plt.subplot(4, 4, 4)
sensor_counts = {}
for data in sensor_data:
sensor_id = data['sensor_id']
sensor_counts[sensor_id] = sensor_counts.get(sensor_id, 0) + 1
sensors = list(sensor_counts.keys())[:10] # 显示前10个
counts = [sensor_counts[s] for s in sensors]
ax4.barh(range(len(sensors)), counts, color='lightcoral', alpha=0.7)
ax4.set_yticks(range(len(sensors)))
ax4.set_yticklabels(sensors, fontsize=8)
ax4.set_xlabel('Transaction Count')
ax4.set_title('Top 10 Active Sensors')
ax4.grid(True, alpha=0.3)
# 5. 哈希值分布
ax5 = plt.subplot(4, 4, 5)
hashes = [block.hash[:8] for block in blockchain.chain] # 取前8位
hash_values = [int(h, 16) for h in hashes]
ax5.plot(range(len(hash_values)), hash_values, 'b-', linewidth=1, alpha=0.7)
ax5.set_xlabel('Block Index')
ax5.set_ylabel('Hash Value (int)')
ax5.set_title('Block Hash Distribution')
ax5.grid(True, alpha=0.3)
# 6. 挖矿难度与Nonce
ax6 = plt.subplot(4, 4, 6)
nonces = [block.nonce for block in blockchain.chain]
ax6.scatter(range(len(nonces)), nonces, c=range(len(nonces)),
cmap='viridis', s=50, alpha=0.6)
ax6.set_xlabel('Block Index')
ax6.set_ylabel('Nonce Value')
ax6.set_title('Mining Nonce Distribution')
ax6.grid(True, alpha=0.3)
# 7. 时间戳分布
ax7 = plt.subplot(4, 4, 7)
timestamps = [block.timestamp for block in blockchain.chain]
time_diffs = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]
ax7.hist(time_diffs, bins=20, color='gold', edgecolor='black', alpha=0.7)
ax7.axvline(np.mean(time_diffs), color='red', linestyle='--',
label=f'Mean: {np.mean(time_diffs):.2f}s')
ax7.set_xlabel('Block Interval (s)')
ax7.set_ylabel('Frequency')
ax7.set_title('Block Generation Time')
ax7.legend()
ax7.grid(True, alpha=0.3)
# 8. Merkle树可视化
ax8 = plt.subplot(4, 4, 8)
ax8.set_xlim(0, 10)
ax8.set_ylim(0, 10)
ax8.set_aspect('equal')
ax8.axis('off')
ax8.set_title('Merkle Tree Structure', fontsize=12, fontweight='bold')
# 绘制简化的Merkle树
# 根节点
root = FancyBboxPatch((4, 8), 2, 1, boxstyle="round,pad=0.05",
facecolor='lightblue', edgecolor='blue', linewidth=2)
ax8.add_patch(root)
ax8.text(5, 8.5, 'Merkle Root', ha='center', fontsize=9, fontweight='bold')
# 中间层
for i in range(2):
x = 2 + i * 4
node = FancyBboxPatch((x, 5), 1.5, 1, boxstyle="round,pad=0.05",
facecolor='lightgreen', edgecolor='green', linewidth=1)
ax8.add_patch(node)
ax8.text(x+0.75, 5.5, f'Hash {i+1}', ha='center', fontsize=8)
ax8.plot([5, x+0.75], [8, 6], 'k-', linewidth=1)
# 叶子节点
for i in range(4):
x = 0.5 + i * 2.5
leaf = FancyBboxPatch((x, 2), 1.5, 1, boxstyle="round,pad=0.05",
facecolor='lightyellow', edgecolor='orange', linewidth=1)
ax8.add_patch(leaf)
ax8.text(x+0.75, 2.5, f'Tx {i+1}', ha='center', fontsize=7)
# 9. 数据完整性验证
ax9 = plt.subplot(4, 4, 9)
verification_results = [shm_system.verify_data_integrity(i) for i in range(len(sensor_data))]
verified = sum(verification_results)
not_verified = len(verification_results) - verified
ax9.pie([verified, not_verified], labels=['Verified', 'Not Verified'],
autopct='%1.1f%%', colors=['green', 'red'], startangle=90)
ax9.set_title('Data Integrity Verification')
# 10. 节点分布
ax10 = plt.subplot(4, 4, 10)
node_activity = {node: 0 for node in shm_system.nodes}
for block in blockchain.chain:
for tx in block.transactions:
if tx.receiver in node_activity:
node_activity[tx.receiver] += 1
ax10.bar(range(len(node_activity)), list(node_activity.values()),
color='lightblue', alpha=0.7)
ax10.set_xlabel('Node ID')
ax10.set_ylabel('Activity Count')
ax10.set_title('Node Activity Distribution')
ax10.set_xticks(range(len(node_activity)))
ax10.set_xticklabels([f'N{i}' for i in range(len(node_activity))], fontsize=8)
ax10.grid(True, alpha=0.3)
# 11. 区块链性能指标
ax11 = plt.subplot(4, 4, 11)
metrics = ['Blocks', 'Tx/Block', 'Difficulty', 'Integrity']
values = [
stats['chain_stats']['block_count'],
stats['chain_stats']['total_transactions'] / max(1, stats['chain_stats']['block_count']),
stats['chain_stats']['difficulty'] * 10,
100 if stats['chain_stats']['chain_valid'] else 0
]
colors = ['green' if v > 50 else 'yellow' if v > 20 else 'red' for v in values]
bars = ax11.barh(metrics, values, color=colors, alpha=0.7)
ax11.set_xlabel('Value')
ax11.set_title('Blockchain Performance')
ax11.grid(True, alpha=0.3, axis='x')
# 12. 数据价值分布
ax12 = plt.subplot(4, 4, 12)
data_values = [abs(data['data']['value']) for data in sensor_data]
ax12.hist(data_values, bins=30, color='lightgreen', edgecolor='black', alpha=0.7)
ax12.axvline(np.mean(data_values), color='red', linestyle='--',
label=f'Mean: {np.mean(data_values):.2f}')
ax12.set_xlabel('Data Value')
ax12.set_ylabel('Frequency')
ax12.set_title('Sensor Data Value Distribution')
ax12.legend()
ax12.grid(True, alpha=0.3)
# 13. 地理位置分布
ax13 = plt.subplot(4, 4, 13)
locations = {}
for data in sensor_data:
loc = data['data']['location']
locations[loc] = locations.get(loc, 0) + 1
loc_names = list(locations.keys())[:8]
loc_counts = [locations[loc] for loc in loc_names]
ax13.bar(range(len(loc_names)), loc_counts, color='orange', alpha=0.7)
ax13.set_xlabel('Location ID')
ax13.set_ylabel('Data Count')
ax13.set_title('Data by Location')
ax13.set_xticks(range(len(loc_names)))
ax13.set_xticklabels([f'L{i}' for i in range(len(loc_names))], fontsize=8)
ax13.grid(True, alpha=0.3)
# 14. 交易时间线
ax14 = plt.subplot(4, 4, 14)
tx_times = [data['data']['timestamp'] for data in sensor_data]
tx_times_normalized = [(t - tx_times[0]) / 60 for t in tx_times] # 转换为分钟
ax14.scatter(tx_times_normalized, range(len(tx_times)),
c=range(len(tx_times)), cmap='plasma', s=20, alpha=0.6)
ax14.set_xlabel('Time (minutes)')
ax14.set_ylabel('Transaction Index')
ax14.set_title('Transaction Timeline')
ax14.grid(True, alpha=0.3)
# 15. 系统架构图
ax15 = plt.subplot(4, 4, 15)
ax15.set_xlim(0, 10)
ax15.set_ylim(0, 10)
ax15.set_aspect('equal')
ax15.axis('off')
ax15.set_title('SHM Blockchain Architecture', fontsize=12, fontweight='bold')
# 传感器层
for i in range(5):
x = 0.5 + i * 1.8
sensor = Circle((x+0.5, 2), 0.3, facecolor='orange', edgecolor='red', linewidth=1)
ax15.add_patch(sensor)
ax15.text(x+0.5, 2, f'S{i}', ha='center', va='center', fontsize=7)
# 区块链层
blockchain_layer = FancyBboxPatch((1, 4), 8, 2, boxstyle="round,pad=0.1",
facecolor='lightblue', edgecolor='blue', linewidth=2)
ax15.add_patch(blockchain_layer)
ax15.text(5, 5, 'Blockchain Network', ha='center', fontsize=10, fontweight='bold')
# 应用层
app_layer = FancyBboxPatch((2, 7), 6, 1.5, boxstyle="round,pad=0.1",
facecolor='lightgreen', edgecolor='green', linewidth=2)
ax15.add_patch(app_layer)
ax15.text(5, 7.75, 'SHM Applications', ha='center', fontsize=10, fontweight='bold')
# 连接
for i in range(5):
x = 1 + i * 1.8
ax15.plot([x, 5], [2.3, 4], 'k--', linewidth=0.5, alpha=0.5)
# 16. 系统统计信息
ax16 = plt.subplot(4, 4, 16)
ax16.axis('off')
stats_text = f"""
SHM Blockchain System
======================
System Configuration:
• Blockchain Nodes: {stats['n_nodes']}
• Difficulty: {stats['chain_stats']['difficulty']}
• Mining Reward: {blockchain.mining_reward}
Blockchain Stats:
• Total Blocks: {stats['chain_stats']['block_count']}
• Total Transactions: {stats['chain_stats']['total_transactions']}
• Pending Transactions: {stats['chain_stats']['pending_transactions']}
• Chain Valid: {stats['chain_stats']['chain_valid']}
Sensor Data:
• Total Records: {stats['total_sensor_data']}
• Verified Records: {verified}
• Data Types: {len(tx_types)}
Latest Block:
• Index: {blockchain.get_latest_block().index}
• Hash: {blockchain.get_latest_block().hash[:16]}...
• Transactions: {len(blockchain.get_latest_block().transactions)}
• Merkle Root: {blockchain.get_latest_block().merkle_root[:16]}...
"""
ax16.text(0.05, 0.95, stats_text, fontsize=8, verticalalignment='top',
fontfamily='monospace', bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
plt.suptitle('Blockchain Technology for Structural Health Monitoring',
fontsize=16, fontweight='bold', y=0.995)
plt.tight_layout(rect=[0, 0, 1, 0.99])
plt.savefig('blockchain_shm_analysis.png', dpi=150, bbox_inches='tight')
print(" 综合分析图已保存: blockchain_shm_analysis.png")
plt.close()
def create_animation(results):
"""创建数据流动画"""
print("\n生成动画...")
fig, axes = plt.subplots(1, 2, figsize=(16, 8))
shm_system = results['shm_system']
sensor_data = results['sensor_data']
blockchain = shm_system.blockchain
# 左图：区块链增长
ax1 = axes[0]
ax1.set_xlim(0, 15)
ax1.set_ylim(0, 15)
ax1.set_aspect('equal')
ax1.axis('off')
ax1.set_title('Blockchain Growth', fontsize=14, fontweight='bold')
# 右图：交易统计
ax2 = axes[1]
ax2.set_xlim(0, 100)
ax2.set_ylim(0, 20)
ax2.set_xlabel('Transaction Index', fontsize=12)
ax2.set_ylabel('Block Index', fontsize=12)
ax2.set_title('Transaction to Block Mapping', fontsize=14, fontweight='bold')
ax2.grid(True, alpha=0.3)
# 添加文本信息
text_info = ax2.text(0.02, 0.98, '', transform=ax2.transAxes, fontsize=10,
verticalalignment='top', fontfamily='monospace',
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8))
n_frames = min(50, len(blockchain.chain))
def init():
text_info.set_text('Initializing...')
return text_info,
def update(frame):
# 清除之前的绘制
ax1.clear()
ax1.set_xlim(0, 15)
ax1.set_ylim(0, 15)
ax1.set_aspect('equal')
ax1.axis('off')
ax1.set_title('Blockchain Growth', fontsize=14, fontweight='bold')
# 绘制到当前帧的区块
block_colors = ['lightblue', 'lightgreen', 'lightyellow', 'lightcoral', 'lightpink']
for i in range(min(frame + 1, len(blockchain.chain))):
block = blockchain.chain[i]
x = 1 + (i % 5) * 2.5
y = 12 - (i // 5) * 3
# 区块
block_rect = FancyBboxPatch((x, y), 2, 1.5, boxstyle="round,pad=0.05",
facecolor=block_colors[i % len(block_colors)],
edgecolor='black', linewidth=1.5)
ax1.add_patch(block_rect)
ax1.text(x+1, y+1, f'B{block.index}', ha='center', fontsize=9, fontweight='bold')
ax1.text(x+1, y+0.5, f'{len(block.transactions)} txs', ha='center', fontsize=7)
# 更新右图
ax2.clear()
ax2.set_xlim(0, 100)
ax2.set_ylim(0, 20)
ax2.set_xlabel('Transaction Index', fontsize=12)
ax2.set_ylabel('Block Index', fontsize=12)
ax2.set_title('Transaction to Block Mapping', fontsize=14, fontweight='bold')
ax2.grid(True, alpha=0.3)
# 绘制交易分布
tx_index = 0
for block_idx in range(min(frame + 1, len(blockchain.chain))):
block = blockchain.chain[block_idx]
for tx in block.transactions:
ax2.scatter(tx_index, block_idx, c=block_idx, cmap='tab10', s=30, alpha=0.6)
tx_index += 1
# 更新文本
current_block = blockchain.chain[min(frame, len(blockchain.chain)-1)]
info_text = f"""Frame: {frame}
Block: {current_block.index}
Transactions: {len(current_block.transactions)}
Hash: {current_block.hash[:16]}...
Nonce: {current_block.nonce}"""
text_info.set_text(info_text)
return text_info,
# 创建动画
anim = animation.FuncAnimation(fig, update, init_func=init,
frames=n_frames, interval=300, blit=False)
# 保存动画
anim.save('blockchain_shm_animation.gif', writer='pillow', fps=3, dpi=100)
print(" 动画已保存: blockchain_shm_animation.gif")
plt.close()
def main():
"""主函数"""
print("\n" + "="*60)
print("结构健康监测中的区块链技术仿真")
print("="*60)
# 运行仿真
results = simulate_blockchain_shm()
# 创建可视化
create_visualization(results)
# 创建动画
create_animation(results)
print("\n" + "="*60)
print("仿真完成！")
print("="*60)
print("\n生成的文件:")
print(" - blockchain_shm_analysis.png (综合分析图)")
print(" - blockchain_shm_animation.gif (数据流动画)")
if __name__ == "__main__":
main()
案例分析
8.1 桥梁监测数据存证系统
8.1.1 系统架构
某跨海大桥区块链监测系统：
参与方：
- 桥梁管理局（数据所有者）
- 监测服务商（数据提供者）
- 政府监管部门（审计者）
- 保险公司（数据使用者）
- 研究机构（分析者）
部署架构：
- 联盟链节点：5个（每方一个）
- 传感器：200个
- 智能合约：3个
* 数据存证合约
* 权限管理合约
* 数据查询合约
8.1.2 实施效果
数据安全性：
- 数据篡改检测率：100%
- 数据完整性验证：实时
- 审计追溯：完整历史记录
运营效率：
- 数据查询时间：秒级
- 跨机构数据共享：自动化
- 争议解决：证据确凿
8.2 建筑结构全生命周期管理
8.2.1 应用场景
生命周期阶段：
设计阶段 → 施工阶段 → 运营阶段 → 维护阶段 → 拆除阶段
↓ ↓ ↓ ↓ ↓
设计数据 施工记录 监测数据 维修记录 拆除报告
↓ ↓ ↓ ↓ ↓
区块100 区块500 区块2000 区块3500 区块4000
8.2.2 价值体现
- 责任追溯：明确各阶段责任主体
- 数据传承：完整的数据历史传承
- 价值评估：基于完整历史的资产评估
总结与展望
9.1 技术总结
区块链技术为结构健康监测带来了革命性的变化：
优势：
- 数据不可篡改：确保监测数据的真实性
- 全程可追溯：完整记录数据生命周期
- 多方协作：打破数据孤岛，促进共享
- 智能合约：自动化执行监测协议
- 隐私保护：加密技术保护敏感数据
挑战：
- 性能瓶颈：交易处理速度有限
- 存储成本：链上存储成本较高
- 技术复杂：需要专业知识维护
- 标准缺失：行业应用标准待完善
9.2 发展趋势
技术发展方向：
- Layer 2扩展：提高交易处理能力
- 跨链技术：实现不同链之间的互操作
- 隐私计算：同态加密、零知识证明的应用
- 物联网融合：区块链与IoT的深度结合
应用前景：
- 智慧城市：城市基础设施的统一监测平台
- 数字孪生：区块链保障的数字孪生数据
- 碳中和：结构碳排放的追踪与验证
- 保险科技：基于区块链的自动化理赔
9.3 实施建议
对于SHM系统的区块链应用：
- 选择合适的链类型：联盟链适合大多数场景
- 设计合理的数据上链策略：哈希上链为主
- 建立完善的权限管理：分级授权机制
- 做好与传统系统的集成：渐进式改造
- 培养专业人才：区块链+工程复合人才
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
8
0- 0
已为社区贡献156条内容
所有评论(0)