RailWise KB已发布

监测数据导入标准化模板

RailWise监测数据导入的标准化代码模板，支持全站仪、水准仪、自动化传感器等多源数据格式，实现数据清洗、校验与入库的完整流程。

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

template-doc

监测数据导入标准化模板

1. 模板概述

1.1 用途

本模板提供监测数据从原始采集到数据库入库的完整标准化流程，覆盖：

全站仪数据：Leica、Trimble、南方测绘等品牌原始观测文件
水准仪数据：电子水准仪观测数据、手工记录数据
自动化传感器数据：振弦式、压阻式、MEMS等传感器数据流
第三方系统数据：JSON/XML/CSV等通用交换格式

1.2 适用场景

场景	数据源	频率	数据量
自动化全站仪监测	TCP/串口实时流	1-4小时	中
人工测量数据录入	Excel/手簿文件	日/周	小
传感器网络接入	MQTT/HTTP API	分钟级	大
历史数据迁移	旧系统导出文件	一次性	大

1.3 输入输出

输入：原始观测文件、实时数据流、第三方接口数据
输出：标准化监测记录（数据库表/Parquet/CSV）
中间产物：清洗日志、校验报告、异常数据标记

2. 代码示例

2.1 核心数据导入类

"""
RailWise 监测数据导入引擎
支持多源异构监测数据的统一导入、清洗与校验

作者: RailWise技术部
版本: 1.0.0
"""

import csv
import json
import logging
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union

import pandas as pd
from pydantic import BaseModel, Field, validator


# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("RailWise.DataImport")


class DataSourceType(Enum):
    """数据源类型枚举"""
    TOTAL_STATION = "total_station"      # 全站仪
    LEVEL_INSTRUMENT = "level"           # 水准仪
    SENSOR_VIBRATING_WIRE = "vw_sensor"  # 振弦式传感器
    SENSOR_MEMS = "mems_sensor"          # MEMS传感器
    THIRD_PARTY_CSV = "csv"              # 第三方CSV
    THIRD_PARTY_JSON = "json"            # 第三方JSON


class DataQualityLevel(Enum):
    """数据质量等级"""
    EXCELLENT = "A"      # 优秀，可直接使用
    GOOD = "B"           # 良好，建议复核
    SUSPICIOUS = "C"     # 可疑，需人工确认
    INVALID = "D"        # 无效，丢弃或标记


@dataclass
class ImportResult:
    """导入结果容器"""
    success: bool
    records_imported: int = 0
    records_skipped: int = 0
    records_invalid: int = 0
    quality_summary: Dict[str, int] = field(default_factory=dict)
    errors: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)
    processing_time_ms: float = 0.0


class MonitoringRecord(BaseModel):
    """
    标准化监测记录模型
    所有数据源最终都转换为此统一格式
    """
    record_id: str = Field(..., description="记录唯一标识")
    project_id: str = Field(..., description="项目编号")
    point_id: str = Field(..., description="监测点编号")
    point_name: Optional[str] = Field(None, description="监测点名称")
    instrument_id: str = Field(..., description="仪器编号")
    observation_time: datetime = Field(..., description="观测时间(UTC)")
    source_type: DataSourceType = Field(..., description="数据源类型")

    # 观测值（根据监测类型不同，字段含义不同）
    value_x: Optional[float] = Field(None, description="X方向观测值(mm)")
    value_y: Optional[float] = Field(None, description="Y方向观测值(mm)")
    value_z: Optional[float] = Field(None, description="Z方向观测值(mm)")
    value_d: Optional[float] = Field(None, description="累计位移(mm)")
    value_rate: Optional[float] = Field(None, description="变化速率(mm/d)")

    # 原始观测值（保留原始数据）
    raw_data: Dict[str, Any] = Field(default_factory=dict, description="原始数据")

    # 元数据
    weather: Optional[str] = Field(None, description="天气状况")
    temperature: Optional[float] = Field(None, description="温度(℃)")
    operator: Optional[str] = Field(None, description="观测员")
    quality_level: DataQualityLevel = Field(DataQualityLevel.GOOD, description="数据质量等级")
    remarks: Optional[str] = Field(None, description="备注")

    @validator('observation_time')
    def ensure_timezone(cls, v):
        """确保时间带有时区信息"""
        if v.tzinfo is None:
            return v.replace(tzinfo=timezone.utc)
        return v

    @validator('value_x', 'value_y', 'value_z', 'value_d', 'value_rate')
    def validate_precision(cls, v):
        """验证数值精度（保留到0.01mm）"""
        if v is not None:
            return round(v, 2)
        return v

    class Config:
        use_enum_values = True


