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

👍️[Update] Model : OSC の処理を見直し

Browse Source
This commit is contained in:
misyaguziya
2024-11-04 06:27:08 +09:00
parent f380616bdb
commit e21661df1e
4 changed files with 108 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
src-python/models/osc/osc_tools.py
View File

@@ -1,112 +1,98 @@
import asyncio
from typing import Any
from time import sleep
from pythonosc import osc_message_builder, udp_client, dispatcher, osc_server
from threading import Thread
from pythonosc import udp_client, dispatcher, osc_server
from tinyoscquery.queryservice import OSCQueryService
from tinyoscquery.query import OSCQueryBrowser, OSCQueryClient
from tinyoscquery.utility import get_open_udp_port, get_open_tcp_port
from psutil import process_iter
from tinyoscquery.shared.node import OSCAccess
# send OSC message typing
def sendTyping(flag=False, ip_address="127.0.0.1", port=9000):
typing = osc_message_builder.OscMessageBuilder(address="/chatbox/typing")
typing.add_arg(flag)
b_typing = typing.build()
client = udp_client.SimpleUDPClient(ip_address, port)
client.send(b_typing)
class OSCHandler:
def __init__(self, ip_address="127.0.0.1", port=9000) -> None:
self.osc_ip_address = ip_address
self.osc_port = port
self.osc_parameter_muteself = "/avatar/parameters/MuteSelf"
self.osc_parameter_chatbox_typing = "/chatbox/typing"
self.osc_parameter_chatbox_input = "/chatbox/input"
self.udp_client = udp_client.SimpleUDPClient(self.osc_ip_address, self.osc_port)
self.osc_server_name = "VRChat-Client"
self.osc_server = None
self.osc_query_service = None
self.osc_query_service_name = "VRCT"
self.osc_server_ip_address = ip_address
self.http_port = None
self.osc_server_port = None
# send OSC message
def sendMessage(message=None, ip_address="127.0.0.1", port=9000):
if message is not None:
msg = osc_message_builder.OscMessageBuilder(address="/chatbox/input")
msg.add_arg(f"{message}")
msg.add_arg(True)
msg.add_arg(True)
b_msg = msg.build()
client = udp_client.SimpleUDPClient(ip_address, port)
client.send(b_msg)
def setOscIpAddress(self, ip_address:str) -> None:
self.osc_ip_address = ip_address
self.udp_client = udp_client.SimpleUDPClient(self.osc_ip_address, self.osc_port)
def sendTestAction(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 setOscPort(self, port:int) -> None:
self.osc_port = port
self.udp_client = udp_client.SimpleUDPClient(self.osc_ip_address, self.osc_port)
# send Input Voice
def sendInputVoice(flag=False, ip_address="127.0.0.1", port=9000):
input_voice = osc_message_builder.OscMessageBuilder(address="/input/Voice")
input_voice.add_arg(flag)
b_input_voice = input_voice.build()
client = udp_client.SimpleUDPClient(ip_address, port)
client.send(b_input_voice)
# send OSC message typing
def sendTyping(self, flag:bool=False) -> None:
self.udp_client.send_message(self.osc_parameter_chatbox_typing, [flag])
def sendChangeVoice(ip_address="127.0.0.1", port=9000):
sendInputVoice(flag=0, ip_address=ip_address, port=port)
sleep(0.05)
sendInputVoice(flag=1, ip_address=ip_address, port=port)
sleep(0.05)
sendInputVoice(flag=0, ip_address=ip_address, port=port)
sleep(0.05)
# send OSC message
def sendMessage(self, message:str="", notification:bool=True) -> None:
if len(message) > 0:
self.udp_client.send_message(self.osc_parameter_chatbox_input, [f"{message}", True, notification])
def getOSCParameterValue(address, server_name="VRChat-Client"):
value = None
try:
browser = OSCQueryBrowser()
sleep(1)
service = browser.find_service_by_name(server_name)
if service is not None:
oscq = OSCQueryClient(service)
mute_self_node = oscq.query_node(address)
value = mute_self_node.value[0]
browser.zc.close()
browser.browser.cancel()
except Exception:
pass
return value
def checkVRChatRunning() -> bool:
_proc_name = "VRChat.exe"
return _proc_name in (p.name() for p in process_iter())
def receiveOscParameters(dict_filter_and_target, ip_address="127.0.0.1", title="VRCT"):
while True:
if not checkVRChatRunning():
def getOSCParameterValue(self, address:str) -> Any:
value = None
try:
browser = OSCQueryBrowser()
sleep(1)
else:
try:
osc_port = get_open_udp_port()
http_port = get_open_tcp_port()
osc_dispatcher = dispatcher.Dispatcher()
for filter, target in dict_filter_and_target.items():
osc_dispatcher.map(filter, target)
osc_udp_server = osc_server.ThreadingOSCUDPServer((ip_address, osc_port), osc_dispatcher)
service = browser.find_service_by_name(self.osc_server_name)
if service is not None:
osc_query_client = OSCQueryClient(service)
mute_self_node = osc_query_client.query_node(address)
value = mute_self_node.value[0]
browser.zc.close()
browser.browser.cancel()
osc_client = OSCQueryService(title, http_port, osc_port)
for filter, target in dict_filter_and_target.items():
osc_client.advertise_endpoint(filter)
except Exception:
pass
return value
osc_udp_server.serve_forever()
except Exception:
pass
def getOSCParameterMuteSelf(self) -> bool:
return self.getOSCParameterValue(self.osc_parameter_muteself)
def receiveOscParameters(self, dict_filter_and_target:dict) -> None:
self.osc_server_port = get_open_udp_port()
self.http_port = get_open_tcp_port()
osc_dispatcher = dispatcher.Dispatcher()
for filter, target in dict_filter_and_target.items():
osc_dispatcher.map(filter, target)
self.osc_server = osc_server.ThreadingOSCUDPServer((self.osc_server_ip_address, self.osc_server_port), osc_dispatcher, asyncio.get_event_loop())
Thread(target=self.oscServerServe, daemon=True).start()
self.osc_query_service = OSCQueryService(self.osc_query_service_name, self.http_port, self.osc_server_port)
for filter, target in dict_filter_and_target.items():
self.osc_query_service.advertise_endpoint(filter, access=OSCAccess.READWRITE_VALUE)
def oscServerServe(self) -> None:
self.osc_server.serve_forever(100)
def oscServerStop(self) -> None:
if isinstance(self.osc_server, osc_server.ThreadingOSCUDPServer):
self.osc_server.shutdown()
self.osc_server = None
if isinstance(self.osc_query_service, OSCQueryService):
self.osc_query_service.http_server.shutdown()
self.osc_query_service = None
if __name__ == "__main__":
osc_parameter_prefix = "/avatar/parameters/"
osc_avatar_change_path = "/avatar/change"
param_MuteSelf = "MuteSelf"
param_Voice = "Voice"
def print_handler_all(address, *args):
print(f"all {address}: {args}")
def print_handler_muteself(address, *args):
print(f"muteself {address}: {args}")
def print_handler_voice(address, *args):
print(f"voice {address}: {args}")
dict_filter_and_target = {
# osc_parameter_prefix + "*": print_handler_all,
osc_parameter_prefix + param_MuteSelf: print_handler_muteself,
osc_parameter_prefix + param_Voice: print_handler_voice,
}
receiveOscParameters(dict_filter_and_target)
handler = OSCHandler()
handler.receiveOscParameters({
"/avatar/parameters/MuteSelf": print,
})
sleep(5)
handler.sendTyping(True)
sleep(1)
handler.sendMessage(message="Hello World", notification=True)
sleep(60)
handler.oscServerStop()