![vosk-ASR asterisk-ari调用[AI人工智能(五十四)]—东方仙盟](images/news1.jpg)
目录结构核心代码完整代码#!/usr/bin/python3 import anyio import asyncari import logging import aioudp import os import vosk import array ast_host os.getenv(AST_HOST, 127.0.0.1) ast_port int(os.getenv(AST_ARI_PORT, 8088)) ast_url os.getenv(AST_URL, http://%s:%d/%(ast_host,ast_port)) ast_username os.getenv(AST_USER, asterisk) ast_password os.getenv(AST_PASS, asterisk) ast_app os.getenv(AST_APP, hello-world) model vosk.Model(langen-us) channels {} class Channel: async def rtp_handler(self, connection): async for message in connection: data array.array(h, message[12:]) data.byteswap() if self.rec.AcceptWaveform(data.tobytes()): res self.rec.Result() else: res self.rec.PartialResult() print(res) async def init(self, client, channel): self.port 45000 len(channels) self.rec vosk.KaldiRecognizer(model, 16000) self.udp aioudp.serve(127.0.0.1, self.port, self.rtp_handler) await self.udp.__aenter__() bridge await client.bridges.create(typemixing) media_id client.generate_id() await client.channels.externalMedia(channelIdmedia_id, appclient._app, external_host127.0.0.1: str(self.port), formatslin16) await bridge.addChannel(channel[media_id, channel.id]) async def statis_handler(objs, ev, client): channel objs[channel] channel.answer() if UnicastRTP in channel.name: return local_channel Channel() await local_channel.init(client, channel) channels[channel.id] local_channel async def main(): async with asyncari.connect(ast_url, ast_app, ast_username,ast_password) as client: async with client.on_channel_event(StasisStart) as listener: async for objs, event in listener: await statis_handler(objs, event, client) if __name__ __main__: logging.basicConfig(levellogging.DEBUG) anyio.run(main)调用方法This is an example of channel transcription through ARI/externalMedia To configure statis make sure modules are loaded: load res_rtp_multicast.so load res_ari.so ; Asterisk RESTful Interface load res_ari_applications load res_ari_asterisk.so load res_ari_bridges load res_ari_channels.so load res_ari_device_states.so ; RESTful API module - Device state resources load res_ari_endpoints.so load res_ari_events.so load res_ari_model.so ; ARI Model validators load res_ari_playbacks.so load res_ari_recordings.so load res_ari_sounds load res_stasis.so ; Stasis application support load app_stasis.so ; Stasis dialplan application load res_stasis_answer.so load res_stasis_device_state.so ; Stasis application device state support load res_stasis_playback.so ; Stasis application playback support load res_stasis_recording.so ; Stasis application recording support load res_stasis_snoop.so ; Stasis application snoop support then check ari.conf: [general] enabled yes pretty yes allowed_origins * [asterisk] type user read_only no password asterisk Then the dialplan exten 4,1,Answer() same n,Stasis(hello-world) same n,Hangup() Install required python modules pip3 install aioudp asyncari vosk And start the statis app: python3 vosk_ari.py Dial the extension, you should see the transcript on console示意图实现原理一、整体功能仙盟视角总览这段内容本质是搭建一套「仙盟语音传令系统」Asterisk星号服务器是仙盟总坛负责接收各路弟子电话的语音传令ARIAsterisk RESTful 接口是仙盟传讯令符让外部弟子Python 程序能和总坛互通消息Stasis 是总坛议事堂专门处理需要特殊处理的传令比如语音转写Vosk 是仙盟译经师把弟子的口语传令语音转成文字转写整个流程弟子拨打总坛分机号→总坛接起→把传令转到议事堂→译经师实时翻译→把文字版传令显示在总坛控制台。二、逐段拆解仙盟场景对应解释1. 加载 Asterisk 模块给总坛配备法器plaintextload res_rtp_multicast.so load res_ari.so ; Asterisk RESTful Interface ... load res_stasis_snoop.so ; Stasis application snoop support比喻这些load xxx.so是给仙盟总坛配备核心法器 ——res_rtp_multicast.so负责语音信号传输的「传音法器」res_ari.so及相关激活「传讯令符ARI」让外部程序能和总坛对话res_stasis.so/app_stasis.so搭建「议事堂Stasis」专门处理特殊传令其他res_stasis_xxx.so给议事堂配辅助人员处理接听、录音、监听等。作用只有加载这些模块Asterisk 才能支持 ARI 接口和 Stasis 应用是后续功能的基础。2. 配置 ari.conf给传讯令符设使用规则plaintext[general] enabled yes ; 启用传讯令符 pretty yes ; 令符返回的消息格式工整方便人看 allowed_origins * ; 允许所有外部弟子使用令符无限制 [asterisk] type user ; 定义使用令符的弟子身份 read_only no ; 弟子不仅能看消息还能发指令 password asterisk ; 使用令符的密码简单密码仅作示例比喻给「传讯令符ARI」定规矩 —— 允许使用、谁能用、能做什么。作用开启 ARI 接口设置访问权限和认证信息让 Python 程序能通过 ARI 控制 Asterisk。3. 配置拨号计划给总坛设接令规则plaintextexten 4,1,Answer() ; 弟子拨打分机号4时总坛先接起传令 same n,Stasis(hello-world) ; 把这个传令转到「hello-world议事堂」处理 same n,Hangup() ; 处理完后挂断传令比喻仙盟总坛的接令规则 —— 只要有人打分机 4先接电话再把这个通话交给名为hello-world的议事堂处理最后挂电话。作用定义电话呼入后的处理逻辑核心是把通话交给 Stasis 应用让外部 Python 程序能接管通话的语音处理。4. 安装 Python 模块给译经师配工具plaintextpip3 install aioudp asyncari vosk比喻给「译经师Vosk」配备工作工具 ——vosk译经师的核心本领语音转文字asyncari让译经师能看懂「传讯令符ARI」的消息格式aioudp译经师接收语音数据的「收音法器」。作用安装 Python 依赖库让后续的vosk_ari.py程序能通过 ARI 连接 Asterisk并调用 Vosk 实现语音转写。5. 启动 Stasis 应用让译经师到议事堂上班plaintextpython3 vosk_ari.py比喻让带着工具的译经师vosk_ari.py程序到「hello-world 议事堂」上班等待处理传令。作用运行 Python 程序该程序会通过 ARI 连接 Asterisk监听 Stasis 应用的事件当有通话进入 Stasis 时实时获取语音流用 Vosk 转写并输出到控制台。6. 最终效果完整流程拨号分机 4 → Asterisk 接起通话 → 把通话交给 Stasis → Python 程序通过 ARI 获取通话语音 → Vosk 转写语音为文字 → 控制台显示转写结果。东方仙盟拥抱知识开源共筑数字新生态在全球化与数字化浪潮中东方仙盟始终秉持开放协作、知识共享的理念积极拥抱开源技术与开放标准。我们相信唯有打破技术壁垒、汇聚全球智慧才能真正推动行业的可持续发展。开源赋能中小商户通过将前端异常检测、跨系统数据互联等核心能力开源化东方仙盟为全球中小商户提供了低成本、高可靠的技术解决方案让更多商家能够平等享受数字转型的红利。共建行业标准我们积极参与国际技术社区与全球开发者、合作伙伴共同制定开放协议 与技术规范推动跨境零售、文旅、餐饮等多业态的系统互联互通构建更加公平、高效的数字生态。知识普惠共促发展通过开源社区 、技术文档与培训体系东方仙盟致力于将前沿技术转化为可落地的行业实践赋能全球合作伙伴共同培育创新人才推动数字经济 的普惠式增长阿雪技术观在科技发展浪潮中我们不妨积极投身技术共享。不满足于做受益者更要主动担当贡献者。无论是分享代码、撰写技术博客还是参与开源项目 维护改进每一个微小举动都可能蕴含推动技术进步的巨大能量。东方仙盟是汇聚力量的天地我们携手在此探索硅基 生命为科技进步添砖加瓦。Hey folks, in this wild tech - driven world, why not dive headfirst into the whole tech - sharing scene? Dont just be the one reaping all the benefits; step up and be a contributor too. Whether youre tossing out your code snippets , hammering out some tech blogs, or getting your hands dirty with maintaining and sprucing up open - source projects, every little thing you do might just end up being a massive force that pushes tech forward. And guess what? The Eastern FairyAlliance is this awesome place where we all come together. Were gonna team up and explore the whole silicon - based life thing, and in the process, well be fueling the growth of technology
vosk-ASR asterisk-ari调用[AI人工智能(五十四)]—东方仙盟
目录结构核心代码完整代码#!/usr/bin/python3 import anyio import asyncari import logging import aioudp import os import vosk import array ast_host os.getenv(AST_HOST, 127.0.0.1) ast_port int(os.getenv(AST_ARI_PORT, 8088)) ast_url os.getenv(AST_URL, http://%s:%d/%(ast_host,ast_port)) ast_username os.getenv(AST_USER, asterisk) ast_password os.getenv(AST_PASS, asterisk) ast_app os.getenv(AST_APP, hello-world) model vosk.Model(langen-us) channels {} class Channel: async def rtp_handler(self, connection): async for message in connection: data array.array(h, message[12:]) data.byteswap() if self.rec.AcceptWaveform(data.tobytes()): res self.rec.Result() else: res self.rec.PartialResult() print(res) async def init(self, client, channel): self.port 45000 len(channels) self.rec vosk.KaldiRecognizer(model, 16000) self.udp aioudp.serve(127.0.0.1, self.port, self.rtp_handler) await self.udp.__aenter__() bridge await client.bridges.create(typemixing) media_id client.generate_id() await client.channels.externalMedia(channelIdmedia_id, appclient._app, external_host127.0.0.1: str(self.port), formatslin16) await bridge.addChannel(channel[media_id, channel.id]) async def statis_handler(objs, ev, client): channel objs[channel] channel.answer() if UnicastRTP in channel.name: return local_channel Channel() await local_channel.init(client, channel) channels[channel.id] local_channel async def main(): async with asyncari.connect(ast_url, ast_app, ast_username,ast_password) as client: async with client.on_channel_event(StasisStart) as listener: async for objs, event in listener: await statis_handler(objs, event, client) if __name__ __main__: logging.basicConfig(levellogging.DEBUG) anyio.run(main)调用方法This is an example of channel transcription through ARI/externalMedia To configure statis make sure modules are loaded: load res_rtp_multicast.so load res_ari.so ; Asterisk RESTful Interface load res_ari_applications load res_ari_asterisk.so load res_ari_bridges load res_ari_channels.so load res_ari_device_states.so ; RESTful API module - Device state resources load res_ari_endpoints.so load res_ari_events.so load res_ari_model.so ; ARI Model validators load res_ari_playbacks.so load res_ari_recordings.so load res_ari_sounds load res_stasis.so ; Stasis application support load app_stasis.so ; Stasis dialplan application load res_stasis_answer.so load res_stasis_device_state.so ; Stasis application device state support load res_stasis_playback.so ; Stasis application playback support load res_stasis_recording.so ; Stasis application recording support load res_stasis_snoop.so ; Stasis application snoop support then check ari.conf: [general] enabled yes pretty yes allowed_origins * [asterisk] type user read_only no password asterisk Then the dialplan exten 4,1,Answer() same n,Stasis(hello-world) same n,Hangup() Install required python modules pip3 install aioudp asyncari vosk And start the statis app: python3 vosk_ari.py Dial the extension, you should see the transcript on console示意图实现原理一、整体功能仙盟视角总览这段内容本质是搭建一套「仙盟语音传令系统」Asterisk星号服务器是仙盟总坛负责接收各路弟子电话的语音传令ARIAsterisk RESTful 接口是仙盟传讯令符让外部弟子Python 程序能和总坛互通消息Stasis 是总坛议事堂专门处理需要特殊处理的传令比如语音转写Vosk 是仙盟译经师把弟子的口语传令语音转成文字转写整个流程弟子拨打总坛分机号→总坛接起→把传令转到议事堂→译经师实时翻译→把文字版传令显示在总坛控制台。二、逐段拆解仙盟场景对应解释1. 加载 Asterisk 模块给总坛配备法器plaintextload res_rtp_multicast.so load res_ari.so ; Asterisk RESTful Interface ... load res_stasis_snoop.so ; Stasis application snoop support比喻这些load xxx.so是给仙盟总坛配备核心法器 ——res_rtp_multicast.so负责语音信号传输的「传音法器」res_ari.so及相关激活「传讯令符ARI」让外部程序能和总坛对话res_stasis.so/app_stasis.so搭建「议事堂Stasis」专门处理特殊传令其他res_stasis_xxx.so给议事堂配辅助人员处理接听、录音、监听等。作用只有加载这些模块Asterisk 才能支持 ARI 接口和 Stasis 应用是后续功能的基础。2. 配置 ari.conf给传讯令符设使用规则plaintext[general] enabled yes ; 启用传讯令符 pretty yes ; 令符返回的消息格式工整方便人看 allowed_origins * ; 允许所有外部弟子使用令符无限制 [asterisk] type user ; 定义使用令符的弟子身份 read_only no ; 弟子不仅能看消息还能发指令 password asterisk ; 使用令符的密码简单密码仅作示例比喻给「传讯令符ARI」定规矩 —— 允许使用、谁能用、能做什么。作用开启 ARI 接口设置访问权限和认证信息让 Python 程序能通过 ARI 控制 Asterisk。3. 配置拨号计划给总坛设接令规则plaintextexten 4,1,Answer() ; 弟子拨打分机号4时总坛先接起传令 same n,Stasis(hello-world) ; 把这个传令转到「hello-world议事堂」处理 same n,Hangup() ; 处理完后挂断传令比喻仙盟总坛的接令规则 —— 只要有人打分机 4先接电话再把这个通话交给名为hello-world的议事堂处理最后挂电话。作用定义电话呼入后的处理逻辑核心是把通话交给 Stasis 应用让外部 Python 程序能接管通话的语音处理。4. 安装 Python 模块给译经师配工具plaintextpip3 install aioudp asyncari vosk比喻给「译经师Vosk」配备工作工具 ——vosk译经师的核心本领语音转文字asyncari让译经师能看懂「传讯令符ARI」的消息格式aioudp译经师接收语音数据的「收音法器」。作用安装 Python 依赖库让后续的vosk_ari.py程序能通过 ARI 连接 Asterisk并调用 Vosk 实现语音转写。5. 启动 Stasis 应用让译经师到议事堂上班plaintextpython3 vosk_ari.py比喻让带着工具的译经师vosk_ari.py程序到「hello-world 议事堂」上班等待处理传令。作用运行 Python 程序该程序会通过 ARI 连接 Asterisk监听 Stasis 应用的事件当有通话进入 Stasis 时实时获取语音流用 Vosk 转写并输出到控制台。6. 最终效果完整流程拨号分机 4 → Asterisk 接起通话 → 把通话交给 Stasis → Python 程序通过 ARI 获取通话语音 → Vosk 转写语音为文字 → 控制台显示转写结果。东方仙盟拥抱知识开源共筑数字新生态在全球化与数字化浪潮中东方仙盟始终秉持开放协作、知识共享的理念积极拥抱开源技术与开放标准。我们相信唯有打破技术壁垒、汇聚全球智慧才能真正推动行业的可持续发展。开源赋能中小商户通过将前端异常检测、跨系统数据互联等核心能力开源化东方仙盟为全球中小商户提供了低成本、高可靠的技术解决方案让更多商家能够平等享受数字转型的红利。共建行业标准我们积极参与国际技术社区与全球开发者、合作伙伴共同制定开放协议 与技术规范推动跨境零售、文旅、餐饮等多业态的系统互联互通构建更加公平、高效的数字生态。知识普惠共促发展通过开源社区 、技术文档与培训体系东方仙盟致力于将前沿技术转化为可落地的行业实践赋能全球合作伙伴共同培育创新人才推动数字经济 的普惠式增长阿雪技术观在科技发展浪潮中我们不妨积极投身技术共享。不满足于做受益者更要主动担当贡献者。无论是分享代码、撰写技术博客还是参与开源项目 维护改进每一个微小举动都可能蕴含推动技术进步的巨大能量。东方仙盟是汇聚力量的天地我们携手在此探索硅基 生命为科技进步添砖加瓦。Hey folks, in this wild tech - driven world, why not dive headfirst into the whole tech - sharing scene? Dont just be the one reaping all the benefits; step up and be a contributor too. Whether youre tossing out your code snippets , hammering out some tech blogs, or getting your hands dirty with maintaining and sprucing up open - source projects, every little thing you do might just end up being a massive force that pushes tech forward. And guess what? The Eastern FairyAlliance is this awesome place where we all come together. Were gonna team up and explore the whole silicon - based life thing, and in the process, well be fueling the growth of technology
