RailWise Python SDK 文档

RailWise Python SDK 文档

1. SDK 概述

RailWise Python SDK（railwise-sdk）是官方提供的 Python 客户端库，用于与 RailWise API 服务进行交互。SDK 提供了完整的类型提示、请求封装和错误处理，兼容 Python 3.9+。

1.1 核心特性

特性	说明
完整类型提示	基于 typing 模块的类型注解，支持 IDE 智能提示
异步支持	同时提供同步和异步 API（asyncio）
自动重试	内置指数退避重试机制
请求拦截	支持自定义请求和响应钩子
错误处理	统一的异常类型和详细的错误信息
Pydantic 模型	请求和响应数据使用 Pydantic 模型验证
数据转换	自动处理日期时间、枚举类型等数据转换

1.2 技术栈

• 语言：Python 3.9+
• HTTP 客户端：httpx（同步 + 异步）
• 数据验证：Pydantic 2.x
• 依赖管理：支持 pip、poetry、conda

2. 安装

2.1 pip 安装

# 使用 pip
pip install railwise-sdk

# 使用 poetry
poetry add railwise-sdk

# 使用 conda
conda install -c railwise railwise-sdk

2.2 版本要求

依赖	最低版本	说明
Python	3.9.0	运行时环境
httpx	0.27.0	HTTP 客户端
pydantic	2.0.0	数据验证和序列化
typing-extensions	4.0.0	类型扩展（Python < 3.11）

2.3 验证安装

python -c "import railwise; print(railwise.__version__)"
# 输出：1.2.0

2.4 版本更新

# 检查当前版本
pip show railwise-sdk

# 更新到最新版本
pip install --upgrade railwise-sdk

# 安装特定版本
pip install railwise-sdk==1.2.0

3. 快速开始

3.1 初始化客户端

from railwise import RailWiseClient

# 使用 API Key 初始化
client = RailWiseClient(
    api_key="rw_live_xxxxxxxxxxxxxxxx",
    base_url="https://api.railwise.cn/v1",  # 可选，默认为生产环境
    timeout=30.0,  # 可选，默认 30.0 秒
    max_retries=3,  # 可选，默认 3 次
)

3.2 环境变量配置

import os
from railwise import RailWiseClient

# 从环境变量读取配置
client = RailWiseClient(
    api_key=os.environ["RAILWISE_API_KEY"],
    base_url=os.environ.get("RAILWISE_API_URL"),
)

3.3 第一个请求

from railwise import RailWiseClient
from railwise.types import ProjectStatus

client = RailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

# 查询项目列表
projects = client.projects.list(
    status=ProjectStatus.ACTIVE,
    limit=10,
)

print(f"找到 {projects.total} 个活跃项目")

for project in projects.items:
    print(f"- {project.project_name} ({project.project_id})")

4. 核心 API

4.1 项目 API（Projects）

列出项目

from railwise import RailWiseClient
from railwise.types import ProjectStatus

client = RailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

projects = client.projects.list(
    status=ProjectStatus.ACTIVE,  # ACTIVE, PAUSED, COMPLETED, ALL
    limit=50,  # 1-200
    offset=0,  # 分页偏移
)

print(f"总计: {projects.total}")
print(f"是否有更多: {projects.has_more}")

for project in projects.items:
    print(f"{project.project_id}: {project.project_name}")

返回类型：ListProjectsResponse

from dataclasses import dataclass
from typing import List

@dataclass
class ListProjectsResponse:
    total: int
    items: List[Project]
    has_more: bool

@dataclass
class Project:
    project_id: str
    project_name: str
    project_type: str
    status: ProjectStatus
    location: Location

获取项目详情

project = client.projects.get(
    project_id="PRJ-A1B2C3D4",
    include_points=True,   # 包含测点信息
    include_devices=False, # 包含设备信息
)

print(f"项目名称: {project.project_name}")
print(f"项目类型: {project.project_type}")
print(f"状态: {project.status}")

if project.points:
    print(f"测点数量: {len(project.points)}")
    for point in project.points:
        print(f"  - {point.point_name} ({point.point_id})")

4.2 监测数据 API（Monitoring Data）

查询监测数据

from datetime import datetime, timezone
from railwise import RailWiseClient
from railwise.types import DataType, AggregationType

client = RailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

