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

🚧 [WIP/TEST] Deviceの自動変更に対応中

Browse Source
This commit is contained in:
misyaguziya
2024-10-02 12:34:11 +09:00
parent d2e03e0093
commit 92f21c32f6
3 changed files with 172 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src-python/model.py
View File

@@ -31,6 +31,8 @@ from models.overlay.overlay_image import OverlayImage
from config import config
import utils
class threadFnc(Thread):
def __init__(self, fnc, end_fnc=None, daemon=True, *args, **kwargs):
super(threadFnc, self).__init__(daemon=daemon, target=fnc, *args, **kwargs)
@@ -600,14 +602,8 @@ class Model:
self.mic_energy_recorder = None
def startSpeakerTranscript(self, fnc):
if config.AUTO_SPEAKER_SELECT is True:
default_device = device_manager.getDefaultSpeakerDevice()
speaker_device_name = default_device["device"]["name"]
else:
speaker_device_name = config.SELECTED_SPEAKER_DEVICE
speaker_device_list = device_manager.getSpeakerDevices()
selected_speaker_device = [device for device in speaker_device_list if device["name"] == speaker_device_name]
selected_speaker_device = [device for device in speaker_device_list if device["name"] == config.SELECTED_SPEAKER_DEVICE]
if len(selected_speaker_device) == 0:
return False
@@ -678,12 +674,16 @@ class Model:
def stopSpeakerTranscript(self):
if isinstance(self.speaker_print_transcript, threadFnc):
utils.printLog("stop speaker_print_transcript")
self.speaker_print_transcript.stop()
self.speaker_print_transcript.join()
self.speaker_print_transcript = None
utils.printLog("stopped speaker_audio_recorder")
if isinstance(self.speaker_audio_recorder, SelectedSpeakerEnergyAndAudioRecorder):
utils.printLog("stop speaker_audio_recorder")
self.speaker_audio_recorder.stop()
self.speaker_audio_recorder = None
utils.printLog("stopped speaker_audio_recorder")
# if isinstance(self.speaker_get_energy, threadFnc):
# self.speaker_get_energy.stop()
# self.speaker_get_energy = None
@@ -692,14 +692,8 @@ class Model:
if isinstance(fnc, Callable):
self.check_speaker_energy_fnc = fnc
if config.AUTO_SPEAKER_SELECT is True:
default_device = device_manager.getDefaultSpeakerDevice()
speaker_device_name = default_device["device"]["name"]
else:
speaker_device_name = config.SELECTED_SPEAKER_DEVICE
speaker_device_list = device_manager.getSpeakerDevices()
selected_speaker_device = [device for device in speaker_device_list if device["name"] == speaker_device_name]
selected_speaker_device = [device for device in speaker_device_list if device["name"] == config.SELECTED_SPEAKER_DEVICE]
if len(selected_speaker_device) == 0:
return False
@@ -723,13 +717,17 @@ class Model:
def stopCheckSpeakerEnergy(self):
if isinstance(self.speaker_energy_plot_progressbar, threadFnc):
utils.printLog("stop speaker_energy_plot_progressbar")
self.speaker_energy_plot_progressbar.stop()
self.speaker_energy_plot_progressbar.join()
self.speaker_energy_plot_progressbar = None
utils.printLog("stopped speaker_energy_plot_progressbar")
if isinstance(self.speaker_energy_recorder, SelectedSpeakerEnergyRecorder):
utils.printLog("stopped speaker_energy_recorder")
self.speaker_energy_recorder.resume()
self.speaker_energy_recorder.stop()
self.speaker_energy_recorder = None
utils.printLog("stopped speaker_energy_recorder")
def createOverlayImageShort(self, message, translation):
your_language = config.SELECTED_TARGET_LANGUAGES[config.SELECTED_TAB_NO]["primary"]["language"]

156
src-python/models/transcription/transcription_utils.py
View File