class DataImporter(ABC):
    """数据导入器抽象基类"""

    def __init__(self, project_id: str, instrument_id: str):
        self.project_id = project_id
        self.instrument_id = instrument_id
        self.logger = logging.getLogger(self.__class__.__name__)

    @abstractmethod
    def parse(self, source: Union[str, Path, bytes]) -> List[MonitoringRecord]:
        """解析原始数据为标准化记录"""
        pass

    @abstractmethod
    def validate_format(self, source: Union[str, Path, bytes]) -> bool:
        """验证数据源格式是否正确"""
        pass

    def preprocess(self, records: List[MonitoringRecord]) -> List[MonitoringRecord]:
        """数据预处理：去重、排序、基础清洗"""
        # 按时间排序
        records.sort(key=lambda r: r.observation_time)

        # 去重（基于point_id + observation_time）
        seen = set()
        unique_records = []
        for rec in records:
            key = (rec.point_id, rec.observation_time.isoformat())
            if key not in seen:
                seen.add(key)
                unique_records.append(rec)
            else:
                self.logger.warning(f"重复记录已跳过: {key}")

        return unique_records

    def quality_check(self, records: List[MonitoringRecord]) -> List[MonitoringRecord]:
        """数据质量检查"""
        checked = []
        for rec in records:
            issues = []

            # 检查时间合理性
            now = datetime.now(timezone.utc)
            if rec.observation_time > now:
                issues.append("观测时间在未来")
            if (now - rec.observation_time).days > 365:
                issues.append("观测时间超过一年前")

            # 检查数值合理性
            for field_name in ['value_x', 'value_y', 'value_z', 'value_d']:
                val = getattr(rec, field_name)
                if val is not None and abs(val) > 10000:
                    issues.append(f"{field_name}值异常过大: {val}")

            # 根据问题数量评定质量等级
            if len(issues) == 0:
                rec.quality_level = DataQualityLevel.EXCELLENT
            elif len(issues) <= 1:
                rec.quality_level = DataQualityLevel.GOOD
            elif len(issues) <= 2:
                rec.quality_level = DataQualityLevel.SUSPICIOUS
            else:
                rec.quality_level = DataQualityLevel.INVALID

            if issues:
                rec.remarks = "; ".join(issues)

            checked.append(rec)

        return checked

    def import_data(self, source: Union[str, Path, bytes]) -> ImportResult:
        """完整导入流程"""
        import time
        start_time = time.time()

        result = ImportResult(success=False)

        try:
            # 1. 格式验证
            if not self.validate_format(source):
                result.errors.append("数据源格式验证失败")
                return result

            # 2. 解析数据
            records = self.parse(source)
            result.records_imported = len(records)

            # 3. 预处理
            records = self.preprocess(records)

            # 4. 质量检查
            records = self.quality_check(records)

            # 5. 统计质量分布
            for rec in records:
                level = rec.quality_level.value
                result.quality_summary[level] = result.quality_summary.get(level, 0) + 1
                if rec.quality_level == DataQualityLevel.INVALID:
                    result.records_invalid += 1

            result.success = True
            result.processing_time_ms = (time.time() - start_time) * 1000

            self.logger.info(
                f"导入完成: {result.records_imported}条记录, "
                f"无效: {result.records_invalid}, "
                f"耗时: {result.processing_time_ms:.2f}ms"
            )

        except Exception as e:
            result.errors.append(str(e))
            self.logger.error(f"导入失败: {e}")

        return result


