118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
import time
|
|
import copy
|
|
import warnings
|
|
|
|
from typing import List
|
|
|
|
from UniTAP.common.audio_mode import AudioFrameData
|
|
from UniTAP.dev.modules.capturer.capture import Capturer, CaptureConfig
|
|
from UniTAP.dev.modules.capturer.statuses import AudioCaptureStatus
|
|
from .result_audio import ResultAudioObject
|
|
|
|
|
|
class AudioCapturer:
|
|
"""
|
|
Class `AudioCapturer` allows working with capturing audio frames on Sink (RX - receiver) side.
|
|
You can `start` capturing in several modes, `stop` capturing, getting current `status` and result of capturing
|
|
`capture_result`.
|
|
"""
|
|
def __init__(self, capturer: Capturer):
|
|
self.__capturer = capturer
|
|
self.__result = ResultAudioObject()
|
|
|
|
@property
|
|
def status(self) -> AudioCaptureStatus:
|
|
"""
|
|
Returns current audio capturer status.
|
|
|
|
Returns:
|
|
object of `AudioCaptureStatus` type
|
|
"""
|
|
return self.__capturer.audio_capturer_status
|
|
|
|
@property
|
|
def capture_result(self) -> ResultAudioObject:
|
|
"""
|
|
Returns result of audio capturing.
|
|
|
|
Returns:
|
|
object of `ResultAudioObject` type
|
|
"""
|
|
return self.__result
|
|
|
|
def start(self, frames_count=0, m_sec=0, timeout=None):
|
|
"""
|
|
Start capturing. Possible some variants of capturing:
|
|
- Capture with fixed frames count (will be captured fixed frames count and capturing will be stopped).
|
|
- Capture with fixed audio duration (captures audio for the specified duration in milliseconds).
|
|
- Capture without parameters - Live capturing (for getting frames you need to use functions `pop_element`)
|
|
- Capture with timeout (maximum duration for capturing operations before stopping, in seconds).
|
|
|
|
All results can be obtained using the function `capture_result`.
|
|
|
|
Args:
|
|
frames_count (int)
|
|
m_sec (int)
|
|
"""
|
|
config = CaptureConfig()
|
|
config.audio = True
|
|
config.type = CaptureConfig.Type.LIVE
|
|
config.video = True
|
|
|
|
self.__capturer.start_capture(config)
|
|
|
|
self.__result.clear()
|
|
self.__result.start_capture_time = time.time()
|
|
|
|
if frames_count > 0:
|
|
self.__result.buffer.extend(self.__capturer.capture_audio_by_n_frames(frames_count, timeout))
|
|
self.__fill_audio_mode()
|
|
elif m_sec > 0:
|
|
self.__result.buffer.extend(self.__capturer.capture_audio_by_m_sec(m_sec))
|
|
self.__fill_audio_mode()
|
|
|
|
def __fill_audio_mode(self):
|
|
if len(self.__result.buffer) > 0:
|
|
self.__result.audio_mode.channel_count = self.__result.buffer[0].channel_count
|
|
self.__result.audio_mode.bits = self.__result.buffer[0].sample_size
|
|
self.__result.audio_mode.sample_rate = self.__result.buffer[0].sample_rate
|
|
|
|
def stop(self):
|
|
"""
|
|
Stop capture audio.
|
|
"""
|
|
config = CaptureConfig()
|
|
config.audio = True
|
|
config.type = CaptureConfig.Type.LIVE
|
|
config.video = True
|
|
|
|
self.__capturer.stop_capture(config)
|
|
self.__result.end_capture_time = time.time()
|
|
|
|
def pop_element(self) -> List[AudioFrameData]:
|
|
"""
|
|
Return first object of `AudioFrameData`.
|
|
|
|
Returns:
|
|
object of `AudioFrameData` type
|
|
"""
|
|
if self.status == AudioCaptureStatus.Stop:
|
|
warnings.warn("Audio capture is not working now. Please, turn it on.")
|
|
captured_audio_frames = self.__capturer.capture_audio_by_n_frames(1)
|
|
self.__result.buffer.extend(copy.deepcopy(captured_audio_frames))
|
|
self.__fill_audio_mode()
|
|
return captured_audio_frames[0]
|
|
|
|
def pop_element_as_result_object(self) -> ResultAudioObject:
|
|
"""
|
|
Return captured audio frame(objects of `AudioFrameData`) as `ResultAudioObject`.
|
|
|
|
Returns:
|
|
object of `ResultAudioObject` type
|
|
"""
|
|
captured_audio_frames = self.__capturer.capture_audio_by_n_frames(1)
|
|
|
|
res = ResultAudioObject()
|
|
res.buffer.extend(captured_audio_frames)
|
|
return res
|