An example script for ChatdollKit remote control using socket communication

chatdollkit_client.py

import json
import socket
import traceback

class ChatdollKitClient:
    def __init__(self, host: str = "localhost", port: int = 8080) -> None:
        self.host = host
        self.port = port

    def connect(self):
        self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client_socket.connect((self.host, self.port))

    def close(self):
        self.client_socket.close()

    def send_message(self, endpoint: str, operation: str, text: str, priority: int = 10, payloads: dict = None):
        try:
            self.connect()

            message_dict = {
                "Endpoint": endpoint,
                "Operation": operation,
                "Text": text,
                "Priority": priority,
            }
            if payloads:
                message_dict["Payloads"] = payloads
            message = json.dumps(message_dict)

            self.client_socket.sendall((message + "\n").encode("utf-8"))
            print(f"Message sent: {message}")

        except Exception as ex:
            print(f"Failed to send message: {ex}\n{traceback.format_exc()}")

        finally:
            self.close()

    def chat(self, text: str, priority: int = 10):
        self.send_message("dialog", "start", text, priority)
    
    def clear_chat_queue(self, priority: int = 0):
        self.send_message("dialog", "clear", None, priority)

    def model(self, text: str):
        self.send_message("model", "perform", text)

    def comment(self, text: str, name: str = None):
        if name:
            self.send_message("comment", "add", text, payloads={"userName": name})
        else:
            self.send_message("comment", "add", text)


if __name__ == "__main__":
    chatdollkit = ChatdollKitClient()

    while True:
        user_input = input("> ").strip()
        if not user_input:
            continue

        if user_input.startswith("!"):
            chatdollkit.model(user_input[1:].strip())

        elif user_input.startswith("#"):
            name_body = user_input[1:].split("::")
            if len(name_body) > 1:
                chatdollkit.comment(name_body[1].strip(), name_body[0].strip())
            else:
                chatdollkit.comment(user_input[1:].strip())

        else:
            temp = user_input.split("|")
            text = temp[0].strip()

            if len(temp) > 1:
                priority = int(temp[1])
            else:
                priority = 10

            if text == "clear":
                chatdollkit.clear_chat_queue(priority)
            else:
                chatdollkit.chat(text, priority)