fixtracks/fixtracks/utils/reader.py

106 lines
2.9 KiB
Python

import logging
import pickle
import cv2 as cv
import pandas as pd
from PySide6.QtCore import QRunnable, Signal, Slot
from fixtracks.utils.signals import ProducerSignals
class ImageReader(QRunnable):
finished = Signal(bool)
def __init__(self, filename, frame=1000) -> None:
super().__init__()
self._filename = filename
self._framenumber = frame
self._signals = ProducerSignals()
self._frame = None
@Slot()
def run(self):
'''
Your code goes in this function
'''
logging.debug("ImageReader: trying to open file %s", self._filename)
cap = cv.VideoCapture(self._filename)
framecount = int(cap.get(cv.CAP_PROP_FRAME_COUNT))
if self._framenumber >= framecount:
logging.warning("ImageReader: desired frame number (%i) exceeds the frame count (%i)! Redefined to frame count." % (self._framenumber, framecount))
if not cap.isOpened():
logging.debug("ImageReader: failed to open file %s", self._filename)
self._signals.finished.emit(False)
fn = 0
while cap.isOpened() and fn < self._framenumber:
ret, frame = cap.read()
if not ret:
logging.warning("ImageReader: failed to read frame %i", fn)
self._signals.finished.emit(False)
break
fn += 1
self._frame = frame # cv.cvtColor(frame, cv.COLOR_BGR2RGB)
cap.release()
self._signals.finished.emit(True)
@property
def signals(self):
return self._signals
@property
def frame(self):
return self._frame
class PickleLoader(QRunnable):
finished = Signal(bool)
def __init__(self, filename) -> None:
super().__init__()
self._filename = filename
self._signals = ProducerSignals()
self._data = None
@Slot()
def run(self):
'''
Your code goes in this function
'''
logging.debug("PickleLoader: trying to open file %s", self._filename)
with open(self._filename, "rb") as f:
self._data = pickle.load(f)
self._signals.finished.emit(True)
@property
def signals(self):
return self._signals
@property
def data(self):
return self._data
class DataFrameReader(QRunnable):
finished = Signal(bool)
def __init__(self, filename) -> None:
super().__init__()
self._filename = filename
self._signals = ProducerSignals()
self._dataframe = None
@Slot()
def run(self):
'''
Your code goes in this function
'''
logging.debug("DataFrameReader: trying to open file %s", self._filename)
self._dataframe = pd.read_csv(self._filename, sep=";", index_col=0)
self._signals.finished.emit(True)
@property
def signals(self):
return self._signals
@property
def dataframe(self):
return self._dataframe