(
self,
omni_work_dir: str,
whep_url: str,
session_id: str,
account: str,
config_files: list[str],
config_schema_path: str,
seg_duration: float,
model_runner,
huoshan_tts_voice_type,
stream_config: dict,
)
| 84 | |
| 85 | class ChatAdapter: |
| 86 | def __init__( |
| 87 | self, |
| 88 | omni_work_dir: str, |
| 89 | whep_url: str, |
| 90 | session_id: str, |
| 91 | account: str, |
| 92 | config_files: list[str], |
| 93 | config_schema_path: str, |
| 94 | seg_duration: float, |
| 95 | model_runner, |
| 96 | huoshan_tts_voice_type, |
| 97 | stream_config: dict, |
| 98 | ): |
| 99 | assert os.path.exists(omni_work_dir), f"OMNI work directory {omni_work_dir} does not exist" |
| 100 | self.omni_work_dir = omni_work_dir |
| 101 | self.stream_config = stream_config |
| 102 | self.context = zmq.Context() |
| 103 | self.w2f_socket = self.context.socket(zmq.PULL) |
| 104 | self.w2f_url = ChatAdapter.select_and_bind(self.w2f_socket) |
| 105 | self.f2w_socket = self.context.socket(zmq.PUSH) |
| 106 | self.f2w_url = ChatAdapter.select_and_bind(self.f2w_socket) |
| 107 | self.recv_thread = None |
| 108 | self.audio_buffer = ByteBuffer() |
| 109 | self.audio_info = None |
| 110 | self.chat_server_cmd = [ |
| 111 | os.path.join(self.omni_work_dir, "bin", "seko-chatter"), |
| 112 | "--session-id", |
| 113 | session_id, |
| 114 | "--account", |
| 115 | account, |
| 116 | "--whep-server-url", |
| 117 | whep_url, |
| 118 | "--w2f-endpoint", |
| 119 | self.w2f_url, |
| 120 | "--f2w-endpoint", |
| 121 | self.f2w_url, |
| 122 | "--config-files", |
| 123 | *config_files, |
| 124 | ] |
| 125 | override_config = {} |
| 126 | if huoshan_tts_voice_type is not None: |
| 127 | logger.info(f"Use Huoshan TTS voice type: {huoshan_tts_voice_type}") |
| 128 | override_config["TTS"] = { |
| 129 | "default_voice_info": { |
| 130 | "voice_type": huoshan_tts_voice_type, |
| 131 | "provider": "huoshan_stream_tts", |
| 132 | } |
| 133 | } |
| 134 | system_prompt = stream_config.get("system_prompt", "") |
| 135 | if system_prompt: |
| 136 | override_config["model"] = {"system_prompt": system_prompt} |
| 137 | logger.info(f"Omni use custom system prompt: {system_prompt}") |
| 138 | with open(config_schema_path, "r") as f: |
| 139 | schema = json.load(f) |
| 140 | jsonschema.validate(instance=override_config, schema=schema) |
| 141 | if override_config is not None: |
| 142 | self.chat_server_cmd.extend(["--override-config", json.dumps(override_config)]) |
| 143 | self.chatter_proc = None |
nothing calls this directly
no test coverage detected