class TotalStationImporter(DataImporter):
    """全站仪数据导入器"""

    def validate_format(self, source: Union[str, Path, bytes]) -> bool:
        """验证全站仪数据格式（支持GSI、TXT、CSV格式）"""
        try:
            content = self._read_source(source)
            # 检查是否包含全站仪特征字段
            indicators = ['Hz', 'V', 'SD', 'HD', 'N', 'E', 'Z']
            return any(ind in content for ind in indicators)
        except Exception:
            return False

    def parse(self, source: Union[str, Path, bytes]) -> List[MonitoringRecord]:
        """解析全站仪观测数据"""
        content = self._read_source(source)
        records = []

        # 尝试按行解析（支持多种格式）
        lines = content.strip().split('\n')

        for line in lines:
            line = line.strip()
            if not line or line.startswith('#'):
                continue

            try:
                # 尝试解析GSI格式: *110001+0000000000000000
                if line.startswith('*'):
                    record = self._parse_gsi_line(line)
                # 尝试解析CSV格式
                elif ',' in line:
                    record = self._parse_csv_line(line)
                else:
                    continue

                if record:
                    records.append(record)

            except Exception as e:
                self.logger.warning(f"解析行失败: {line[:50]}... 错误: {e}")

        return records

    def _parse_gsi_line(self, line: str) -> Optional[MonitoringRecord]:
        """解析Leica GSI格式数据"""
        # GSI格式简化解析示例
        # 实际项目中需根据具体仪器型号调整
        match = re.match(r'\*(\d{2})(\d{4})([+-]\d+)', line)
        if not match:
            return None

        word_index = match.group(1)
        point_id = match.group(2)
        value = float(match.group(3)) / 10000  # 转换为mm

        return MonitoringRecord(
            record_id=f"{self.project_id}_{point_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}",
            project_id=self.project_id,
            point_id=point_id,
            instrument_id=self.instrument_id,
            observation_time=datetime.now(timezone.utc),
            source_type=DataSourceType.TOTAL_STATION,
            value_x=value if word_index == '11' else None,
            value_y=value if word_index == '12' else None,
            value_z=value if word_index == '13' else None,
            raw_data={"gsi_line": line, "word_index": word_index}
        )

    def _parse_csv_line(self, line: str) -> Optional[MonitoringRecord]:
        """解析CSV格式全站仪数据"""
        parts = line.split(',')
        if len(parts) < 4:
            return None

        point_id = parts[0].strip()
        obs_time = datetime.fromisoformat(parts[1].strip())

        return MonitoringRecord(
            record_id=f"{self.project_id}_{point_id}_{obs_time.strftime('%Y%m%d%H%M%S')}",
            project_id=self.project_id,
            point_id=point_id,
            instrument_id=self.instrument_id,
            observation_time=obs_time.replace(tzinfo=timezone.utc),
            source_type=DataSourceType.TOTAL_STATION,
            value_x=float(parts[2]) if len(parts) > 2 else None,
            value_y=float(parts[3]) if len(parts) > 3 else None,
            value_z=float(parts[4]) if len(parts) > 4 else None,
            raw_data={"csv_line": line}
        )

    def _read_source(self, source: Union[str, Path, bytes]) -> str:
        """统一读取数据源"""
        if isinstance(source, bytes):
            return source.decode('utf-8', errors='ignore')
        elif isinstance(source, (str, Path)):
            return Path(source).read_text(encoding='utf-8', errors='ignore')
        return str(source)


