320 lines
11 KiB
Python
320 lines
11 KiB
Python
#!/usr/bin/python3.8
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Read data from stream and record sound and video at the same time
|
|
"""
|
|
import dataclasses, logging, os, time, h5py
|
|
from .lasp_avstream import StreamManager, StreamMetaData, StreamMsg
|
|
from .lasp_common import AvType
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class RecordStatus:
|
|
curT: float
|
|
done: bool
|
|
|
|
|
|
class Recording:
|
|
"""
|
|
Class used to perform a recording.
|
|
"""
|
|
def __init__(self, fn: str, streammgr: StreamManager,
|
|
rectime: float = None, wait: bool = True,
|
|
progressCallback=None,
|
|
startDelay: float=0):
|
|
"""
|
|
Start a recording. Blocks if wait is set to True.
|
|
|
|
Args:
|
|
fn: Filename to record to. Extension is automatically added if not
|
|
provided.
|
|
stream: AvStream instance to record from. Should have input
|
|
channels!
|
|
rectime: Recording time [s], None for infinite, in seconds. If set
|
|
to None, or np.inf, the recording continues indefintely.
|
|
progressCallback: callable that is called with an instance of
|
|
RecordStatus instance as argument.
|
|
startDelay: Optional delay added before the recording is *actually*
|
|
started in [s].
|
|
"""
|
|
ext = '.h5'
|
|
if ext not in fn:
|
|
fn += ext
|
|
|
|
self.smgr = streammgr
|
|
self.metadata = None
|
|
|
|
assert startDelay >= 0
|
|
self.startDelay = startDelay
|
|
# Flag used to indicate that we have passed the start delay
|
|
self.startDelay_passed = False
|
|
self.rectime = rectime
|
|
self.fn = fn
|
|
|
|
self.video_frame_positions = []
|
|
self.curT_rounded_to_seconds = 0
|
|
|
|
# Counter of the number of blocks
|
|
self.ablockno = 0
|
|
self.vframeno = 0
|
|
|
|
self.progressCallback = progressCallback
|
|
self.wait = wait
|
|
|
|
self.f = h5py.File(self.fn, 'w')
|
|
|
|
# This flag is used to delete the file on finish(), and can be used
|
|
# when a recording is canceled.
|
|
self.deleteFile = False
|
|
|
|
try:
|
|
# Input queue
|
|
self.inq = streammgr.addInQueueListener()
|
|
|
|
except RuntimeError:
|
|
# Cleanup stuff, something is going wrong when starting the stream
|
|
try:
|
|
self.f.close()
|
|
except Exception as e:
|
|
logging.error(
|
|
'Error preliminary closing measurement file {fn}: {str(e)}')
|
|
|
|
self.__deleteFile()
|
|
raise
|
|
|
|
# Try to obtain stream metadata
|
|
streammgr.getStreamStatus(AvType.audio_input)
|
|
streammgr.getStreamStatus(AvType.audio_duplex)
|
|
|
|
self.ad = None
|
|
|
|
logging.debug('Starting record....')
|
|
# TODO: Fix this later when we want video
|
|
# if stream.hasVideo():
|
|
# stream.addCallback(self.aCallback, AvType.audio_input)
|
|
self.stop = False
|
|
|
|
if self.wait:
|
|
logging.debug('Stop recording with CTRL-C')
|
|
try:
|
|
while not self.stop:
|
|
self.handleQueue()
|
|
time.sleep(0.01)
|
|
except KeyboardInterrupt:
|
|
logging.debug("Keyboard interrupt on record")
|
|
finally:
|
|
self.finish()
|
|
|
|
def handleQueue(self):
|
|
"""
|
|
This method should be called to grab data from the input queue, which
|
|
is filled by the stream, and put it into a file. It should be called at
|
|
a regular interval to prevent overflowing of the queue. It is called
|
|
within the start() method of the recording, if block is set to True.
|
|
Otherwise, it should be called from its parent at regular intervals.
|
|
For example, in Qt this can be done using a QTimer.
|
|
|
|
|
|
"""
|
|
# logging.debug('handleQueue()')
|
|
while self.inq.qsize() > 0:
|
|
msg, data = self.inq.get()
|
|
# logging.debug(f'Obtained message: {msg}')
|
|
if msg == StreamMsg.streamData:
|
|
samples, = data
|
|
self.__addTimeData(samples)
|
|
elif msg == StreamMsg.streamStarted:
|
|
logging.debug(f'handleQueue obtained message {msg}')
|
|
avtype, metadata = data
|
|
if metadata is None:
|
|
raise RuntimeError('BUG: no stream metadata')
|
|
if avtype in (AvType.audio_duplex, AvType.audio_input):
|
|
self.processStreamMetaData(metadata)
|
|
elif msg == StreamMsg.streamMetaData:
|
|
logging.debug(f'handleQueue obtained message {msg}')
|
|
avtype, metadata = data
|
|
if metadata is not None:
|
|
self.processStreamMetaData(metadata)
|
|
elif msg == StreamMsg.streamTemporaryError:
|
|
pass
|
|
else:
|
|
logging.debug(f'handleQueue obtained message {msg}')
|
|
# An error occured, we do not remove the file, but we stop.
|
|
self.stop = True
|
|
logging.debug(f'Stream message: {msg}. Recording stopped unexpectedly')
|
|
raise RuntimeError('Recording stopped unexpectedly')
|
|
|
|
|
|
def processStreamMetaData(self, md: StreamMetaData):
|
|
"""
|
|
Stream metadata has been catched. This is used to set all metadata in
|
|
the measurement file
|
|
|
|
"""
|
|
logging.debug('Recording::processStreamMetaData()')
|
|
if self.metadata is not None:
|
|
# Metadata already obtained. We check whether the new metadata is
|
|
# compatible. Otherwise an error occurs
|
|
if md != self.metadata:
|
|
raise RuntimeError('BUG: Incompatible stream metadata!')
|
|
return
|
|
|
|
# The 'Audio' dataset as specified in lasp_measurement, where data is
|
|
# sent to. We use gzip as compression, this gives moderate a moderate
|
|
# compression to the data.
|
|
f = self.f
|
|
blocksize = md.blocksize
|
|
nchannels = len(md.in_ch)
|
|
self.ad = f.create_dataset('audio',
|
|
(1, blocksize, nchannels),
|
|
dtype=md.dtype,
|
|
maxshape=(
|
|
None, # This means, we can add blocks
|
|
# indefinitely
|
|
blocksize,
|
|
nchannels),
|
|
compression='gzip'
|
|
)
|
|
|
|
# TODO: This piece of code is not up-to-date and should be changed at a
|
|
# later instance once we really want to record video simultaneously
|
|
# with audio.
|
|
# if smgr.hasVideo():
|
|
# video_x, video_y = smgr.video_x, smgr.video_y
|
|
# self.vd = f.create_dataset('video',
|
|
# (1, video_y, video_x, 3),
|
|
# dtype='uint8',
|
|
# maxshape=(
|
|
# None, video_y, video_x, 3),
|
|
# compression='gzip'
|
|
# )
|
|
|
|
# Set the bunch of attributes
|
|
f.attrs['samplerate'] = md.fs
|
|
f.attrs['nchannels'] = nchannels
|
|
f.attrs['blocksize'] = blocksize
|
|
f.attrs['sensitivity'] = [ch.sensitivity for ch in md.in_ch]
|
|
f.attrs['channelNames'] = [ch.channel_name for ch in md.in_ch]
|
|
f.attrs['time'] = time.time()
|
|
self.blocksize = blocksize
|
|
self.fs = md.fs
|
|
|
|
# Measured physical quantity metadata
|
|
f.attrs['qtys'] = [ch.qty.to_json() for ch in md.in_ch]
|
|
self.metadata = md
|
|
|
|
def setDelete(self, val: bool):
|
|
"""
|
|
Set the delete flag. If set, measurement file is deleted at the end of
|
|
the recording. Typically used for cleaning up after canceling a
|
|
recording.
|
|
"""
|
|
self.deleteFile = val
|
|
|
|
def finish(self):
|
|
"""
|
|
This method should be called to finish and a close a recording file,
|
|
remove the queue from the stream, etc.
|
|
|
|
"""
|
|
logging.debug('Recording::finish()')
|
|
smgr = self.smgr
|
|
|
|
# TODO: Fix when video
|
|
# if smgr.hasVideo():
|
|
# smgr.removeCallback(self.vCallback, AvType.video_input)
|
|
# self.f['video_frame_positions'] = self.video_frame_positions
|
|
|
|
try:
|
|
smgr.removeInQueueListener(self.inq)
|
|
except Exception as e:
|
|
logging.error(f'Could not remove queue from smgr: {e}')
|
|
|
|
try:
|
|
# Close the recording file
|
|
self.f.close()
|
|
except Exception as e:
|
|
logging.error(f'Error closing file: {e}')
|
|
|
|
logging.debug('Recording ended')
|
|
if self.deleteFile:
|
|
self.__deleteFile()
|
|
|
|
def __deleteFile(self):
|
|
"""
|
|
Cleanup the recording file.
|
|
"""
|
|
try:
|
|
os.remove(self.fn)
|
|
except Exception as e:
|
|
logging.error(f'Error deleting file: {self.fn}')
|
|
|
|
def __addTimeData(self, indata):
|
|
"""
|
|
Called by handleQueue() and adds new time data to the storage file.
|
|
"""
|
|
# logging.debug('Recording::__addTimeData()')
|
|
|
|
if self.stop:
|
|
# Stop flag is raised. We stop recording here.
|
|
return
|
|
|
|
# The current time that is recorded and stored into the file, without
|
|
# the new data
|
|
if not self.metadata:
|
|
# We obtained stream data, but metadata is not yet available.
|
|
# Therefore, we request it explicitly and then we return
|
|
logging.info('Requesting stream metadata')
|
|
self.smgr.getStreamStatus(AvType.audio_input)
|
|
self.smgr.getStreamStatus(AvType.audio_duplex)
|
|
return
|
|
|
|
curT = self.ablockno*self.blocksize/self.fs
|
|
|
|
# Increase the block counter
|
|
self.ablockno += 1
|
|
|
|
if curT < self.startDelay and not self.startDelay_passed:
|
|
# Start delay has not been passed
|
|
return
|
|
elif curT >= 0 and not self.startDelay_passed:
|
|
# Start delay passed, switch the flag!
|
|
self.startDelay_passed = True
|
|
# Reset the audio block counter and the time
|
|
self.ablockno = 1
|
|
curT = 0
|
|
|
|
recstatus = RecordStatus(
|
|
curT=curT,
|
|
done=False)
|
|
|
|
if self.progressCallback is not None:
|
|
self.progressCallback(recstatus)
|
|
|
|
curT_rounded_to_seconds = int(curT)
|
|
if curT_rounded_to_seconds > self.curT_rounded_to_seconds:
|
|
self.curT_rounded_to_seconds = curT_rounded_to_seconds
|
|
print(f'{curT_rounded_to_seconds}', end='', flush=True)
|
|
else:
|
|
print('.', end='', flush=True)
|
|
|
|
if self.rectime is not None and curT > self.rectime:
|
|
# We are done!
|
|
if self.progressCallback is not None:
|
|
recstatus.done = True
|
|
self.progressCallback(recstatus)
|
|
self.stop = True
|
|
return
|
|
|
|
# Add the data to the file, and resize the audio data blocks
|
|
self.ad.resize(self.ablockno, axis=0)
|
|
self.ad[self.ablockno-1, :, :] = indata
|
|
|
|
|
|
# def _vCallback(self, frame, framectr):
|
|
# self.video_frame_positions.append(self.ablockno())
|
|
# vframeno = self.vframeno
|
|
# self.vd.resize(vframeno+1, axis=0)
|
|
# self.vd[vframeno, :, :] = frame
|
|
# self.vframeno += 1
|