data = client.monitoring_data.query(
    project_id="PRJ-A1B2C3D4",
    point_ids=["PT-001", "PT-002"],  # 可选，为空查询所有测点
    data_type=DataType.SETTLEMENT,    # SETTLEMENT, DISPLACEMENT, CONVERGENCE, STRESS, ALL
    start_time=datetime(2025, 1, 14, 0, 0, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
    aggregation=AggregationType.HOURLY,  # RAW, HOURLY, DAILY, WEEKLY
    limit=1000,
)

print(f"总记录数: {data.total_records}")
print(f"正常测点: {data.summary.normal_count}")
print(f"黄色预警: {data.summary.yellow_warning_count}")
print(f"橙色预警: {data.summary.orange_warning_count}")
print(f"红色预警: {data.summary.red_warning_count}")

for point in data.points:
    print(f"\n测点: {point.point_name} ({point.point_id})")
    for record in point.records:
        print(f"  {record.timestamp}: {record.value} {record.unit}")

查询测量历史

history = client.monitoring_data.get_history(
    point_id="PT-001",
    start_time=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
    include_raw=False,  # 是否包含原始观测数据
)

4.3 设备 API（Devices）

查询设备状态

devices = client.devices.list(
    project_id="PRJ-A1B2C3D4",  # 可选
    device_ids=["DEV-TS-001"],   # 可选
)

print(f"设备总数: {devices.summary.total}")
print(f"在线: {devices.summary.online}")
print(f"离线: {devices.summary.offline}")
print(f"警告: {devices.summary.warning}")

for device in devices.devices:
    print(f"\n设备: {device.device_name}")
    print(f"  状态: {device.status}")
    print(f"  型号: {device.model}")
    print(f"  电量: {device.battery_level}%")
    print(f"  最后在线: {device.last_online}")

4.4 计算 API（Compute）

坐标转换

from railwise.types import CoordinateSystem, TransformMethod

result = client.compute.coordinate_transform(
    coordinates=[
        {"x": 384567.123, "y": 3156784.456, "z": 12.345},
        {"x": 384568.456, "y": 3156785.789, "z": 12.456},
    ],
    source_crs=CoordinateSystem.CGCS2000,
    target_crs=CoordinateSystem.LOCAL,
    transform_method=TransformMethod.SEVEN_PARAMETER,
    # 可选：自定义转换参数
    transform_params={
        "dx": 100.0,
        "dy": 200.0,
        "dz": 0.0,
        "rx": 0.0,
        "ry": 0.0,
        "rz": 0.0,
        "scale": 1.0,
    },
)

print(f"转换ID: {result.transform_id}")
print(f"转换方法: {result.transform_method}")
print(f"点数: {result.statistics.point_count}")
print(f"最大残差: {result.statistics.max_residual}")
print(f"RMS: {result.statistics.rms}")

for r in result.results:
    print(f"  源坐标: ({r.source.x}, {r.source.y}, {r.source.z})")
    print(f"  目标坐标: ({r.target.x}, {r.target.y}, {r.target.z})")
    print(f"  残差: ({r.residual.dx}, {r.residual.dy}, {r.residual.dz})")

平差计算

from railwise.types import AdjustmentType, ObservationType, UnknownType

result = client.compute.adjustment(
    adjustment_type=AdjustmentType.INDIRECT,
    observations=[
        {
            "obs_id": "OBS-001",
            "obs_type": ObservationType.DISTANCE,
            "value": 100.523,
            "precision": 0.002,
        },
        {
            "obs_id": "OBS-002",
            "obs_type": ObservationType.ANGLE,
            "value": 89.5234,
            "precision": 0.001,
        },
    ],
    unknowns=[
        {
            "unknown_id": "X1",
            "unknown_type": UnknownType.COORDINATE,
            "approximate_value": 1000.0,
        },
    ],
)

print(f"平差结果:")
print(f"  观测数: {result.obs_count}")
print(f"  未知数: {result.unknown_count}")
print(f"  单位权中误差: {result.sigma0}")

变形分析

from railwise.types import AnalysisType

analysis = client.compute.deformation_analysis(
    point_id="PT-001",
    analysis_type=AnalysisType.COMPREHENSIVE,
    start_time=datetime(2024, 6, 1, 0, 0, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
    forecast_days=7,  # 预测天数
)

print(f"测点: {analysis.point_id}")
print(f"分析类型: {analysis.analysis_type}")

print(f"\n累计变形:")
print(f"  当前值: {analysis.cumulative_deformation.current_value} {analysis.cumulative_deformation.unit}")
print(f"  最大值: {analysis.cumulative_deformation.max_value}")
print(f"  最小值: {analysis.cumulative_deformation.min_value}")

print(f"\n变形速率:")
print(f"  当前: {analysis.deformation_rate.current_rate} mm/天")
print(f"  平均: {analysis.deformation_rate.average_rate} mm/天")
print(f"  最大: {analysis.deformation_rate.max_rate} mm/天")

print(f"\n趋势分析:")
print(f"  方向: {analysis.trend.direction}")
print(f"  稳定性: {analysis.trend.stability}")
print(f"  置信度: {analysis.trend.confidence}")

if analysis.forecast:
    print(f"\n未来 {analysis.forecast.forecast_days} 天预测:")
    print(f"  预测值: {analysis.forecast.predicted_value}")
    print(f"  置信区间: [{analysis.forecast.confidence_interval.lower}, {analysis.forecast.confidence_interval.upper}]")

print(f"\n预警评估:")
print(f"  当前级别: {analysis.warning_assessment.current_level}")
print(f"  趋势级别: {analysis.warning_assessment.trend_level}")
print(f"  建议: {analysis.warning_assessment.recommendation}")

5. 异步 API

SDK 同时提供异步客户端，适用于高并发场景：

import asyncio
from railwise import AsyncRailWiseClient
from railwise.types import DataType, AggregationType
from datetime import datetime, timezone

async def fetch_multiple_projects():
    client = AsyncRailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

    async with client:  # 自动管理连接池
        # 并发查询多个项目
        tasks = [
            client.projects.get("PRJ-A1B2C3D4"),
            client.projects.get("PRJ-E5F6G7H8"),
            client.projects.get("PRJ-I9J0K1L2"),
        ]

        projects = await asyncio.gather(*tasks)

        for project in projects:
            print(f"{project.project_id}: {project.project_name}")

        # 并发查询数据
        data_tasks = [
            client.monitoring_data.query(
                project_id="PRJ-A1B2C3D4",
                data_type=DataType.SETTLEMENT,
                start_time=datetime(2025, 1, 14, 0, 0, 0, tzinfo=timezone.utc),
                end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
            ),
            client.monitoring_data.query(
                project_id="PRJ-E5F6G7H8",
                data_type=DataType.SETTLEMENT,
                start_time=datetime(2025, 1, 14, 0, 0, 0, tzinfo=timezone.utc),
                end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
            ),
        ]

        data_results = await asyncio.gather(*data_tasks)

        for data in data_results:
            print(f"项目 {data.project_id}: {data.total_records} 条记录")

# 运行
asyncio.run(fetch_multiple_projects())

6. 高级配置

6.1 自定义 HTTP 客户端

import httpx
from railwise import RailWiseClient

# 使用自定义 httpx 客户端
http_client = httpx.Client(
    timeout=60.0,
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    proxies={
        "http://": "http://proxy.company.com:8080",
        "https://": "http://proxy.company.com:8080",
    },
)

client = RailWiseClient(
    api_key="rw_live_xxxxxxxxxxxxxxxx",
    http_client=http_client,
)

6.2 请求拦截器

from railwise import RailWiseClient

client = RailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

# 请求前钩子
def on_request(request):
    print(f"发送请求: {request.method} {request.url}")
    # 添加自定义请求头
    request.headers["X-Request-ID"] = generate_request_id()
    return request

# 响应后钩子
def on_response(response):
    print(f"收到响应: {response.status_code}")
    return response

client.add_request_hook(on_request)
client.add_response_hook(on_response)

6.3 重试配置

from railwise import RailWiseClient

client = RailWiseClient(
    api_key="rw_live_xxxxxxxxxxxxxxxx",
    max_retries=5,
    retry_delay=1.0,  # 初始重试延迟（秒）
    retry_multiplier=2.0,  # 指数退避乘数
    retry_max_delay=30.0,  # 最大重试延迟
    # 自定义重试条件
    retry_condition=lambda error: error.status_code >= 500 or error.code == "TIMEOUT",
)

7. 错误处理

7.1 异常类型

from railwise import (
    RailWiseClient,
    RailWiseError,
    AuthenticationError,
    PermissionError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ServerError,
)

client = RailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

try:
    data = client.monitoring_data.query(
        project_id="PRJ-A1B2C3D4",
        data_type=DataType.SETTLEMENT,
        start_time=datetime(2025, 1, 14, 0, 0, 0, tzinfo=timezone.utc),
        end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
    )
except AuthenticationError as e:
    print(f"认证失败: {e.message}")
    print("请检查 API Key 是否有效")
except PermissionError as e:
    print(f"权限不足: {e.message}")
    print("请检查 API Key 的权限范围")
except NotFoundError as e:
    print(f"资源不存在: {e.message}")
    print("请检查项目ID或测点ID是否正确")
except ValidationError as e:
    print(f"请求参数错误: {e.message}")
    for detail in e.details:
        print(f"  字段 {detail.field}: {detail.reason}")
except RateLimitError as e:
    print(f"请求过于频繁: {e.message}")
    print(f"配额重置时间: {e.retry_after} 秒后")
    print(f"剩余配额: {e.remaining}/{e.limit}")
except ServerError as e:
    print(f"服务器错误: {e.message}")
    print("请稍后重试")
except RailWiseError as e:
    print(f"RailWise 错误: {e.message}")
    print(f"错误码: {e.code}")
    print(f"请求ID: {e.request_id}")
except Exception as e:
    print(f"未知错误: {e}")

7.2 异常类定义

class RailWiseError(Exception):
    """基础异常类"""
    def __init__(self, message: str, code: str, status: int, request_id: str, details=None):
        self.message = message
        self.code = code
        self.status = status
        self.request_id = request_id
        self.details = details
        super().__init__(message)

class AuthenticationError(RailWiseError):
    """认证失败（401）"""
    pass

class PermissionError(RailWiseError):
    """权限不足（403）"""
    pass

class NotFoundError(RailWiseError):
    """资源不存在（404）"""
    pass

class ValidationError(RailWiseError):
    """请求参数错误（400）"""
    pass

class RateLimitError(RailWiseError):
    """请求过于频繁（429）"""
    def __init__(self, *args, retry_after: int, limit: int, remaining: int, reset_time, **kwargs):
        self.retry_after = retry_after
        self.limit = limit
        self.remaining = remaining
        self.reset_time = reset_time
        super().__init__(*args, **kwargs)

class ServerError(RailWiseError):
    """服务器错误（500）"""
    pass

8. 完整示例

8.1 监测数据批量导出

import csv
import os
from datetime import datetime, timezone
from railwise import RailWiseClient
from railwise.types import DataType, AggregationType

def export_monitoring_data(
    project_id: str,
    start_date: datetime,
    end_date: datetime,
    output_path: str,
):
    client = RailWiseClient(api_key=os.environ["RAILWISE_API_KEY"])

    # 查询项目信息
    project = client.projects.get(project_id, include_points=True)
    print(f"导出项目: {project.project_name}")
    print(f"测点数量: {len(project.points) if project.points else 0}")

    # 查询监测数据
    data = client.monitoring_data.query(
        project_id=project_id,
        data_type=DataType.ALL,
        start_time=start_date,
        end_time=end_date,
        aggregation=AggregationType.DAILY,
    )

    # 写入 CSV
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["timestamp", "point_id", "point_name", "value", "unit", "change_rate", "warning_level"])

        for point in data.points:
            for record in point.records:
                writer.writerow([
                    record.timestamp.isoformat(),
                    point.point_id,
                    point.point_name,
                    record.value,
                    record.unit,
                    record.change_rate,
                    record.warning_level,
                ])

    print(f"数据已导出到: {output_path}")
    print(f"总记录数: {data.total_records}")
    print(f"预警统计: 正常={data.summary.normal_count}, 黄色={data.summary.yellow_warning_count}, "
          f"橙色={data.summary.orange_warning_count}, 红色={data.summary.red_warning_count}")

# 使用示例
export_monitoring_data(
    project_id="PRJ-A1B2C3D4",
    start_date=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
    end_date=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
    output_path="./monitoring-data.csv",
)

8.2 自动化预警检查脚本

import os
import time
from datetime import datetime, timezone, timedelta
from railwise import RailWiseClient
from railwise.types import DataType, AnalysisType

def check_warnings(project_id: str):
    client = RailWiseClient(api_key=os.environ["RAILWISE_API_KEY"])

    # 查询最近 24 小时数据
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=24)

    data = client.monitoring_data.query(
        project_id=project_id,
        data_type=DataType.SETTLEMENT,
        start_time=start_time,
        end_time=end_time,
    )

    # 筛选预警测点
    warning_points = []
    for point in data.points:
        if any(r.warning_level != "normal" for r in point.records):
            warning_points.append(point)

    if not warning_points:
        print("✅ 所有测点正常")
        return

    print(f"⚠️ 发现 {len(warning_points)} 个预警测点:")

    for point in warning_points:
        latest_record = point.records[-1]

        # 执行变形分析
        analysis = client.compute.deformation_analysis(
            point_id=point.point_id,
            analysis_type=AnalysisType.COMPREHENSIVE,
            start_time=end_time - timedelta(days=7),
            end_time=end_time,
        )

        print(f"\n测点: {point.point_name} ({point.point_id})")
        print(f"  当前值: {latest_record.value} {latest_record.unit}")
        print(f"  预警级别: {latest_record.warning_level}")
        print(f"  变化速率: {analysis.deformation_rate.current_rate} mm/天")
        print(f"  建议: {analysis.warning_assessment.recommendation}")

