Python图表无法正确显示CSV传感器数据问题求助

我正在开发一款应用，从连接3个传感器的芯片获取数据并每日保存为CSV文件（数据保存整体正常，仅偶尔莫名丢失几秒数据）。但在图表标签页展示数据时，时而正常，时而出现异常：明明记录时段有数据，却显示无数据、数值与时间错配、数值已增长却保持恒定等。

具体异常表现 #

• pH和温度值应从8:40开始上升且非连续（图表点频率为10秒/次），但显示为连续值；CSV显示除DO外数值并非恒定。
• 从开始到9:00应每30分钟有数据，图表无对应显示但CSV存在数据。
• 图表提前停止显示且数值错误，而CSV显示数据持续至23:59:59。

怀疑问题与get_today_data函数的时间日期格式化逻辑有关，相关代码如下：

class GraphMenu(Menu):
    def __init__(self, master,
                 title: str, func: Callable,
                 _type: Literal["4H", "Total", "24H"]):
        super().__init__(master=master, title=title)
        self.extra = lambda: self.create_graphs()
        self.get_data = func # 获取图表数据的函数
        self.graphs: list[FigureCanvasTkAgg] = [] 
        self._type = _type
        self.displayed_graph: ctk.StringVar = self.master.displayed_graph
        self.displayed_graph.trace_add("write", self.change_displayed_graph)
        self.create_widgets()

    def create_widgets(self) -> None:
…

    def create_graphs(self, instant_reload: bool = False) -> None:
        if not instant_reload:
            self.graph_frame.place_forget()
            self.loading_frame.place(relx=0, rely=0, relwidth=1, relheight=1)
            self.update()
        self.data = self.get_data()
        self.graph_frame.place(
            relx=0,
            rely=0.125 if self._type == "4H" else 0.075, 
            relwidth=1, 
            relheight=0.8 if self._type == "4H" else 0.85
        )
        
        self.graph_do.initialize_data(data={"time": self.data["time"], "values": self.data["values"]["DO"]})
        self.graph_ph.initialize_data(data={"time": self.data["time"], "values": self.data["values"]["pH"]})
        self.graph_temp.initialize_data(data={"time": self.data["time"], "values": self.data["values"]["Temperature"]})
        self.update()
        
    def get_estimated_time(self) -> float: return 1.5

    def get_today_data(self, graph_type: Literal["4H", "24H", "Total"],
                       selected_csv: str = None, data = None) -> dict[Literal["time", "values"], dict[Literal["DO", "pH", "Temperature"], list[float]]]:
        timestamps_sorting: list[Literal[0, 1]]
        data: dict[str, Union[list[datetime], dict[str, list[float]]]]
        if graph_type == "4H":
            if self.upper_edge.get() == "" or self.lower_edge.get() == "" :
                return 
        self.upper_edge: IntVar 
        self.lower_edge: IntVar
        
        if selected_csv is None:
            selected_csv = self.master.csv_filename.get()
        
        day_carry = 0
        if graph_type == "4H":
            current_time = dt.time(self.upper_edge.get()%24, 0, 0)
            if self.upper_edge.get() == 24: day_carry = 1
        else: current_time = dt.time(0, 0, 0)
                
        selected_csv_date = datetime.strptime(selected_csv, "%Y-%m-%d").date()
        
        now = datetime.combine(
            date=selected_csv_date + timedelta(days=day_carry), 
            time=current_time
        ) # 创建now对象，用于切换CSV时更新显示和日期
        now_str = now.strftime("%Y/%m/%d-%H:%M:%S") # 转换为日期时间字符串

        lower_hour_time = dt.time((self.lower_edge.get()%24 if graph_type == "4H" else 0), 0, 0) # 读取时间变量
        lower_hour = datetime.combine(
            date=selected_csv_date, 
            time=lower_hour_time
        )
        
        lower_hour_str = lower_hour.strftime("%Y/%m/%d-%H:%M:%S")

        day_data = self.master.load_data(self.master.csv_filenames[selected_csv]) # 获取当前CSV的数据
        day_data_values = list(day_data.values())[0]
        
        timestamps: list[datetime] = [datetime.combine(now, datetime.strptime(t, "%H:%M:%S").time()) for t in day_data_values["Timestamp"]]
        
        now += timedelta(days=1 if graph_type != "4H" else 0)
        now_str = now.strftime("%Y/%m/%d-%H:%M:%S")

        timestamps = pd.DatetimeIndex([t for t in timestamps if lower_hour <= t <= now])
        dates = pd.date_range(start=lower_hour_str, end=now_str, freq="s")

        timestamps_sorting = [] # 掩码
        i = 0
        j = 0
        
        if len(timestamps) > len(dates):
            while j < len(dates) and i < len(timestamps):
                recorded_timestamp = timestamps[j]
                check_timestamp = dates[i]
                
                if recorded_timestamp == check_timestamp:
                    timestamps_sorting.append(1)
                    i += 1 
                else:
                    timestamps_sorting.append(0)
                j += 1
                
        else:
            while i < len(dates) and j < len(timestamps):
                recorded_timestamp = timestamps[j]
                check_timestamp = dates[i]

                if recorded_timestamp == check_timestamp:
                    timestamps_sorting.append(1) 
                    j += 1
                else:
                    timestamps_sorting.append(0) 
                i += 1
                               
        if j == len(timestamps):
            timestamps_sorting.extend([0] * (len(dates) - i))
            
        if data is None:
            data = {
                    "time": [],
                    "values": {
                        "DO": [],
                        "pH": [],
                        "Temperature": []
                        }
                    }
        
        if graph_type == "Total":
            timestamps_sorting = timestamps_sorting[1:]
        lacking_data_offset = 0
        # 将时间和值的添加合并到一个循环中，带条件赋值
        for i, timestamp_sorting in enumerate(timestamps_sorting):
            data["time"].append(dates[i])
            if timestamp_sorting:
                data["values"]["DO"].append(day_data_values["DO"][i - lacking_data_offset])
                data["values"]["pH"].append(day_data_values["pH"][i - lacking_data_offset])
                data["values"]["Temperature"].append(day_data_values["Temperature"][i - lacking_data_offset])
            else:
                data["values"]["DO"].append(-1)
                data["values"]["pH"].append(-1)
                data["values"]["Temperature"].append(-1)
            lacking_data_offset += not timestamp_sorting  # 仅当数据缺失时增加偏移量

        return data

