label.py

import matplotlib.pyplot as plt
import numpy as np
from libPEAK import *
import os


# 64Hz
class RPPGLabel:
    def __init__(self, filename):
        self.filename = filename
        self.raw_sig = self._read_signal(filename)
        self.smooth_sig = self._smooth(self.raw_sig)


    def _read_signal(self, path):
        ret = []
        if os.path.exists(path):
            fh = open(path, 'r')
            line = fh.readline()

            while line:
                ret.append(float(line))
                line = fh.readline()

            fh.close()
            return ret
        else:
            return None

    def _smooth(self, sig):
        smooth_dat = side_pass(sig, 2, 64, 2, passtype='lowpass')
        return smooth_dat

    def plot_signal(self, sig):
        plt.plot(sig)
        plt.show()

    # 3-minute per video
    # bvp sensor 64Hz
    # camera 35.14 fps
    # every 1.8212862834376777 bvp signal per frame

    # index of inter-frame within a file
    def get_label_by_frame(self, iiframe, interval=5, diff_tole=10):
        bpf = 1.8212862834376777
        beg_bvp = int(round(iiframe * bpf))
        end_bvp = int(round((iiframe + interval) * bpf))
        full_diff = 0
        for i in range(beg_bvp, end_bvp):
            if i + 1 < len(self.smooth_sig):
                if self.smooth_sig[i] != None and self.smooth_sig[i + 1] != None:
                    full_diff += (self.smooth_sig[i + 1] - self.smooth_sig[i])
        
        if full_diff >= diff_tole:# number can be optimized
            return 1
        elif full_diff <= -1 * diff_tole:
            return 0
        else:
            return None
        
    # get labels at peaks and valleys
    # return frame number [[peaks], [valleys]]
    def get_frm_num_at_pvs(self):
        sec_per_sig = 1 / 64
        sec_per_frm = 1 / 35.14
        pk_frm_num = []
        vl_frm_num = []

        peakl, peakr, valleyl, valleyr =\
            find_valleys_peaks(self.smooth_sig)
        for onepeak in peakl:
            pk_frm_num.append(int(round(onepeak * sec_per_sig / sec_per_frm)))
        for onevalley in valleyl:
            vl_frm_num.append(int(round(onevalley * sec_per_sig / sec_per_frm)))

        return [vl_frm_num, pk_frm_num]

    def get_raw_size(self):
        return len(self.smooth_sig)

if __name__ == "__main__":
    rlable = RPPGLabel("G:\\ecg\\raw\\s2\\bvp_s2_T1.csv")
    # smooth_dat = _smooth(sig)
    rlable.plot_signal(rlable.raw_sig[128:1024])

    # all_labels = RPPGLabel().make_all_labels("F:/ecg")
    # z = 0