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

[Add] init時にOSCの起動確認を行う処理を追加

Browse Source 
裏でOSC receive処理は回りっぱなし
This commit is contained in:
misygauziya
2023-07-24 23:50:35 +09:00
parent 8e31472978
commit 36374d6979
2 changed files with 43 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
VRCT.py
View File

@@ -10,7 +10,7 @@ from flashtext import KeywordProcessor
from threading import Thread
from utils import save_json, print_textbox, thread_fnc, get_localized_text, widget_main_window_label_setter
from osc_tools import send_typing, send_message
from osc_tools import send_typing, send_message, send_test_action, receive_osc_parameters
from window_config import ToplevelWindowConfig
from window_information import ToplevelWindowInformation
from languages import transcription_lang, translators, translation_lang, selectable_languages
@@ -77,6 +77,7 @@ class App(CTk):
self.MESSAGE_FORMAT = "[message]([translation])"
# Others
self.ENABLE_AUTO_CLEAR_CHATBOX = False
self.ENABLE_OSC = False
# load config
if os_path.isfile(self.PATH_CONFIG) is not False:
@@ -403,6 +404,14 @@ class App(CTk):
self.config_window = ToplevelWindowConfig(self)
# start receive osc
th_receive_osc_parameters = Thread(target=receive_osc_parameters, args=(self.check_osc_receive,))
th_receive_osc_parameters.daemon = True
th_receive_osc_parameters.start()
# check osc started
send_test_action()
def button_config_callback(self):
self.checkbox_translation.configure(state="disabled")
self.checkbox_transcription_send.configure(state="disabled")
@@ -477,8 +486,12 @@ class App(CTk):
voice_message = self.MESSAGE_FORMAT.replace("[message]", message).replace("[translation]", result)
if self.checkbox_transcription_send.get() is True:
# send OSC message
send_message(voice_message, self.OSC_IP_ADDRESS, self.OSC_PORT)
if self.ENABLE_OSC is True:
# send OSC message
send_message(voice_message, self.OSC_IP_ADDRESS, self.OSC_PORT)
else:
print_textbox(self.textbox_message_log, "OSC is not enabled, please enable OSC and rejoin.", "ERROR")
print_textbox(self.textbox_message_system_log, "OSC is not enabled, please enable OSC and rejoin.", "ERROR")
# update textbox message log
print_textbox(self.textbox_message_log, f"{voice_message}", "SEND")
print_textbox(self.textbox_message_send_log, f"{voice_message}", "SEND")
@@ -647,7 +660,11 @@ class App(CTk):
chat_message = self.MESSAGE_FORMAT.replace("[message]", message).replace("[translation]", result)
# send OSC message
send_message(chat_message, self.OSC_IP_ADDRESS, self.OSC_PORT)
if self.ENABLE_OSC is True:
send_message(chat_message, self.OSC_IP_ADDRESS, self.OSC_PORT)
else:
print_textbox(self.textbox_message_log, "OSC is not enabled, please enable OSC and rejoin.", "ERROR")
print_textbox(self.textbox_message_system_log, "OSC is not enabled, please enable OSC and rejoin.", "ERROR")
# update textbox message log
print_textbox(self.textbox_message_log, f"{chat_message}", "SEND")
@@ -747,6 +764,11 @@ class App(CTk):
widget_main_window_label_setter(self, language_yaml_data)
def check_osc_receive(self, address, osc_arguments):
if self.ENABLE_OSC is False:
self.ENABLE_OSC = True
# print(address, osc_arguments)
if __name__ == "__main__":
try:
app = App()

18
osc_tools.py
View File

@@ -1,5 +1,9 @@
from time import sleep
from typing import List
from pythonosc import osc_message_builder
from pythonosc import udp_client
from pythonosc import dispatcher
from pythonosc import osc_server
# send OSC message typing
def send_typing(flag=False, ip_address="127.0.0.1", port=9000):
@@ -18,4 +22,16 @@ def send_message(message=None, ip_address="127.0.0.1", port=9000):
msg.add_arg(True)
b_msg = msg.build()
client = udp_client.SimpleUDPClient(ip_address, port)
client.send(b_msg)
client.send(b_msg)
def send_test_action(ip_address="127.0.0.1", port=9000):
client = udp_client.SimpleUDPClient(ip_address, port)
client.send_message("/input/Vertical", 1)
sleep(0.01)
client.send_message("/input/Vertical", False)
def receive_osc_parameters(target, filter="/*", ip_address="127.0.0.1", port=9001):
_dispatcher = dispatcher.Dispatcher()
_dispatcher.map(filter, target)
server = osc_server.ThreadingOSCUDPServer((ip_address, port), _dispatcher)
server.serve_forever()