问题分析与修复方案 #

核心问题点 #

1. 时间戳组合错误：生成timestamps时用了基于upper_edge的now变量，导致CSV中的时间被错误关联到结束日期而非CSV本身的日期，后续过滤时合法数据被排除，引发数据丢失。
2. 时间范围逻辑混乱：生成timestamps后修改now值，导致过滤条件范围错误，部分数据被误过滤。
3. 掩码匹配逻辑缺陷：双指针匹配在时间戳和日期范围长度不一致时会出现错位，导致数值索引偏移，引发数值错配、恒定等异常。
4. Total模式掩码截断错误：直接截断掩码第一个元素但未同步调整dates和数据索引，导致时间与数值错位。

修复代码 #

def get_today_data(self, graph_type: Literal["4H", "24H", "Total"],
                   selected_csv: str = None, data = None) -> dict[Literal["time", "values"], dict[Literal["DO", "pH", "Temperature"], list[float]]]:
    # 4H模式下验证上下限合法性
    if graph_type == "4H":
        try:
            lower_val = self.lower_edge.get()
            upper_val = self.upper_edge.get()
            if lower_val is None or upper_val is None or lower_val >= upper_val:
                return {"time": [], "values": {"DO": [], "pH": [], "Temperature": []}}
        except:
            return {"time": [], "values": {"DO": [], "pH": [], "Temperature": []}}

    if selected_csv is None:
        selected_csv = self.master.csv_filename.get()
    
    selected_csv_date = datetime.strptime(selected_csv, "%Y-%m-%d").date()
    
    # 根据图表类型确定时间范围
    if graph_type == "4H":
        lower_hour = datetime.combine(selected_csv_date, dt.time(self.lower_edge.get()%24, 0, 0))
        upper_hour = datetime.combine(selected_csv_date, dt.time(self.upper_edge.get()%24, 0, 0))
        # 处理跨天的4H场景（如22:00到次日2:00）
        if upper_hour < lower_hour:
            upper_hour += timedelta(days=1)
    elif graph_type == "24H":
        lower_hour = datetime.combine(selected_csv_date, dt.time(0, 0, 0))
        upper_hour = lower_hour + timedelta(days=1)
    else: # Total模式，取CSV全量数据范围
        lower_hour = datetime.combine(selected_csv_date, dt.time(0, 0, 0))
        upper_hour = lower_hour + timedelta(days=1)

    # 加载CSV数据并生成正确的时间戳
    day_data = self.master.load_data(self.master.csv_filenames[selected_csv])
    day_data_values = list(day_data.values())[0]
    
    # 关联CSV自身日期与时间戳，避免日期错位
    timestamps = pd.DatetimeIndex([
        datetime.combine(selected_csv_date, datetime.strptime(t, "%H:%M:%S").time()) 
        for t in day_data_values["Timestamp"]
    ])
    
    # 过滤时间范围内的有效数据
    mask = (timestamps >= lower_hour) & (timestamps <= upper_hour)
    filtered_timestamps = timestamps[mask]
    filtered_do = [val for val, keep in zip(day_data_values["DO"], mask) if keep]
    filtered_ph = [val for val, keep in zip(day_data_values["pH"], mask) if keep]
    filtered_temp = [val for val, keep in zip(day_data_values["Temperature"], mask) if keep]

    # 生成连续时间轴（补全缺失秒数）
    dates = pd.date_range(start=lower_hour, end=upper_hour, freq="s")
    
    # 初始化数据结构
    data = data or {
        "time": [],
        "values": {
            "DO": [],
            "pH": [],
            "Temperature": []
        }
    }
    
    # 用字典映射快速匹配时间戳与对应数值
    timestamp_map = {ts: (do, ph, temp) for ts, do, ph, temp in zip(filtered_timestamps, filtered_do, filtered_ph, filtered_temp)}
    
    for date in dates:
        data["time"].append(date)
        if date in timestamp_map:
            do, ph, temp = timestamp_map[date]
            data["values"]["DO"].append(do)
            data["values"]["pH"].append(ph)
            data["values"]["Temperature"].append(temp)
        else:
            data["values"]["DO"].append(-1)
            data["values"]["pH"].append(-1)
            data["values"]["Temperature"].append(-1)
    
    return data

修复说明 #

1. 时间戳生成修正：直接使用CSV对应的日期组合时间戳，彻底避免日期错位问题。
2. 时间范围逻辑简化：明确不同模式的时间范围，处理4H模式跨天场景，确保过滤条件准确。
3. 掩码匹配优化：用Pandas布尔掩码过滤原始数据，再通过字典映射匹配连续时间轴，避免双指针逻辑的错位问题。
4. 移除错误截断：Total模式不再截断掩码，通过正确的时间范围过滤获取全量数据。
5. 增加输入验证：4H模式下检查上下限合法性，避免空值或非法范围导致的异常。

内容的提问来源于stack exchange，提问作者Cloud