@@ -4,6 +4,7 @@ import comtypes
from pyaudiowpatch import PyAudio, paWASAPI
from pycaw.callbacks import MMNotificationClient
from pycaw.utils import AudioUtilities
from utils import printLog
class Client(MMNotificationClient):
def __init__(self):
@@ -22,8 +23,8 @@ class Client(MMNotificationClient):
def on_device_state_changed(self, device_id, state):
self.loop = False
def on_property_value_changed(self, device_id, key):
self.loop = False
# def on_property_value_changed(self, device_id, key):
# self.loop = False
class DeviceManager:
_instance = None
@@ -39,6 +40,7 @@ class DeviceManager:
self.default_mic_device = {"host": {"name": "NoHost"}, "device": {"name": "NoDevice"}}
self.speaker_devices = [{"name": "NoDevice"}]
self.default_speaker_device = {"device": {"name": "NoDevice"}}
self.update()
self.prev_mic_host = [host for host in self.mic_devices]
@@ -47,12 +49,19 @@ class DeviceManager:
self.prev_speaker_devices = self.speaker_devices
self.prev_default_speaker_device = self.default_speaker_device
self.update_flag_default_mic_device = False
self.update_flag_default_speaker_device = False
self.update_flag_host_list = False
self.update_flag_mic_device_list = False
self.update_flag_speaker_device_list = False
self.callback_default_mic_device = None
self.callback_default_speaker_device = None
self.callback_host_list = None
self.callback_mic_device_list = None
self.callback_speaker_device_list = None
self.callback_prev_update = None
self.callback_process_before_update_devices = None
self.callback_process_after_update_devices = None
self.monitoring_flag = False
self.startMonitoring()
@@ -130,30 +139,71 @@ class DeviceManager:
self.speaker_devices = buffer_speaker_devices
self.default_speaker_device = buffer_default_speaker_device
def checkUpdate(self):
printLog("checkUpdate")
if self.prev_default_mic_device["device"]["name"] != self.default_mic_device["device"]["name"]:
printLog("checkUpdate: default_mic_device")
self.update_flag_default_mic_device = True
self.prev_default_mic_device = self.default_mic_device
if self.prev_default_speaker_device["device"]["name"] != self.default_speaker_device["device"]["name"]:
printLog("checkUpdate: default_speaker_device")
self.update_flag_default_speaker_device = True
self.prev_default_speaker_device = self.default_speaker_device
if self.prev_mic_host != [host for host in self.mic_devices]:
printLog("checkUpdate: mic_host")
self.update_flag_host_list = True
self.prev_mic_host = [host for host in self.mic_devices]
if self.prev_mic_devices != self.mic_devices:
printLog("checkUpdate: mic_devices")
self.update_flag_mic_device_list = True
self.prev_mic_devices = self.mic_devices
if self.prev_speaker_devices != self.speaker_devices:
printLog("checkUpdate: speaker_devices")
self.update_flag_speaker_device_list = True
self.prev_speaker_devices = self.speaker_devices
update_flag = (
self.update_flag_default_mic_device or
self.update_flag_default_speaker_device or
self.update_flag_host_list or
self.update_flag_mic_device_list or
self.update_flag_speaker_device_list
)
return update_flag
def monitoring(self):
comtypes.CoInitialize()
cb = Client()
enumerator = AudioUtilities.GetDeviceEnumerator()
enumerator.RegisterEndpointNotificationCallback(cb)
try:
while self.monitoring_flag is True:
try:
while cb.loop is True:
comtypes.CoInitialize()
self.cb = Client()
self.enumerator = AudioUtilities.GetDeviceEnumerator()
self.enumerator.RegisterEndpointNotificationCallback(self.cb)
while self.cb.loop is True:
printLog("Monitoring Loop")
sleep(1)
enumerator.UnregisterEndpointNotificationCallback(cb)
self.runPrevUpdateDevices()
self.enumerator.UnregisterEndpointNotificationCallback(self.cb)
except Exception as e:
printLog("Device Monitoring: ", e)
finally:
printLog("Device Monitoring Finally Init")
comtypes.CoUninitialize()
printLog("Run Process Before Update Devices")
self.runProcessBeforeUpdateDevices()
while True:
sleep(2)
self.update()
self.noticeDefaultDevice()
except Exception:
pass
finally:
cb = Client()
enumerator = AudioUtilities.GetDeviceEnumerator()
enumerator.RegisterEndpointNotificationCallback(cb)
except Exception:
pass
comtypes.CoUninitialize()
if self.checkUpdate() is True:
break
printLog("Notice Update Devices")
self.noticeUpdateDevices()
printLog("Run Process After Update Devices")
self.runProcessAfterUpdateDevices()
except Exception as e:
printLog("Device Monitoring Exception: ", e)
printLog("Device Monitoring End")
def startMonitoring(self):
self.monitoring_flag = True
@@ -195,46 +245,56 @@ class DeviceManager:
def clearCallbackSpeakerDeviceList(self):
self.callback_speaker_device_list = None
def setCallbackPrevUpdateDevices(self, callback):
self.callback_prev_update = callback
def setCallbackProcessBeforeUpdateDevices(self, callback):
self.callback_process_before_update_devices = callback
def clearCallbackPrevUpdateDevices(self):
self.callback_prev_update = None
def clearCallbackProcessBeforeUpdateDevices(self):
self.callback_process_before_update_devices = None
def runPrevUpdateDevices(self):
self.callback_prev_update()
def runProcessBeforeUpdateDevices(self):
self.callback_process_before_update_devices()
def noticeDefaultDevice(self):
if self.callback_default_mic_device is not None:
if self.prev_default_mic_device["device"]["name"] != self.default_mic_device["device"]["name"]:
def setCallbackProcessAfterUpdateDevices(self, callback):
self.callback_process_after_update_devices = callback
def clearCallbackProcessAfterUpdateDevices(self):
self.callback_process_after_update_devices = None
def runProcessAfterUpdateDevices(self):
self.callback_process_after_update_devices()
def noticeUpdateDevices(self):
if self.callback_default_mic_device is not None and self.update_flag_default_mic_device is True:
self.setMicDefaultDevice()
if self.callback_default_speaker_device is not None and self.update_flag_default_speaker_device is True:
self.setSpeakerDefaultDevice()
if self.callback_host_list is not None and self.update_flag_host_list is True:
self.setMicHost()
if self.callback_mic_device_list is not None and self.update_flag_mic_device_list is True:
self.setMicDeviceList()
if self.callback_speaker_device_list is not None and self.update_flag_speaker_device_list is True:
self.setSpeakerDeviceList()
self.update_flag_default_mic_device = False
self.update_flag_default_speaker_device = False
self.update_flag_host_list = False
self.update_flag_mic_device_list = False
self.update_flag_speaker_device_list = False
def setMicDefaultDevice(self):
self.callback_default_mic_device(self.default_mic_device["host"]["name"], self.default_mic_device["device"]["name"])
self.prev_default_mic_device = self.default_mic_device
if self.callback_default_speaker_device is not None:
if self.prev_default_speaker_device["device"]["name"] != self.default_speaker_device["device"]["name"]:
def setSpeakerDefaultDevice(self):
self.callback_default_speaker_device(self.default_speaker_device["device"]["name"])
self.prev_default_speaker_device = self.default_speaker_device
if self.callback_host_list is not None:
if self.prev_mic_host != [host for host in self.mic_devices]:
def setMicHost(self):
self.callback_host_list()
self.prev_mic_host = [host for host in self.mic_devices]
if self.callback_mic_device_list is not None:
if {key: [device['name'] for device in devices] for key, devices in self.prev_mic_devices.items()} != {key: [device['name'] for device in devices] for key, devices in self.mic_devices.items()}:
def setMicDeviceList(self):
self.callback_mic_device_list()
self.prev_mic_devices = self.mic_devices
if self.callback_speaker_device_list is not None:
if [device['name'] for device in self.prev_speaker_devices] != [device['name'] for device in self.speaker_devices]:
def setSpeakerDeviceList(self):
self.callback_speaker_device_list()
self.prev_speaker_devices = self.speaker_devices
def forceSetMicDefaultDevice(self):
self.callback_default_mic_device(self.default_mic_device["host"]["name"], self.default_mic_device["device"]["name"])
def forceSetSpeakerDefaultDevice(self):
self.callback_default_speaker_device(self.default_speaker_device["device"]["name"])
def getMicDevices(self):
return self.mic_devices

