every_holiday/VRCT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Update] change transcription_recorder.py/transcription_transcriber.py/notification.py func name

Browse Source
This commit is contained in:
misygauziya
2023-08-20 02:28:17 +09:00
parent 2be54f9aee
commit b7d6477228
4 changed files with 44 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
model.py
View File

@@ -10,7 +10,7 @@ from models.osc.osc_tools import sendTyping, sendMessage, sendTestAction, receiv
from models.transcription.transcription_recorder import SelectedMicRecorder, SelectedSpeakerRecorder
from models.transcription.transcription_recorder import SelectedMicEnergyRecorder, SelectedSpeakeEnergyRecorder
from models.transcription.transcription_transcriber import AudioTranscriber
from models.xsoverlay.notification import notification_xsoverlay_for_vrct
from models.xsoverlay.notification import xsoverlayForVRCT
from config import config
class threadFnc(Thread):
@@ -162,19 +162,19 @@ class Model:
config.INPUT_MIC_DYNAMIC_ENERGY_THRESHOLD,
config.INPUT_MIC_RECORD_TIMEOUT,
)
self.mic_audio_recorder.record_into_queue(mic_audio_queue)
self.mic_audio_recorder.recordIntoQueue(mic_audio_queue)
mic_transcriber = AudioTranscriber(
speaker=False,
source=self.mic_audio_recorder.source,
phrase_timeout=config.INPUT_MIC_PHRASE_TIMEOUT,
max_phrases=config.INPUT_MIC_MAX_PHRASES,
)
def mic_transcript_to_chatbox():
mic_transcriber.transcribe_audio_queue(mic_audio_queue, config.INPUT_MIC_VOICE_LANGUAGE)
message = mic_transcriber.get_transcript()
def sendMicTranscript():
mic_transcriber.transcribeAudioQueue(mic_audio_queue, config.INPUT_MIC_VOICE_LANGUAGE)
message = mic_transcriber.getTranscript()
fnc(message)
self.mic_print_transcript = threadFnc(mic_transcript_to_chatbox)
self.mic_print_transcript = threadFnc(sendMicTranscript)
self.mic_print_transcript.daemon = True
self.mic_print_transcript.start()
@@ -186,7 +186,7 @@ class Model:
self.mic_audio_recorder.stop = None
def startCheckMicEnergy(self, fnc):
def progressBarInputMicEnergyPlot():
def sendMicEnergy():
if mic_energy_queue.empty() is False:
energy = mic_energy_queue.get()
fnc(energy)
@@ -194,8 +194,8 @@ class Model:
mic_energy_queue = Queue()
mic_device = [device for device in getInputDevices()[config.CHOICE_MIC_HOST] if device["name"] == config.CHOICE_MIC_DEVICE][0]
self.mic_energy_recorder = SelectedMicEnergyRecorder(mic_device)
self.mic_energy_recorder.record_into_queue(mic_energy_queue)
self.mic_energy_plot_progressbar = threadFnc(progressBarInputMicEnergyPlot)
self.mic_energy_recorder.recordIntoQueue(mic_energy_queue)
self.mic_energy_plot_progressbar = threadFnc(sendMicEnergy)
self.mic_energy_plot_progressbar.daemon = True
self.mic_energy_plot_progressbar.start()
@@ -214,19 +214,19 @@ class Model:
config.INPUT_SPEAKER_DYNAMIC_ENERGY_THRESHOLD,
config.INPUT_SPEAKER_RECORD_TIMEOUT,
)
self.spk_audio_recorder.record_into_queue(spk_audio_queue)
self.spk_audio_recorder.recordIntoQueue(spk_audio_queue)
spk_transcriber = AudioTranscriber(
speaker=True,
source=self.spk_audio_recorder.source,
phrase_timeout=config.INPUT_SPEAKER_PHRASE_TIMEOUT,
max_phrases=config.INPUT_SPEAKER_MAX_PHRASES,
)
def spk_transcript_to_textbox():
spk_transcriber.transcribe_audio_queue(spk_audio_queue, config.INPUT_SPEAKER_VOICE_LANGUAGE)
message = spk_transcriber.get_transcript()
def sendSpkTranscript():
spk_transcriber.transcribeAudioQueue(spk_audio_queue, config.INPUT_SPEAKER_VOICE_LANGUAGE)
message = spk_transcriber.getTranscript()
fnc(message)
self.spk_print_transcript = threadFnc(spk_transcript_to_textbox)
self.spk_print_transcript = threadFnc(sendSpkTranscript)
self.spk_print_transcript.daemon = True
self.spk_print_transcript.start()
@@ -238,13 +238,13 @@ class Model:
self.spk_audio_recorder.stop = None
def startCheckSpeakerEnergy(self, fnc):
def progressBar_input_speaker_energy_plot():
def sendSpeakerEnergy():
if speaker_energy_queue.empty() is False:
energy = speaker_energy_queue.get()
fnc(energy)
sleep(0.01)
def progressBar_input_speaker_energy_get():
def getSpeakerEnergy():
with self.speaker_energy_recorder.source as source:
energy = self.speaker_energy_recorder.recorder.listen_energy(source)
speaker_energy_queue.put(energy)
@@ -252,10 +252,10 @@ class Model:
speaker_device = [device for device in getOutputDevices() if device["name"] == config.CHOICE_SPEAKER_DEVICE][0]
speaker_energy_queue = Queue()
self.speaker_energy_recorder = SelectedSpeakeEnergyRecorder(speaker_device)
self.speaker_energy_get_progressbar = threadFnc(progressBar_input_speaker_energy_get)
self.speaker_energy_get_progressbar = threadFnc(getSpeakerEnergy)
self.speaker_energy_get_progressbar.daemon = True
self.speaker_energy_get_progressbar.start()
self.speaker_energy_plot_progressbar = threadFnc(progressBar_input_speaker_energy_plot)
self.speaker_energy_plot_progressbar = threadFnc(sendSpeakerEnergy)
self.speaker_energy_plot_progressbar.daemon = True
self.speaker_energy_plot_progressbar.start()
@@ -265,7 +265,7 @@ class Model:
if self.speaker_energy_plot_progressbar != None:
self.speaker_energy_plot_progressbar.stop()
def notificationXsoverlay(self, message):
notification_xsoverlay_for_vrct(content=f"{message}")
def notificationXSOverlay(self, message):
xsoverlayForVRCT(content=f"{message}")
model = Model()

