import time
import logging
import numpy as np
from . input import Input
log = logging.getLogger(__name__)
[docs]class CameraRawVideo(Input):
'''
Treats raw video as a camera.
Args:
id (str): path to the raw video file.
config (dict): Configuration dictionary. Accepted keywords:
res (tuple): frame size in the format (width, height)
'''
def __init__(self, id=0, config=None):
config = config or {}
defaults = {
'res': (1920, 1080),
}
Input.__init__(self, id=id, config=config, defaults=defaults)
[docs] def open(self):
'''Opens raw video as a bytes file.'''
self.input = open(self.id, 'rb')
[docs] def read(self):
'''
Reads in raw video.
config['res'] dictates how many bytes are read in.
'''
frame = None
timestamp = None
try:
# Length of frame in bytes
frame_length = np.uint8().itemsize * np.product(self.config.get('res'))
frame_bytes = self.input.read(frame_length)
timestamp = time.time()
buf = np.frombuffer(frame_bytes, dtype=np.uint8)
if buf.size != 0:
frame = buf.reshape(self.config.get('res'))
except Exception as e:
log.error(f'{str(self)} read error: {e}')
return frame, timestamp
[docs] def close(self):
if self.input:
self.input.close()
self.input = None