じゃぱり書庫

1
/config
2
  └ custom_components
3
      └ sbv2_simple

1
{
2
  "domain": "sbv2_simple",
3
  "name": "Style-Bert-VITS2 Simple",
4
  "documentation": "https://github.com/litagin02/Style-Bert-VITS2",
5
  "dependencies": [],
6
  "codeowners": [],
7
  "requirements": ["requests"],
8
  "version": "1.0.0",
9
  "iot_class": "local_push"
10
}

1
import logging
2
import requests
3
import io
4
import struct
5
import re
6
import voluptuous as vol
7

8
from homeassistant.components.tts import Provider, PLATFORM_SCHEMA, CONF_LANG
9
from homeassistant.const import CONF_HOST, CONF_PORT
10
import homeassistant.helpers.config_validation as cv
11

12
_LOGGER = logging.getLogger(__name__)
13

14
# --- 設定 ---
15
# サーバーのIPアドレスとポートを設定（環境に合わせて変更してください）
16
DEFAULT_HOST = "192.168.x.x"
17
DEFAULT_PORT = 5000
18
DEFAULT_LANG = "ja-JP"
19

20
# 1回のリクエストで送る最大文字数
21
# 100文字を超えるとリクエストが失敗する現象を回避するため、安全マージンを見て75文字に設定
22
MAX_CHARS_PER_REQUEST = 75
23

24
CONF_MODEL_ID = "model_id"
25
CONF_STYLE = "style"
26
CONF_SDP_RATIO = "sdp_ratio"
27
CONF_NOISE = "noise"
28
CONF_LENGTH = "length"
29

30
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
31
    vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
32
    vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
33
    vol.Optional(CONF_MODEL_ID, default=0): cv.positive_int,
34
    vol.Optional(CONF_STYLE, default="Neutral"): cv.string,
35
    vol.Optional(CONF_SDP_RATIO, default=0.2): vol.Coerce(float),
36
    vol.Optional(CONF_NOISE, default=0.6): vol.Coerce(float),
37
    vol.Optional(CONF_LENGTH, default=1.0): vol.Coerce(float),
38
    vol.Optional(CONF_LANG, default=DEFAULT_LANG): cv.string,
39
})
40

41
def get_engine(hass, config, discovery_info=None):
42
    return SBV2Provider(config)
43

44
class SBV2Provider(Provider):
45
    """Style-Bert-VITS2 Provider with Long Text Support"""
46

47
    def __init__(self, config):
48
        self._host = config[CONF_HOST]
49
        self._port = config[CONF_PORT]
50
        self._model_id = config[CONF_MODEL_ID]
51
        self._style = config[CONF_STYLE]
52
        self._sdp_ratio = config[CONF_SDP_RATIO]
53
        self._noise = config[CONF_NOISE]
54
        self._length = config[CONF_LENGTH]
55
        self._lang = config[CONF_LANG]
56
        self.name = "Style-Bert-VITS2"
57

58
    @property
59
    def default_language(self):
60
        return self._lang
61

62
    @property
63
    def supported_languages(self):
64
        return [self._lang]
65

66
    def get_tts_audio(self, message, language, options=None):
67
        """音声を生成する（長文対応版）"""
68

69
        # 1. 文章を分割する
70
        chunks = self._split_text(message, MAX_CHARS_PER_REQUEST)
71
        wav_parts = []
72

73
        # 2. 分割した文章ごとにサーバーへリクエスト
74
        for i, chunk in enumerate(chunks):
75
            _LOGGER.debug(f"Requesting chunk {i+1}/{len(chunks)}: {chunk}")
76
            wav_data = self._request_wav(chunk)
77
            if wav_data:
78
                wav_parts.append(wav_data)
79
            else:
80
                _LOGGER.error(f"Failed to generate audio for chunk: {chunk}")
81
                return None, None
82

83
        if not wav_parts:
84
            return None, None
85

86
        # 3. 複数のWAVデータを1つに結合する
87
        combined_wav = self._combine_wavs(wav_parts)
88
        return "wav", combined_wav
89

90
    def _request_wav(self, text):
91
        """サーバーにリクエストを送る内部関数"""
92
        url = f"http://{self._host}:{self._port}/voice"
93
        params = {
94
            "text": text,
95
            "model_id": self._model_id,
96
            "style": self._style,
97
            "sdp_ratio": self._sdp_ratio,
98
            "noise": self._noise,
99
            "length": self._length,
100
            "language": "JP"
101
        }
102
        try:
103
            # タイムアウトを20秒に設定
104
            response = requests.get(url, params=params, timeout=20)
105
            if response.status_code == 200:
106
                return response.content
107
            return None
108
        except Exception as e:
109
            _LOGGER.error(f"Connection Error: {e}")
110
            return None
111

112
    def _split_text(self, text, max_len):
113
        """句読点などを考慮して文章を分割する"""
114
        parts = re.split(r'([。、！？\n]+)', text)
115
        chunks = []
116
        current_chunk = ""
117

118
        for part in parts:
119
            if len(current_chunk) + len(part) <= max_len:
120
                current_chunk += part
121
            else:
122
                if current_chunk:
123
                    chunks.append(current_chunk)
124

125
                if len(part) > max_len:
126
                    for i in range(0, len(part), max_len):
127
                        chunks.append(part[i:i+max_len])
128
                    current_chunk = ""
129
                else:
130
                    current_chunk = part
131

132
        if current_chunk:
133
            chunks.append(current_chunk)
134

135
        return [c for c in chunks if c.strip()]
136

137
    def _combine_wavs(self, wav_bytes_list):
138
        """
139
        複数のWAVバイナリを1つに結合する
140
        （単純結合ではなく、ヘッダを除去してPCMデータを繋ぎ、ヘッダを再計算する）
141
        """
142
        if len(wav_bytes_list) == 1:
143
            return wav_bytes_list[0]
144

145
        data_chunks = []
146
        fmt_chunk = None
147

148
        for wav_bytes in wav_bytes_list:
149
            try:
150
                data_marker_index = wav_bytes.find(b'data')
151
                if data_marker_index == -1: continue
152

153
                data_size = int.from_bytes(wav_bytes[data_marker_index+4 : data_marker_index+8], 'little')
154
                raw_data = wav_bytes[data_marker_index+8 : data_marker_index+8+data_size]
155

156
                data_chunks.append(raw_data)
157

158
                if fmt_chunk is None:
159
                    fmt_chunk = wav_bytes[:data_marker_index]
160
            except Exception:
161
                continue
162

163
        if fmt_chunk is None or not data_chunks:
164
            return wav_bytes_list[0]
165

166
        combined_data = b''.join(data_chunks)
167
        total_data_len = len(combined_data)
168

169
        file_size = len(fmt_chunk) + 8 + total_data_len
170
        riff_size_bytes = (file_size - 8).to_bytes(4, 'little')
171
        new_header = fmt_chunk[:4] + riff_size_bytes + fmt_chunk[8:]
172

173
        final_wav = new_header + b'data' + total_data_len.to_bytes(4, 'little') + combined_data
174
        return final_wav

1
tts:
2
  - platform: sbv2_simple
3
    host: 192.168.x.x  # サーバーのIPアドレス
4
    port: 5000
5
    model_id: 0
6
    style: "Neutral"   # 必要に応じて "Happy", "Sad" などに変更可能