1. 矿山传感器原始数据加载与时间戳解析

任务边界与数据形态

本例只做一件事：把矿山传感器原始 CSV 读进 pandas，把字符串时间列转成 datetime64 并设为索引，同时把“解析失败”和“时间戳重复”这两类问题量化出来。后续的采样频率校验、去重策略都不在本例展开，但本例输出的 DatetimeIndex 是它们的前置条件。

矿山现场导出的 CSV 常见三类脏点：时间列格式不统一（2024-03-01 08:00:00 与 2024/03/01 08:00 混排）、存在空时间或 -- 占位符、同一传感器同一秒被网关重复上报。下面用一份可复现的教学样例数据走完整流程。

构造可复现的原始数据

先落一份 12 行的样例 CSV，故意埋入 1 行非法时间、1 行空时间、2 行重复时间戳，便于验证解析与统计逻辑：

import pandas as pd

raw = """sensor_id,timestamp,value
CH4-01,2024-03-01 08:00:00,0.42
CH4-01,2024-03-01 08:00:10,0.45
CH4-01,2024-03-01 08:00:20,0.44
CH4-01,2024/03/01 08:00:30,0.47
CH4-01,2024-03-01 08:00:40,0.46
CH4-01,2024-03-01 08:00:40,0.46
CH4-01,2024-03-01 08:00:50,0.48
CO-02,2024-03-01 08:00:00,18.2
CO-02,2024-03-01 08:00:10,18.5
CO-02,not_a_time,18.1
CO-02,,18.3
CO-02,2024-03-01 08:00:30,18.4
"""
with open("mine_sensors_raw.csv", "w", encoding="utf-8") as f:
    f.write(raw)

读取与时间戳解析

关键点有三个：dtype=str 先按字符串读入避免 pandas 自动推断把时间列变成 object 混杂类型；errors="coerce" 让非法值变成 NaT 而不是直接抛异常，从而能定位坏行；format="mixed" 允许同一列内 - 与 / 两种分隔符共存（pandas ≥ 2.0）。

df = pd.read_csv("mine_sensors_raw.csv", dtype=str)

df["timestamp_parsed"] = pd.to_datetime(
    df["timestamp"], errors="coerce", format="mixed"
)

# 解析失败行：原始非空但解析结果为 NaT
bad_mask = df["timestamp"].notna() & df["timestamp_parsed"].isna()
bad_rows = df.loc[bad_mask, ["sensor_id", "timestamp", "value"]]
print("解析失败行数:", len(bad_rows))
print(bad_rows)

# 空时间行单独统计
empty_mask = df["timestamp"].isna() | (df["timestamp"].str.strip() == "")
print("空时间行数:", int(empty_mask.sum()))

样例数据下预期输出：解析失败 1 行（not_a_time），空时间 1 行（CO-02 的空字段）。这两类要分开计数，因为处理动作不同——非法值通常回查网关日志，空值多为采集端丢包。

设为索引并统计重复时间戳

解析成功后丢弃原始字符串列，把 timestamp_parsed 设为索引并命名，便于后续按时间切片。重复统计要按“传感器 + 时间戳”组合判断，因为不同设备在同一秒上报是正常的：

df = df.dropna(subset=["timestamp_parsed"]).copy()
df = df.set_index("timestamp_parsed").rename_axis("ts")
df["value"] = pd.to_numeric(df["value"], errors="coerce")

dup_mask = df.index.duplicated(keep=False)
dup_detail = df.loc[dup_mask, ["sensor_id", "value"]]
print("重复时间戳记录数:", int(dup_mask.sum()))

# 按传感器+时间戳统计重复次数
dup_by_key = (
    df.reset_index()
      .groupby(["sensor_id", "ts"])
      .size()
      .loc[lambda s: s > 1]
)
print(dup_by_key)

样例中 CH4-01 在 08:00:40 出现两次，因此 dup_mask.sum() 为 2，dup_by_key 输出一行计数 2。注意 keep=False 会把重复组内所有记录都标出，便于整组回看；若只想看“多余的那条”，用 keep="first"。

结果判读与通过条件

本例的验收不看“跑通没报错”，而看四个数字是否与原始数据对得上：