20
models/transcription/transcription_recorder.py
View File

@@ -15,11 +15,11 @@ class BaseRecorder:
self.source = source
def adjust_for_noise(self):
def adjustForNoise(self):
with self.source:
self.recorder.adjust_for_ambient_noise(self.source)
def record_into_queue(self, audio_queue):
def recordIntoQueue(self, audio_queue):
def record_callback(_, audio):
audio_queue.put((audio.get_raw_data(), datetime.now()))
@@ -32,7 +32,7 @@ class SelectedMicRecorder(BaseRecorder):
sample_rate=int(device["defaultSampleRate"]),
)
super().__init__(source=source, energy_threshold=energy_threshold, dynamic_energy_threshold=dynamic_energy_threshold, record_timeout=record_timeout)
# self.adjust_for_noise()
# self.adjustForNoise()
class SelectedSpeakerRecorder(BaseRecorder):
def __init__(self, device, energy_threshold, dynamic_energy_threshold, record_timeout):
@@ -44,7 +44,7 @@ class SelectedSpeakerRecorder(BaseRecorder):
channels=device["maxInputChannels"]
)
super().__init__(source=source, energy_threshold=energy_threshold, dynamic_energy_threshold=dynamic_energy_threshold, record_timeout=record_timeout)
# self.adjust_for_noise()
# self.adjustForNoise()
class BaseEnergyRecorder:
def __init__(self, source):
@@ -59,15 +59,15 @@ class BaseEnergyRecorder:
self.source = source
def adjust_for_noise(self):
def adjustForNoise(self):
with self.source:
self.recorder.adjust_for_ambient_noise(self.source)
def record_into_queue(self, energy_queue):
def record_callback(_, energy):
def recordIntoQueue(self, energy_queue):
def recordCallback(_, energy):
energy_queue.put(energy)
self.stop = self.recorder.listen_energy_in_background(self.source, record_callback)
self.stop = self.recorder.listen_energy_in_background(self.source, recordCallback)
class SelectedMicEnergyRecorder(BaseEnergyRecorder):
def __init__(self, device):
@@ -76,7 +76,7 @@ class SelectedMicEnergyRecorder(BaseEnergyRecorder):
sample_rate=int(device["defaultSampleRate"]),
)
super().__init__(source=source)
# self.adjust_for_noise()
# self.adjustForNoise()
class SelectedSpeakeEnergyRecorder(BaseEnergyRecorder):
def __init__(self, device):
@@ -88,4 +88,4 @@ class SelectedSpeakeEnergyRecorder(BaseEnergyRecorder):
channels=device["maxInputChannels"]
)
super().__init__(source=source)
# self.adjust_for_noise()
# self.adjustForNoise()