class LevelInstrumentImporter(DataImporter):
    """水准仪数据导入器"""

    def validate_format(self, source: Union[str, Path, bytes]) -> bool:
        """验证水准仪数据格式"""
        try:
            content = self._read_source(source)
            return any(kw in content for kw in ['后视', '前视', '高差', 'BM', '水准'])
        except Exception:
            return False

    def parse(self, source: Union[str, Path, bytes]) -> List[MonitoringRecord]:
        """解析水准仪观测数据"""
        content = self._read_source(source)
        records = []

        lines = content.strip().split('\n')
        for line in lines:
            line = line.strip()
            if not line or line.startswith('#'):
                continue

            try:
                parts = line.split(',')
                if len(parts) >= 4:
                    point_id = parts[0].strip()
                    obs_time = datetime.fromisoformat(parts[1].strip())
                    elevation = float(parts[2])
                    settlement = float(parts[3])

                    record = MonitoringRecord(
                        record_id=f"{self.project_id}_{point_id}_{obs_time.strftime('%Y%m%d%H%M%S')}",
                        project_id=self.project_id,
                        point_id=point_id,
                        instrument_id=self.instrument_id,
                        observation_time=obs_time.replace(tzinfo=timezone.utc),
                        source_type=DataSourceType.LEVEL_INSTRUMENT,
                        value_z=settlement,
                        raw_data={"elevation": elevation, "csv_line": line}
                    )
                    records.append(record)
            except Exception as e:
                self.logger.warning(f"解析水准数据行失败: {e}")

        return records


class SensorDataImporter(DataImporter):
    """传感器数据导入器（支持JSON/CSV格式）"""

    def validate_format(self, source: Union[str, Path, bytes]) -> bool:
        """验证传感器数据格式"""
        try:
            content = self._read_source(source)
            # 检查JSON或包含传感器字段的CSV
            if content.strip().startswith('{'):
                data = json.loads(content)
                return any(k in data for k in ['sensor_id', 'frequency', 'temperature', 'value'])
            else:
                return 'sensor' in content.lower() or 'frequency' in content.lower()
        except Exception:
            return False

    def parse(self, source: Union[str, Path, bytes]) -> List[MonitoringRecord]:
        """解析传感器数据"""
        content = self._read_source(source)
        records = []

        if content.strip().startswith('[') or content.strip().startswith('{'):
            data = json.loads(content)
            if isinstance(data, list):
                for item in data:
                    records.extend(self._parse_sensor_json(item))
            else:
                records.extend(self._parse_sensor_json(data))
        else:
            # CSV格式
            lines = content.strip().split('\n')
            reader = csv.DictReader(lines)
            for row in reader:
                records.extend(self._parse_sensor_csv(row))

        return records

    def _parse_sensor_json(self, data: Dict) -> List[MonitoringRecord]:
        """解析JSON格式传感器数据"""
        records = []
        sensor_id = data.get('sensor_id', 'UNKNOWN')
        point_id = data.get('point_id', sensor_id)
        obs_time = datetime.fromisoformat(data.get('timestamp', datetime.now().isoformat()))

        # 振弦式传感器
        if 'frequency' in data:
            record = MonitoringRecord(
                record_id=f"{self.project_id}_{point_id}_{obs_time.strftime('%Y%m%d%H%M%S')}",
                project_id=self.project_id,
                point_id=point_id,
                instrument_id=self.instrument_id,
                observation_time=obs_time.replace(tzinfo=timezone.utc),
                source_type=DataSourceType.SENSOR_VIBRATING_WIRE,
                value_d=data.get('value'),
                raw_data=data
            )
            records.append(record)

        # MEMS传感器
        elif 'acceleration_x' in data or 'tilt_x' in data:
            record = MonitoringRecord(
                record_id=f"{self.project_id}_{point_id}_{obs_time.strftime('%Y%m%d%H%M%S')}",
                project_id=self.project_id,
                point_id=point_id,
                instrument_id=self.instrument_id,
                observation_time=obs_time.replace(tzinfo=timezone.utc),
                source_type=DataSourceType.SENSOR_MEMS,
                value_x=data.get('tilt_x') or data.get('acceleration_x'),
                value_y=data.get('tilt_y') or data.get('acceleration_y'),
                value_z=data.get('tilt_z') or data.get('acceleration_z'),
                raw_data=data
            )
            records.append(record)

        return records

    def _parse_sensor_csv(self, row: Dict) -> List[MonitoringRecord]:
        """解析CSV格式传感器数据"""
        records = []
        point_id = row.get('point_id', row.get('sensor_id', 'UNKNOWN'))
        obs_time = datetime.fromisoformat(row.get('timestamp', datetime.now().isoformat()))

        record = MonitoringRecord(
            record_id=f"{self.project_id}_{point_id}_{obs_time.strftime('%Y%m%d%H%M%S')}",
            project_id=self.project_id,
            point_id=point_id,
            instrument_id=self.instrument_id,
            observation_time=obs_time.replace(tzinfo=timezone.utc),
            source_type=DataSourceType.SENSOR_VIBRATING_WIRE,
            value_d=float(row.get('value', 0)) if row.get('value') else None,
            raw_data=dict(row)
        )
        records.append(record)
        return records


