init

custom_components/tencentcloud_asr/__init__.py

@@ -0,0 +1,16 @@
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+DOMAIN = "tencentcloud_asr"
+
+PLATFORMS = (Platform.STT,)
+
+
+async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
+    await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
+    return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS)

custom_components/tencentcloud_asr/config_flow.py

@@ -0,0 +1,44 @@
+import homeassistant.helpers.config_validation as cv
+import voluptuous as vol
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from typing import Any, Dict
+import aiohttp
+import logging
+from tencentcloud_api import Modules
+
+from . import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def validate_path(path: str) -> None:
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(path) as response:
+                _LOGGER.info(f'response.status {response.status}')
+                if response.status != 200:
+                    raise ValueError
+                await response.text()
+    except Exception as e:
+        _LOGGER.error('validate_path', e)
+        raise ValueError
+
+
+class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
+    async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
+        errors: Dict[str, str] = {}
+        if user_input:
+            return self.async_create_entry(title=DOMAIN, data=user_input)
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required("secretid"): str,
+                    vol.Required("secretkey"): str,
+                    vol.Optional("model", default="16k_zh"): vol.In(Modules.keys()),
+                },
+            ),
+            errors=errors
+        )

custom_components/tencentcloud_asr/manifest.json

@@ -0,0 +1,20 @@
+{
+    "domain": "tencentcloud_asr",
+    "name": "TencnetCloud Asr",
+    "codeowners": [
+        "@chliny"
+    ],
+    "config_flow": true,
+    "dependencies": [],
+    "documentation": "https://github.com/chliny/hass-tencentcloud-asr",
+    "iot_class": "cloud_polling",
+    "issue_tracker": "https://github.com/chliny/hass-tencentcloud-asr/issues",
+    "requirements": [
+        "tencentcloud-sdk-python-asr",
+        "tencentcloud-sdk-python-common"
+    ],
+    "loggers": [
+        "tencentcloud_asr"
+    ],
+    "version": "0.0.1"
+}

custom_components/tencentcloud_asr/stt.py

@@ -0,0 +1,74 @@
+import logging
+import base64
+from collections.abc import AsyncIterable
+
+from homeassistant.components import stt
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from tencentcloud_api import TencentCloudAsrAPi
+from tencentcloud_api import ModuleSupportLanguage
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    async_add_entities([ASRSTT(hass, config_entry)])
+
+
+class ASRSTT(stt.SpeechToTextEntity):
+    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+        secretId = config_entry.data.get("secretid", "")
+        secretKey = config_entry.data.get("secretKey", "")
+        self.tencentCloudApi = TencentCloudAsrAPi(secretId, secretKey)
+
+        self.model: str = config_entry.data.get("model", "16k_zh")
+        # self._attr_name = f"TencentCloud Asr id: ({secretId})"
+        self._attr_unique_id = f"{config_entry.entry_id[:7]}-tencentcloud-asr"
+
+    @property
+    def supported_languages(self) -> list[str]:
+        return ModuleSupportLanguage.get(self.model, ["zh"])
+
+    @property
+    def supported_formats(self) -> list[stt.AudioFormats]:
+        return [stt.AudioFormats.WAV]
+
+    @property
+    def supported_codecs(self) -> list[stt.AudioCodecs]:
+        return [stt.AudioCodecs.PCM]
+
+    @property
+    def supported_bit_rates(self) -> list[stt.AudioBitRates]:
+        return [stt.AudioBitRates.BITRATE_16]
+
+    @property
+    def supported_sample_rates(self) -> list[stt.AudioSampleRates]:
+        return [stt.AudioSampleRates.SAMPLERATE_16000]
+
+    @property
+    def supported_channels(self) -> list[stt.AudioChannels]:
+        return [stt.AudioChannels.CHANNEL_MONO]
+
+    async def async_process_audio_stream(
+        self, metadata: stt.SpeechMetadata, stream: AsyncIterable[bytes]
+    ) -> stt.SpeechResult:
+        _LOGGER.debug("process_audio_stream start")
+
+        audio = b""
+        async for chunk in stream:
+            audio += chunk
+
+        data = base64.b64encode(audio).decode("utf8", "ignore").strip()
+        _LOGGER.debug(f"process_audio_stream transcribe: {data}")
+        ret, result = self.tencentCloudApi.SentenceRecognition(self.model, data)
+        if not ret:
+            return stt.SpeechResult(result, stt.SpeechResultState.ERROR)
+        if not result:
+            return stt.SpeechResult("未识别到有效语音", stt.SpeechResultState.SUCCESS)
+
+        return stt.SpeechResult(result, stt.SpeechResultState.SUCCESS)

custom_components/tencentcloud_asr/tencentcloud_api.py

@@ -0,0 +1,54 @@
+import logging
+from homeassistant.components.stt import AudioFormats
+from tencentcloud.common import credential
+from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
+from tencentcloud.asr.v20190614.asr_client import AsrClient
+from tencentcloud.asr.v20190614.models import SentenceRecognitionRequest, SentenceRecognitionResponse
+
+_LOGGER = logging.getLogger(__name__)
+
+
+SentenceRecognitionSourceTypePost = 1
+SentenceRecognitionSourceTypeURL = 0
+Modules = {
+    "16k_zh": "中文",
+    "16k_zh-PY": "中英粤",
+    "16k_en": "英语",
+    "16k_yue": "粤语",
+    "16k_zh_dialect": "多方言",
+}
+ModuleSupportLanguage = {
+    "16k_zh": ["zh"],
+    "16k_zh-PY": ["zh", "en", "yue"],
+    "16k_en": ["en"],
+    "16k_yue": ["yue"],
+    "16k_zh_dialect": ["zh"],
+}
+
+
+class TencentCloudAsrAPi(object):
+    def __init__(self, secretId, secretKey):
+        try:
+            cred = credential.Credential(secretId, secretKey)
+            region = ""
+            self.asrClient = AsrClient(cred, region)
+        except TencentCloudSDKException as err:
+            _LOGGER.error("tencent cloud sdk err:%s", err.message)
+            raise err
+
+    def SentenceRecognition(self, engSerViceType, data: str) -> (tuple[bool, str | None]):
+        req = SentenceRecognitionRequest()
+        req.EngSerViceType = engSerViceType
+        req.SourceType = SentenceRecognitionSourceTypePost
+        req.VoiceFormat = AudioFormats.WAV
+        req.ProjectId = 0
+        req.SubServiceType = 2
+        req.UsrAudioKey = ""
+        req.Data = data
+        req.DataLen = len(data)
+        try:
+            res: SentenceRecognitionResponse = self.asrClient.SentenceRecognition(req)
+            return True, res.Result
+        except TencentCloudSDKException as err:
+            _LOGGER.error("recognition failed:%s", err.message)
+            return False, str(err)