20
models/transcription/transcription_transcriber.py
View File

@@ -24,13 +24,13 @@ class AudioTranscriber:
"last_sample": bytes(),
"last_spoken": None,
"new_phrase": True,
"process_data_func": self.process_speaker_data if speaker else self.process_speaker_data
"process_data_func": self.processSpeakerData if speaker else self.processSpeakerData
}
def transcribe_audio_queue(self, audio_queue, language):
def transcribeAudioQueue(self, audio_queue, language):
# while True:
audio, time_spoken = audio_queue.get()
self.update_last_sample_and_phrase_status(audio, time_spoken)
self.updateLastSampleAndPhraseStatus(audio, time_spoken)
text = ''
try:
@@ -45,9 +45,9 @@ class AudioTranscriber:
# os.unlink(path)
if text != '':
self.update_transcript(text)
self.updateTranscript(text)
def update_last_sample_and_phrase_status(self, data, time_spoken):
def updateLastSampleAndPhraseStatus(self, data, time_spoken):
source_info = self.audio_sources
if source_info["last_spoken"] and time_spoken - source_info["last_spoken"] > timedelta(seconds=self.phrase_timeout):
source_info["last_sample"] = bytes()
@@ -58,11 +58,11 @@ class AudioTranscriber:
source_info["last_sample"] += data
source_info["last_spoken"] = time_spoken
def process_mic_data(self):
def processMicData(self):
audio_data = AudioData(self.audio_sources["last_sample"], self.audio_sources["sample_rate"], self.audio_sources["sample_width"])
return audio_data
def process_speaker_data(self):
def processSpeakerData(self):
temp_file = BytesIO()
with wave.open(temp_file, 'wb') as wf:
wf.setnchannels(self.audio_sources["channels"])
@@ -74,7 +74,7 @@ class AudioTranscriber:
audio = self.audio_recognizer.record(source)
return audio
def update_transcript(self, text):
def updateTranscript(self, text):
source_info = self.audio_sources
transcript = self.transcript_data
@@ -85,14 +85,14 @@ class AudioTranscriber:
else:
transcript[0] = text
def get_transcript(self):
def getTranscript(self):
if len(self.transcript_data) > 0:
text = self.transcript_data.pop(-1)
else:
text = ""
return text
def clear_transcript_data(self):
def clearTranscriptData(self):
self.transcript_data.clear()
self.audio_sources["last_sample"] = bytes()
self.audio_sources["new_phrase"] = True

8
models/xsoverlay/notification.py
View File

@@ -19,7 +19,7 @@ import socket
import json
import base64
def notification_xsoverlay(
def XSOverlay(
endpoint:tuple=("127.0.0.1", 42069), messageType:int=1, index:int=0, timeout:float=2,
height:float=120.0, opacity:float=1.0, volume:float=0.0, audioPath:str="",
title:str="", content:str="", useBase64Icon:bool=False, icon:str="default", sourceApp:str=""
@@ -58,8 +58,8 @@ def notification_xsoverlay(
sock.close()
return response
def notification_xsoverlay_for_vrct(content:str="") -> int:
response = notification_xsoverlay(
def xsoverlayForVRCT(content:str="") -> int:
response = XSOverlay(
title="VRCT",
content=content,
useBase64Icon=True,
@@ -69,4 +69,4 @@ def notification_xsoverlay_for_vrct(content:str="") -> int:
return response
if __name__ == "__main__":
notification_xsoverlay_for_vrct(content="notification test")
xsoverlayForVRCT(content="notification test")