# 工厂函数
def create_importer(
    source_type: DataSourceType,
    project_id: str,
    instrument_id: str
) -> DataImporter:
    """创建对应类型的导入器"""
    importers = {
        DataSourceType.TOTAL_STATION: TotalStationImporter,
        DataSourceType.LEVEL_INSTRUMENT: LevelInstrumentImporter,
        DataSourceType.SENSOR_VIBRATING_WIRE: SensorDataImporter,
        DataSourceType.SENSOR_MEMS: SensorDataImporter,
    }

    importer_class = importers.get(source_type)
    if not importer_class:
        raise ValueError(f"不支持的数据源类型: {source_type}")

    return importer_class(project_id, instrument_id)


# 批量导入函数
def batch_import(
    file_paths: List[Path],
    project_id: str,
    instrument_id: str,
    source_type: DataSourceType
) -> Dict[str, ImportResult]:
    """
    批量导入多个文件

    Args:
        file_paths: 文件路径列表
        project_id: 项目编号
        instrument_id: 仪器编号
        source_type: 数据源类型

    Returns:
        每个文件的导入结果字典
    """
    importer = create_importer(source_type, project_id, instrument_id)
    results = {}

    for file_path in file_paths:
        logger.info(f"开始导入: {file_path}")
        result = importer.import_data(file_path)
        results[str(file_path)] = result

    return results

2.2 数据库存储模块

"""数据库存储模块 - 将标准化记录写入数据库"""

import sqlite3
from contextlib import contextmanager
from typing import List, Optional


