ali:
VTS_PORT = 8001
MOUTH_PARAM = “MouthOpen” # VTuber mouth parameter
TOKEN_FILE = “vts_token.json”
class VTubeStudioClient:
def \__init_\_(self):
self.url = f"ws://localhost:{VTS_PORT}"
self.authenticated = False
self.token = self.load_token()
self.\_auth_requested = False # Prevent repeated auth requests
self.\_connect()
def \_connect(self):
clean_print(f"\[cyan\]Connecting to VTube Studio on {self.url}...\[/cyan\]")
self.ws = websocket.WebSocketApp(
self.url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
self.thread = threading.Thread(target=self.ws.run_forever, daemon=True)
self.thread.start()
def on_open(self, ws):
clean_print("\[green\]Connected to VTube Studio API\[/green\]")
time.sleep(1)
self.\_auth_requested = False # Reset on new connection
if self.token:
self.send({
"apiName": "VTubeStudioPublicAPI",
"apiVersion": "1.0",
"requestID": "auth",
"messageType": "AuthenticationRequest",
"data": {
"pluginName": "Local AI VTuber",
"pluginDeveloper": "Bro77xp",
"authenticationToken": self.token
}
})
elif not self.\_auth_requested:
self.\_auth_requested = True
self.send({
"apiName": "VTubeStudioPublicAPI",
"apiVersion": "1.0",
"requestID": "auth_token",
"messageType": "AuthenticationTokenRequest",
"data": {
"pluginName": "Local AI VTuber",
"pluginDeveloper": "Bro77xp",
"pluginIcon": ""
}
})
def on_message(self, ws, message):
try:
msg = json.loads(message)
mtype = msg.get("messageType", "")
if mtype == "AuthenticationTokenResponse":
token = msg\["data"\]\["authenticationToken"\]
self.save_token(token)
clean_print("\[green\]✅ Token received and saved. Please restart the script.\[/green\]")
self.authenticated = True
elif mtype == "AuthenticationResponse":
self.authenticated = True
clean_print("\[green\]Authenticated with VTube Studio!\[/green\]")
except Exception as e:
clean_print(f"\[red\]Error parsing message:\[/red\] {e}")
def on_error(self, ws, error):
clean_print(f"\[red\]VTS WebSocket error:\[/red\] {error}")
def on_close(self, ws, close_status_code, close_msg):
now = time.time()
if not hasattr(self, '\_last_close_log') or now - self.\_last_close_log > 30:
self.\_last_close_log = now
clean_print("\[yellow\]VTS connection closed. Retrying in 5 seconds...\[/yellow\]")
time.sleep(5)
self.\_connect()
def send(self, payload):
try:
if self.ws and self.ws.sock and self.ws.sock.connected:
self.ws.send(json.dumps(payload))
else:
now = time.time()
if not hasattr(self, '\_last_not_connected_log') or now - self.\_last_not_connected_log > 30:
self.\_last_not_connected_log = now
clean_print("\[yellow\]VTS not connected; could not send message (suppressing for 30s)\[/yellow\]")
except Exception as e:
clean_print(f"\[red\]Error sending to VTS:\[/red\] {e}")
def trigger_hotkey(self, hotkey_id, transition_duration: float = 2.0):
"""Trigger a hotkey with optional transition duration (seconds) for smooth animation fade."""
self.send({
"apiName": "VTubeStudioPublicAPI",
"apiVersion": "1.0",
"requestID": f"hotkey\_{hotkey_id}",
"messageType": "HotkeyTriggerRequest",
"data": {
"hotkeyID": hotkey_id,
"transitionDuration": transition_duration
}
})
def inject_parameters(self, params: dict, request_id: str = "inject"):
"""Inject multiple Live2D parameters in one request.
params: {param_id: value, ...} e.g. {"FaceAngleX": 15.0, "EyeOpenLeft": 0.5}
"""
values = \[{"id": k, "value": float(v)} for k, v in params.items()\]
self.send({
"apiName": "VTubeStudioPublicAPI",
"apiVersion": "1.0",
"requestID": request_id,
"messageType": "InjectParameterDataRequest",
"data": {"parameterValues": values}
})
def set_mouth(self, value):
self.send({
"apiName": "VTubeStudioPublicAPI",
"apiVersion": "1.0",
"requestID": "mouth",
"messageType": "InjectParameterDataRequest",
"data": {
"parameterValues": \[{
"id": MOUTH_PARAM,
"value": float(value)
}\]
}
})
def save_token(self, token):
with open(TOKEN_FILE, "w") as f:
json.dump({"token": token}, f)
def load_token(self):
if os.path.exists(TOKEN_FILE):
with open(TOKEN_FILE, "r") as f:
return json.load(f).get("token")
return None
vtube = VTubeStudioClient()
idle_anim = IdleAnimationEngine(vtube, lambda: tts_active or tts_generating or tts_playing)
hara:
VTS_PORT = 8003
MOUTH_PARAM = “MouthOpen”
TOKEN_FILE = “vts_token1.json”
class VTubeStudioClient:
API_NAME = "VTubeStudioPublicAPI"
API_VERSION = "1.0"
def \__init_\_(self):
self.url = f"ws://localhost:{VTS_PORT}"
self.ws = None
self.thread = None
self.authenticated = False
self.token = self.load_token()
self.hotkeys = {}
self.\_last_mouth = -1.0
self.\_last_mouth_time = 0.0
self.\_connect()
try:
print("Authenticated:", self.authenticated)
except Exception:
pass
def \_connect(self):
console.print(f"\[cyan\]Connecting to VTube Studio on {self.url}...\[/cyan\]")
self.ws = websocket.WebSocketApp(
self.url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
self.thread = threading.Thread(target=self.ws.run_forever, daemon=True)
self.thread.start()
def on_open(self, ws):
console.print("\[green\]Connected to VTube Studio API\[/green\]")
time.sleep(1)
if self.token:
console.print("\[cyan\]Attempting to authenticate with stored token...\[/cyan\]")
self.send_request(
"auth",
"AuthenticationRequest",
{
"pluginName": "Local AI VT3.0",
"pluginDeveloper": "Bro77xp",
"authenticationToken": self.token
}
)
else:
console.print("\[cyan\]No stored token found. Requesting authentication token...\[/cyan\]")
self.send_request(
"auth_token",
"AuthenticationTokenRequest",
{
"pluginName": "Local AI VT3.0",
"pluginDeveloper": "Bro77xp",
"pluginIcon": ""
}
)
def on_message(self, ws, message):
try:
msg = json.loads(message)
mtype = msg.get("messageType", "")
if mtype == "AuthenticationTokenResponse":
token = msg\["data"\]\["authenticationToken"\]
self.save_token(token)
console.print("\[green\]✅ Token received and saved. Authenticating...\[/green\]")
self.send_request(
"auth",
"AuthenticationRequest",
{
"pluginName": "Local AI VT3.0",
"pluginDeveloper": "Bro77xp",
"authenticationToken": token
}
)
elif mtype == "AuthenticationResponse":
self.authenticated = True
console.print("\[green\]Authenticated with VTube Studio!\[/green\]")
except Exception as e:
console.print(f"\[red\]Error parsing message:\[/red\] {e}")
def on_error(self, ws, error):
console.print(f"\[red\]VTS WebSocket error:\[/red\] {error}")
def on_close(self, ws, close_status_code, close_msg):
console.print("\[yellow\]VTS connection closed. Retrying in 5 seconds...\[/yellow\]")
time.sleep(5)
self.\_connect()
def save_token(self, token):
with open(TOKEN_FILE, "w") as f:
json.dump({"token": token}, f)
def send_request(
self,
request_id: str,
message_type: str,
data: dict | None = None
):
if not self.is_connected():
return False
payload = {
"apiName": self.API_NAME,
"apiVersion": self.API_VERSION,
"requestID": request_id,
"messageType": message_type,
}
if data:
payload\["data"\] = data
try:
self.ws.send(json.dumps(payload))
return True
except Exception as e:
console.print(f"\[red\]VTS Send Error:\[/red\] {e}")
return False
def is_connected(self):
return (
self.ws
and self.ws.sock
and self.ws.sock.connected
)
def load_token(self):
try:
if not os.path.exists(TOKEN_FILE):
return None
with open(TOKEN_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
return data.get("token") or data.get("authenticationToken")
except Exception as e:
console.print(f"\[yellow\]Could not load token: {e}\[/yellow\]")
return None
source & further reading
discuss.huggingface.co — original article
Rakarrack-0.6.1 port making progress! ( AI assisted )
Cloud Storage Poll
Welcome to Haiku basic(Haiku Docs, Haiku slide and Haiku sheets)