# 定时执行（每 4 小时）
if __name__ == "__main__":
    project_id = "PRJ-A1B2C3D4"

    while True:
        print(f"\n{'='*50}")
        print(f"预警检查: {datetime.now(timezone.utc).isoformat()}")
        print(f"{'='*50}")

        try:
            check_warnings(project_id)
        except Exception as e:
            print(f"检查失败: {e}")

        print(f"\n下次检查: 4小时后")
        time.sleep(4 * 60 * 60)  # 4 小时

8.3 Jupyter Notebook 数据分析

# 在 Jupyter Notebook 中使用
from railwise import RailWiseClient
from railwise.types import DataType
from datetime import datetime, timezone
import pandas as pd
import matplotlib.pyplot as plt

client = RailWiseClient(api_key="rw_live_xxxxxxxxxxxxxxxx")

# 查询数据
data = client.monitoring_data.query(
    project_id="PRJ-A1B2C3D4",
    data_type=DataType.SETTLEMENT,
    start_time=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 1, 15, 23, 59, 59, tzinfo=timezone.utc),
)

# 转换为 DataFrame
records = []
for point in data.points:
    for record in point.records:
        records.append({
            "timestamp": record.timestamp,
            "point_id": point.point_id,
            "point_name": point.point_name,
            "value": record.value,
            "warning_level": record.warning_level,
        })