73
src-python/webui_controller.py
View File

@@ -56,7 +56,25 @@ class Controller:
model.getListSpeakerDevice(),
)
def prevUpdateSelectedDevices(self) -> None:
def restartAccessDevices(self) -> None:
if config.ENABLE_TRANSCRIPTION_SEND is True:
printLog("Restart Access Devices", "Start Mic Transcript")
self.startThreadingTranscriptionSendMessage()
if config.ENABLE_TRANSCRIPTION_RECEIVE is True:
printLog("Restart Access Devices", "Start Speaker Transcript")
self.startThreadingTranscriptionReceiveMessage()
if config.ENABLE_CHECK_ENERGY_SEND is True:
printLog("Restart Access Devices", "Start Check Mic Energy")
model.startCheckMicEnergy(
self.progressBarMicEnergy,
)
if config.ENABLE_CHECK_ENERGY_RECEIVE is True:
printLog("Restart Access Devices", "Start Check Speaker Energy")
model.startCheckSpeakerEnergy(
self.progressBarSpeakerEnergy,
)
def stopAccessDevices(self) -> None:
if config.ENABLE_TRANSCRIPTION_SEND is True:
model.stopMicTranscript()
if config.ENABLE_TRANSCRIPTION_RECEIVE is True:
@@ -70,9 +88,11 @@ class Controller:
config.SELECTED_MIC_HOST = host
config.SELECTED_MIC_DEVICE = device
if config.ENABLE_TRANSCRIPTION_SEND is True:
model.startMicTranscript()
self.startThreadingTranscriptionSendMessage()
if config.ENABLE_CHECK_ENERGY_SEND is True:
model.startCheckMicEnergy()
model.startCheckMicEnergy(
self.progressBarMicEnergy,
)
self.run(
200,
self.run_mapping["selected_mic_device"],
@@ -82,9 +102,11 @@ class Controller:
def updateSelectedSpeakerDevice(self, device) -> None:
config.SELECTED_SPEAKER_DEVICE = device
if config.ENABLE_TRANSCRIPTION_RECEIVE is True:
model.startSpeakerTranscript()
self.startThreadingTranscriptionReceiveMessage()
if config.ENABLE_CHECK_ENERGY_RECEIVE is True:
model.startCheckSpeakerEnergy()
model.startCheckSpeakerEnergy(
self.progressBarSpeakerEnergy,
)
self.run(
200,
self.run_mapping["selected_speaker_device"],
@@ -599,16 +621,18 @@ class Controller:
def setEnableAutoMicSelect(self, *args, **kwargs) -> dict:
config.AUTO_MIC_SELECT = True
device_manager.setCallbackPrevUpdateDevices(self.prevUpdateSelectedDevices)
device_manager.setCallbackProcessBeforeUpdateDevices(self.stopAccessDevices)
device_manager.setCallbackDefaultMicDevice(self.updateSelectedMicDevice)
device_manager.noticeDefaultDevice()
device_manager.forceSetMicDefaultDevice()
device_manager.setCallbackProcessAfterUpdateDevices(self.restartAccessDevices)
device_manager.noticeUpdateDevices()
device_manager.setMicDefaultDevice()
return {"status":200, "result":config.AUTO_MIC_SELECT}
@staticmethod
def setDisableAutoMicSelect(*args, **kwargs) -> dict:
device_manager.clearCallbackPrevUpdateDevices()
device_manager.clearCallbackProcessBeforeUpdateDevices()
device_manager.clearCallbackDefaultMicDevice()
device_manager.clearCallbackProcessAfterUpdateDevices()
config.AUTO_MIC_SELECT = False
return {"status":200, "result":config.AUTO_MIC_SELECT}
@@ -616,13 +640,14 @@ class Controller:
def getSelectedMicHost(*args, **kwargs) -> dict:
return {"status":200, "result":config.SELECTED_MIC_HOST}
@staticmethod
def setSelectedMicHost(data, *args, **kwargs) -> dict:
def setSelectedMicHost(self, data, *args, **kwargs) -> dict:
config.SELECTED_MIC_HOST = data
config.SELECTED_MIC_DEVICE = model.getMicDefaultDevice()
if config.ENABLE_CHECK_ENERGY_SEND is True:
model.stopCheckMicEnergy()
model.startCheckMicEnergy()
model.startCheckMicEnergy(
self.progressBarMicEnergy,
)
return {"status":200,
"result":{
"host":config.SELECTED_MIC_HOST,
@@ -634,12 +659,13 @@ class Controller:
def getSelectedMicDevice(*args, **kwargs) -> dict:
return {"status":200, "result":config.SELECTED_MIC_DEVICE}
@staticmethod
def setSelectedMicDevice(data, *args, **kwargs) -> dict:
def setSelectedMicDevice(self, data, *args, **kwargs) -> dict:
config.SELECTED_MIC_DEVICE = data
if config.ENABLE_CHECK_ENERGY_SEND is True:
model.stopCheckMicEnergy()
model.startCheckMicEnergy()
model.startCheckMicEnergy(
self.progressBarMicEnergy,
)
return {"status":200, "result": config.SELECTED_MIC_DEVICE}
@staticmethod
@@ -784,16 +810,18 @@ class Controller:
def setEnableAutoSpeakerSelect(self, *args, **kwargs) -> dict:
config.AUTO_SPEAKER_SELECT = True
device_manager.setCallbackPrevUpdateDevices(self.prevUpdateSelectedDevices)
device_manager.setCallbackProcessBeforeUpdateDevices(self.stopAccessDevices)
device_manager.setCallbackDefaultSpeakerDevice(self.updateSelectedSpeakerDevice)
device_manager.noticeDefaultDevice()
device_manager.forceSetSpeakerDefaultDevice()
device_manager.setCallbackProcessAfterUpdateDevices(self.restartAccessDevices)
device_manager.noticeUpdateDevices()
device_manager.setSpeakerDefaultDevice()
return {"status":200, "result":config.AUTO_SPEAKER_SELECT}
@staticmethod
def setDisableAutoSpeakerSelect(*args, **kwargs) -> dict:
device_manager.clearCallbackPrevUpdateDevices()
device_manager.clearCallbackProcessBeforeUpdateDevices()
device_manager.clearCallbackDefaultSpeakerDevice()
device_manager.clearCallbackProcessAfterUpdateDevices()
config.AUTO_SPEAKER_SELECT = False
return {"status":200, "result":config.AUTO_SPEAKER_SELECT}
@@ -801,12 +829,13 @@ class Controller:
def getSelectedSpeakerDevice(*args, **kwargs) -> dict:
return {"status":200, "result":config.SELECTED_SPEAKER_DEVICE}
@staticmethod
def setSelectedSpeakerDevice(data, *args, **kwargs) -> dict:
def setSelectedSpeakerDevice(self, data, *args, **kwargs) -> dict:
config.SELECTED_SPEAKER_DEVICE = data
if config.ENABLE_CHECK_ENERGY_RECEIVE is True:
model.stopCheckSpeakerEnergy()
model.startCheckSpeakerEnergy()
model.startCheckSpeakerEnergy(
self.progressBarSpeakerEnergy,
)
return {"status":200, "result":config.SELECTED_SPEAKER_DEVICE}
@staticmethod