ecg_history.py

import csv

import numpy as np

from ecg_constants import FS


class HistoryPlayer:
    def __init__(self):
        self.data = None
        self.playing = False
        self.speed = 1.0
        self.position = 0
        self.total_samples = 0
        self.file_path = None

    def load(self, path):
        with open(path, "r", encoding="utf-8") as f:
            rows = list(csv.DictReader(f))
        if not rows:
            return False
        self.file_path = path
        self.total_samples = len(rows)
        self.position = 0
        self.playing = False
        self.data = {
            "raw": np.array([float(r.get("raw", 0) or 0) for r in rows]),
            "filtered": np.array([float(r.get("filtered", 0) or 0) for r in rows]),
            "energy": np.array([float(r.get("energy", 0) or 0) for r in rows]),
            "hr": np.array([int(float(r.get("hr", 0) or 0)) for r in rows]),
            "beats": np.array([int(float(r.get("beats", 0) or 0)) for r in rows]),
            "leads_off": np.array([int(float(r.get("leads_off", 0) or 0)) for r in rows]),
        }
        return True

    @property
    def duration_s(self):
        return self.total_samples / FS if self.total_samples else 0

    @property
    def current_time_s(self):
        return self.position / FS

    def get_window(self, window_samples):
        if self.data is None:
            return None
        start = max(0, self.position - window_samples)
        end = self.position
        t = (np.arange(start, end) - self.position) / FS
        return {
            "t": t,
            "raw": self.data["raw"][start:end],
            "filtered": self.data["filtered"][start:end],
            "energy": self.data["energy"][start:end],
        }

    def advance(self, samples):
        self.position = min(self.position + samples, self.total_samples)
        if self.position >= self.total_samples:
            self.playing = False