df = pd.DataFrame(records)

# 绘制趋势图
plt.figure(figsize=(12, 6))
for point_id in df["point_id"].unique()[:5]:  # 前5个测点
    point_df = df[df["point_id"] == point_id]
    plt.plot(point_df["timestamp"], point_df["value"], label=point_id)

plt.xlabel("时间")
plt.ylabel("沉降量 (mm)")
plt.title("沉降监测趋势")
plt.legend()
plt.grid(True)
plt.show()

9. 类型提示

SDK 提供完整的类型提示，支持 IDE 智能补全和类型检查：

from railwise import RailWiseClient
from railwise.types import (
    Project,
    ProjectDetail,
    MonitoringPoint,
    MonitoringPointData,
    PointRecord,
    Device,
    DeviceStatus,
    DataType,
    AggregationType,
    WarningLevel,
    MonitoringDataResponse,
    CoordinateSystem,
    TransformMethod,
    AdjustmentType,
    AnalysisType,
    DeformationAnalysisResult,
    RailWiseError,
    AuthenticationError,
    # ...
)

# 类型检查器（如 mypy）可以验证类型正确性
# mypy --strict your_script.py

10. 相关文档

• TypeScript SDK — TypeScript/JavaScript 客户端库文档
• MCP Server 概述 — MCP Server 功能介绍
• MCP Server 安装配置 — MCP Server 安装步骤
• MCP Server 工具列表 — 完整的工具清单

---

本文档由 RailWise 技术文档团队维护，最后更新于 2025-01-15。