class MonitoringDatabase:
    """监测数据库操作类"""

    def __init__(self, db_path: str = "monitoring.db"):
        self.db_path = db_path
        self._init_tables()

    @contextmanager
    def _get_connection(self):
        """获取数据库连接上下文"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def _init_tables(self):
        """初始化数据库表结构"""
        with self._get_connection() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS monitoring_records (
                    record_id TEXT PRIMARY KEY,
                    project_id TEXT NOT NULL,
                    point_id TEXT NOT NULL,
                    point_name TEXT,
                    instrument_id TEXT NOT NULL,
                    observation_time TIMESTAMP NOT NULL,
                    source_type TEXT NOT NULL,
                    value_x REAL,
                    value_y REAL,
                    value_z REAL,
                    value_d REAL,
                    value_rate REAL,
                    raw_data TEXT,
                    weather TEXT,
                    temperature REAL,
                    operator TEXT,
                    quality_level TEXT DEFAULT 'B',
                    remarks TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)

            # 创建索引
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_records_project
                ON monitoring_records(project_id)
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_records_point_time
                ON monitoring_records(point_id, observation_time)
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_records_quality
                ON monitoring_records(quality_level)
            """)

    def insert_records(self, records: List[MonitoringRecord]) -> int:
        """批量插入记录"""
        with self._get_connection() as conn:
            cursor = conn.executemany("""
                INSERT OR REPLACE INTO monitoring_records (
                    record_id, project_id, point_id, point_name,
                    instrument_id, observation_time, source_type,
                    value_x, value_y, value_z, value_d, value_rate,
                    raw_data, weather, temperature, operator,
                    quality_level, remarks
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, [
                (
                    r.record_id, r.project_id, r.point_id, r.point_name,
                    r.instrument_id, r.observation_time.isoformat(), r.source_type.value,
                    r.value_x, r.value_y, r.value_z, r.value_d, r.value_rate,
                    json.dumps(r.raw_data), r.weather, r.temperature, r.operator,
                    r.quality_level.value, r.remarks
                )
                for r in records
            ])
            return cursor.rowcount

    def query_by_point(
        self,
        point_id: str,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
        min_quality: DataQualityLevel = DataQualityLevel.GOOD
    ) -> List[MonitoringRecord]:
        """按监测点查询记录"""
        with self._get_connection() as conn:
            query = """
                SELECT * FROM monitoring_records
                WHERE point_id = ? AND quality_level <= ?
            """
            params = [point_id, min_quality.value]

            if start_time:
                query += " AND observation_time >= ?"
                params.append(start_time.isoformat())
            if end_time:
                query += " AND observation_time <= ?"
                params.append(end_time.isoformat())

            query += " ORDER BY observation_time"

            cursor = conn.execute(query, params)
            rows = cursor.fetchall()

            return [self._row_to_record(row) for row in rows]

    def _row_to_record(self, row: sqlite3.Row) -> MonitoringRecord:
        """将数据库行转换为记录对象"""
        return MonitoringRecord(
            record_id=row['record_id'],
            project_id=row['project_id'],
            point_id=row['point_id'],
            point_name=row['point_name'],
            instrument_id=row['instrument_id'],
            observation_time=datetime.fromisoformat(row['observation_time']),
            source_type=DataSourceType(row['source_type']),
            value_x=row['value_x'],
            value_y=row['value_y'],
            value_z=row['value_z'],
            value_d=row['value_d'],
            value_rate=row['value_rate'],
            raw_data=json.loads(row['raw_data']) if row['raw_data'] else {},
            weather=row['weather'],
            temperature=row['temperature'],
            operator=row['operator'],
            quality_level=DataQualityLevel(row['quality_level']),
            remarks=row['remarks']
        )

3. 参数说明

3.1 MonitoringRecord 模型参数

参数名	类型	必填	默认值	说明
`record_id`	str	是	-	记录唯一标识，建议格式：`{项目编号}_{测点编号}_{时间戳}`
`project_id`	str	是	-	项目编号，如“RW-2024-001”
`point_id`	str	是	-	监测点编号，如“JC-01”
`point_name`	str	否	None	监测点名称，如“3号线K12+350左线”
`instrument_id`	str	是	-	仪器编号，如“TS-001”
`observation_time`	datetime	是	-	观测时间，必须带时区信息
`source_type`	DataSourceType	是	-	数据源类型枚举
`value_x`	float	否	None	X方向位移(mm)，保留2位小数
`value_y`	float	否	None	Y方向位移(mm)，保留2位小数
`value_z`	float	否	None	Z方向位移/沉降(mm)，保留2位小数
`value_d`	float	否	None	累计位移(mm)
`value_rate`	float	否	None	变化速率(mm/d)
`raw_data`	dict	否	{}	原始数据保留，用于追溯
`weather`	str	否	None	天气状况
`temperature`	float	否	None	环境温度(℃)
`operator`	str	否	None	观测员姓名
`quality_level`	DataQualityLevel	否	GOOD	数据质量等级
`remarks`	str	否	None	备注信息

3.2 导入器配置参数

参数名	类型	说明
`project_id`	str	项目唯一标识
`instrument_id`	str	仪器唯一标识
`db_path`	str	数据库文件路径，默认`monitoring.db`

4. 使用示例

4.1 基本使用流程

from pathlib import Path
from template_data_import import (
    create_importer,
    DataSourceType,
    MonitoringDatabase
)

# 1. 创建导入器
importer = create_importer(
    source_type=DataSourceType.TOTAL_STATION,
    project_id="RW-2024-001",
    instrument_id="TS-001"
)

# 2. 导入数据
result = importer.import_data(Path("./data/20240708_obs.csv"))

print(f"导入成功: {result.success}")
print(f"导入记录数: {result.records_imported}")
print(f"无效记录数: {result.records_invalid}")
print(f"质量分布: {result.quality_summary}")
print(f"处理耗时: {result.processing_time_ms:.2f}ms")

# 3. 写入数据库
db = MonitoringDatabase("project.db")
records = importer.parse(Path("./data/20240708_obs.csv"))
records = importer.preprocess(records)
records = importer.quality_check(records)

# 过滤有效数据
valid_records = [r for r in records if r.quality_level.value != 'D']
inserted = db.insert_records(valid_records)
print(f"成功写入数据库: {inserted}条")

4.2 批量导入多个文件

from pathlib import Path
from template_data_import import batch_import, DataSourceType

# 获取所有数据文件
data_files = list(Path("./data/").glob("*.csv"))

# 批量导入
results = batch_import(
    file_paths=data_files,
    project_id="RW-2024-001",
    instrument_id="TS-001",
    source_type=DataSourceType.TOTAL_STATION
)

# 输出汇总报告
total_imported = sum(r.records_imported for r in results.values() if r.success)
total_invalid = sum(r.records_invalid for r in results.values() if r.success)

print(f"批量导入完成: {total_imported}条成功, {total_invalid}条无效")

# 输出失败文件
for path, result in results.items():
    if not result.success:
        print(f"导入失败: {path}")
        for error in result.errors:
            print(f"  错误: {error}")

4.3 自定义导入器

from template_data_import import DataImporter, MonitoringRecord, DataSourceType

class CustomLaserScannerImporter(DataImporter):
    """自定义激光扫描仪数据导入器"""

    def validate_format(self, source):
        content = self._read_source(source)
        return "LASER" in content or "SCAN" in content

    def parse(self, source):
        # 实现自定义解析逻辑
        content = self._read_source(source)
        records = []
        # ... 解析逻辑 ...
        return records

    def _read_source(self, source):
        # 继承或重写读取方法
        if isinstance(source, bytes):
            return source.decode('utf-8')
        return Path(source).read_text()

# 注册到工厂
from template_data_import import create_importer

# 使用自定义导入器
importer = CustomLaserScannerImporter("RW-2024-001", "LS-001")
result = importer.import_data("./data/laser_scan.txt")

5. 输出示例

5.1 导入结果示例

导入成功: True
导入记录数: 156
无效记录数: 3
质量分布: {'A': 142, 'B': 11, 'C': 3, 'D': 0}
处理耗时: 245.67ms

警告信息:
  - 重复记录已跳过: ('JC-01', '2024-07-08T10:00:00')
  - 解析行失败: *110001+0000000000000000... 错误: 格式不匹配

5.2 数据库查询示例

# 查询某测点最近7天的数据
from datetime import datetime, timedelta

records = db.query_by_point(
    point_id="JC-01",
    start_time=datetime.now(timezone.utc) - timedelta(days=7),
    min_quality=DataQualityLevel.GOOD
)

for r in records:
    print(f"{r.observation_time}: X={r.value_x}, Y={r.value_y}, Z={r.value_z}")

输出：

2024-07-01 08:00:00+00:00: X=0.12, Y=-0.05, Z=0.00
2024-07-01 12:00:00+00:00: X=0.15, Y=-0.03, Z=0.01
2024-07-01 16:00:00+00:00: X=0.18, Y=-0.02, Z=0.01
...

6. 扩展指南

6.1 添加新的数据源类型

定义新的枚举值：

class DataSourceType(Enum):
    # ... 现有类型 ...
    GNSS_RECEIVER = "gnss"  # GNSS接收机

创建导入器类：

class GNSSImporter(DataImporter):
    def validate_format(self, source):
        # 验证RINEX或NMEA格式
        pass

    def parse(self, source):
        # 解析GNSS数据
        pass

注册到工厂：

importers[DataSourceType.GNSS_RECEIVER] = GNSSImporter

6.2 自定义质量检查规则

class StrictQualityImporter(TotalStationImporter):
    """严格质量检查导入器"""

    def quality_check(self, records):
        checked = super().quality_check(records)

        # 添加自定义规则
        for rec in checked:
            # 检查变化速率是否超过阈值
            if rec.value_rate and abs(rec.value_rate) > 5.0:
                rec.quality_level = DataQualityLevel.SUSPICIOUS
                rec.remarks = (rec.remarks or "") + "; 变化速率异常"

        return checked

6.3 集成到工作流

# 与RAILWISE-CLI集成
from railwise_cli import Pipeline

pipeline = Pipeline(project_id="RW-2024-001")
pipeline.add_step("import", importer.import_data)
pipeline.add_step("validate", validate_against_thresholds)
pipeline.add_step("store", db.insert_records)
pipeline.add_step("notify", send_alert_if_needed)

pipeline.run(source="./data/incoming/")

7. 常见问题

Q1: 如何处理编码问题？

A: 模板默认使用UTF-8编码，如遇GBK编码文件：

# 在自定义导入器中重写读取方法
def _read_source(self, source):
    if isinstance(source, (str, Path)):
        # 尝试多种编码
        for encoding in ['utf-8', 'gbk', 'gb2312']:
            try:
                return Path(source).read_text(encoding=encoding)
            except UnicodeDecodeError:
                continue
    return str(source)

Q2: 大数据量导入内存不足怎么办？

A: 使用流式处理：

import pandas as pd

# 分块读取CSV
chunk_size = 10000
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
    records = parse_chunk(chunk)  # 自定义解析
    db.insert_records(records)     # 分批写入

Q3: 如何支持实时数据流？

A: 结合消息队列使用：

import asyncio
from aio_pika import connect_robust

async def consume_mqtt():
    connection = await connect_robust("amqp://guest:guest@localhost/")
    channel = await connection.channel()
    queue = await channel.declare_queue("sensor_data")

    async with queue.iterator() as queue_iter:
        async for message in queue_iter:
            async with message.process():
                data = json.loads(message.body)
                records = importer.parse(data)
                db.insert_records(records)

Q4: 导入时如何保留原始文件？

A: 建议建立归档机制：

import shutil
from datetime import datetime

def archive_source(file_path: Path, archive_dir: Path):
    """归档原始文件"""
    today = datetime.now().strftime("%Y%m%d")
    archive_path = archive_dir / today / file_path.name
    archive_path.parent.mkdir(parents=True, exist_ok=True)
    shutil.move(str(file_path), str(archive_path))
    return archive_path

Q5: 如何与现有系统（如WorkWise）集成？

A: 使用API接口推送：

import requests

def push_to_workwise(records: List[MonitoringRecord], api_key: str):
    """推送数据到WorkWise平台"""
    url = "https://api.workwise.railwise.com/v1/monitoring/data"
    headers = {"Authorization": f"Bearer {api_key}"}

    payload = [r.dict() for r in records]
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

元数据标签

ai_tags:
  domain: ["工程监测", "数据管理", "测量数据处理"]
  technology: ["Python", "SQLite", "Pydantic", "Pandas"]
  standard: ["GB 50497", "CJJ/T 202", "JGJ 8"]
  product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"]
  difficulty: "intermediate"
  role: ["监测工程师", "数据分析师", "软件开发工程师"]
  task_type: ["数据导入", "数据清洗", "数据存储"]

监测数据导入标准化模板

监测数据导入标准化模板

1. 模板概述

1.1 用途

1.2 适用场景

1.3 输入输出

2. 代码示例

2.1 核心数据导入类

2.2 数据库存储模块

3. 参数说明

3.1 MonitoringRecord 模型参数

3.2 导入器配置参数

4. 使用示例

4.1 基本使用流程

4.2 批量导入多个文件

4.3 自定义导入器

5. 输出示例

5.1 导入结果示例

5.2 数据库查询示例

6. 扩展指南

6.1 添加新的数据源类型

6.2 自定义质量检查规则

6.3 集成到工作流

7. 常见问题

Q1: 如何处理编码问题？

Q2: 大数据量导入内存不足怎么办？

Q3: 如何支持实时数据流？

Q4: 导入时如何保留原始文件？

Q5: 如何与现有系统（如WorkWise）集成？

相关文档链接

元数据标签