火山引擎实时语音合成WebSocket V3协议Python实现demo
火山引擎语音整体特点
火山引擎(字节跳动旗下)的语音合成产品确实非常面向多媒体内容创作,特别是短视频、有声书和多人场景。
1. 音色多样性与场景细分
火山引擎提供了极其丰富的音色选择(100+音色),并按以下场景精细分类:
多情感音色:支持情感变化,适合短视频叙事
通用场景:基础音色
趣味口音:各地方言口音,非常适合短视频创意内容
角色扮演:大量角色化音色(40+种),明显针对剧情类短视频、对白场景
视频配音:专为视频内容优化的音色
有声阅读:针对长文本朗读的优化音色
2. 适合抖音等短视频平台的特点
丰富的角色音色:如"奶气萌娃"、“病弱少女”、"傲娇霸总"等,这些音色非常适合抖音等平台上的角色扮演、情景剧类短视频
多方言支持:提供北京、台湾、广东、四川等多地方言音色,有助于创作地域特色内容
多情感支持:部分音色支持多种情感变化(如开心、悲伤、生气等),可以制作情感丰富的短视频内容
3. 有声书场景优势
专门的"有声阅读"分类:包含"悬疑解说"、"儒雅青年"等适合长文本朗读的音色
情感化表达:多情感支持使有声书朗读更具表现力
角色化音色:适合对话丰富的小说朗读,可以为不同角色分配不同音色
4. 多人对话场景支持
角色音色的多样性:从"萌娃"到"婆婆",从"病娇弟弟"到"成熟姐姐",提供了几乎所有可能的人物形象
互补性角色设计:如"贴心男友"与"傲娇女友"等角色对,适合多人对话场景
多情感支持:使对话更生动自然
与其他TTS产品的区别
相比其他TTS服务(如微软Azure、Amazon Polly等),火山引擎更加注重:
- 角色化与人设:其他平台往往只提供中性或基础的男女声音色,而火山引擎深度开发了具有鲜明人格特征的音色
- 中文场景优化:特别是中文方言和文化特色的表达 多媒体内容创作导向:明显针对视频内容创作者设计
总结
火山引擎(字节旗下)的TTS模型突出特点是丰富的角色化音色(100+种),特别优化了短视频、有声书和多人对话场景。提供大量具体人设音色(如"傲娇霸总"、“病弱少女”)、多情感表达能力、中文方言支持。
整体更适合有声书,短视频制作等场景,但是目前对于一些具身智能、导览机器人,跨语种的交互机器人上来说,感觉当前提供的模型不太够用。
代码说明
- 基于官方示例,添加注释
- 移除了websocket库的引用
- 修正了正确的音色列表https://www.volcengine.com/docs/6561/1257544
申请语音技术的appid和token
网站链接
https://console.volcengine.com/speech/app
大模型语音合成必选,其他的看自己需求。这个也可以后续来这个控制台修改
然后我们在应用管理就可以看到我们的appid
。
切换到语音合成大模型,下拉,点开显示,就可以获取我们需要的token
了。
后面的代码中我们会用到这两个参数:
appId = "" # 替换为您自己的
token = "" # 替换为您自己的 Token
代码
# -*- coding: utf-8 -*-
# @Time : 2025/4/22 10:28
# @Author : Mark White
# @FileName: vocal_notation.py
# @Software: PyCharm# -*- coding: utf-8 -*-"""
火山引擎实时语音合成 WebSocket V3 协议 Python 示例本脚本演示了如何使用 websockets 库通过 WebSocket V3 协议与火山引擎的实时语音合成服务进行交互。
主要流程包括:
1. 定义协议常量:包括协议版本、消息类型、标志位、序列化方法、压缩方法和事件代码。
2. 定义数据结构:使用 `Header`, `Optional`, `Response` 类来封装协议消息的各个部分。
3. 实现协议辅助函数:- `get_payload_bytes`: 构建符合协议要求的 JSON payload。- `send_event`: 封装发送消息的逻辑,自动添加头部、可选部分和负载。- `read_res_content`, `read_res_payload`: 从响应字节流中读取特定部分(字符串内容、负载)。- `parser_response`: 解析服务器返回的二进制响应,填充 `Response` 对象。- `print_response`: 打印响应对象的详细信息,用于调试。
4. 实现核心 API 调用函数:- `start_connection`: 发送建立连接的请求。- `start_session`: 发送开始一个合成会话的请求。- `send_text`: 发送需要合成的文本。- `finish_session`: 发送结束当前会话的请求。- `finish_connection`: 发送断开连接的请求。
5. 实现主业务逻辑函数 `run_demo`:- 设置必要的认证信息和请求参数。- 建立 WebSocket 连接。- 按顺序调用 API 函数:建立连接 -> 开始会话 -> 发送文本 -> 结束会话 -> 结束连接。- 处理服务器响应,特别是接收音频数据并写入文件。- 进行必要的错误检查。
6. 主程序入口 (`if __name__ == "__main__":`):- 配置用户凭证、合成文本、发音人等参数。- 使用 asyncio 运行主业务逻辑。使用的主要库:
- asyncio: Python 标准库,用于编写异步代码。
- websockets: 一个用于构建 WebSocket 客户端和服务器的库,支持异步操作。
- aiofiles: 提供异步文件操作接口,用于非阻塞地写入音频文件。
- json: Python 标准库,用于处理 JSON 数据(构建 payload)。
- uuid: Python 标准库,用于生成唯一的 session ID 和 connection ID。
"""import asyncio
import json
import uuid# 异步文件操作库
import aiofiles
# WebSocket 客户端库 (注意: 原代码导入了 websocket 和 websockets, 实际主要使用了 websockets)
# import websocket # 这个库在本示例的异步流程中未使用,注释掉以减少混淆
import websockets
from websockets.asyncio.client import ClientConnection # 明确导入 ClientConnection 类型提示# --- 协议常量定义 ---
# 参考文档: https://www.volcengine.com/docs/6561/1329505#%E7%A4%BA%E4%BE%8Bsamples# 协议版本 (固定为 0b0001)
PROTOCOL_VERSION = 0b0001
# 默认头部大小 (固定为 0b0001, 表示 4 字节)
DEFAULT_HEADER_SIZE = 0b0001# --- 消息类型 (Message Type) ---
# 用于标识消息的主要目的# 客户端完整请求 (包含 header, optional, payload)
FULL_CLIENT_REQUEST = 0b0001
# 服务端响应:仅音频数据
AUDIO_ONLY_RESPONSE = 0b1011
# 服务端完整响应 (包含 header, optional, payload)
FULL_SERVER_RESPONSE = 0b1001
# 服务端错误信息
ERROR_INFORMATION = 0b1111# --- 消息类型特定标志位 (Message Type Specific Flags) ---
# 用于提供关于消息类型的额外信息MsgTypeFlagNoSeq = 0b0000 # 非终止数据包,无序列号 (sequence=0)
MsgTypeFlagPositiveSeq = 0b1 # 非终止数据包,有正序列号 (sequence>0)
MsgTypeFlagLastNoSeq = 0b10 # 最后一个数据包,无序列号 (sequence=0)
MsgTypeFlagNegativeSeq = 0b11 # 最后一个数据包,有负序列号 (sequence<0),通常表示错误
MsgTypeFlagWithEvent = 0b100 # 标志位,表示 Optional 部分包含 Event 字段 (用于区分是否携带事件)# --- 消息序列化方法 (Message Serialization) ---
# 指示 Payload 部分的序列化方式NO_SERIALIZATION = 0b0000 # 无序列化 (通常用于纯音频数据)
JSON = 0b0001 # JSON 序列化# --- 消息压缩方法 (Message Compression) ---
# 指示 Payload 部分的压缩方式COMPRESSION_NO = 0b0000 # 无压缩
COMPRESSION_GZIP = 0b0001 # Gzip 压缩 (本示例未使用)# --- 事件代码 (Event Codes) ---
# 用于在 Optional 部分标识具体的业务事件EVENT_NONE = 0 # 无事件# --- 连接管理事件 ---
EVENT_Start_Connection = 1 # 客户端请求建立连接
EVENT_FinishConnection = 2 # 客户端请求关闭连接
EVENT_ConnectionStarted = 50 # 服务端响应:成功建立连接
EVENT_ConnectionFailed = 51 # 服务端响应:建立连接失败 (如认证失败)
EVENT_ConnectionFinished = 52 # 服务端响应:连接已关闭# --- 会话管理事件 (上行 - 客户端发起) ---
EVENT_StartSession = 100 # 客户端请求开始一个 TTS 会话
EVENT_FinishSession = 102 # 客户端请求结束一个 TTS 会话# --- 会话管理事件 (下行 - 服务端响应) ---
EVENT_SessionStarted = 150 # 服务端响应:会话已开始
EVENT_SessionFinished = 152 # 服务端响应:会话已结束
EVENT_SessionFailed = 153 # 服务端响应:会话失败# --- 通用业务事件 (上行 - 客户端发起) ---
EVENT_TaskRequest = 200 # 客户端发送具体的业务请求 (如发送文本进行合成)# --- TTS 业务事件 (下行 - 服务端响应) ---
EVENT_TTSSentenceStart = 350 # 服务端响应:开始处理一个句子
EVENT_TTSSentenceEnd = 351 # 服务端响应:结束处理一个句子
EVENT_TTSResponse = 352 # 服务端响应:返回 TTS 音频数据或相关信息# --- 数据结构类 ---class Header:"""封装 WebSocket V3 协议的消息头部 (Header)。头部固定为 4 字节,包含协议版本、头部大小、消息类型、特定标志、序列化方法、压缩方法等信息。"""def __init__(self,protocol_version=PROTOCOL_VERSION, # 协议版本,默认为 PROTOCOL_VERSIONheader_size=DEFAULT_HEADER_SIZE, # 头部大小,默认为 DEFAULT_HEADER_SIZEmessage_type: int = 0, # 消息类型,如 FULL_CLIENT_REQUESTmessage_type_specific_flags: int = 0, # 消息类型特定标志,如 MsgTypeFlagWithEventserial_method: int = NO_SERIALIZATION, # Payload 序列化方法,如 JSONcompression_type: int = COMPRESSION_NO, # Payload 压缩方法reserved_data=0): # 保留字段,默认为 0"""初始化 Header 对象。参数按协议规范设置默认值或传入指定值。"""self.header_size = header_sizeself.protocol_version = protocol_versionself.message_type = message_typeself.message_type_specific_flags = message_type_specific_flagsself.serial_method = serial_methodself.compression_type = compression_typeself.reserved_data = reserved_datadef as_bytes(self) -> bytes:"""将 Header 对象转换为符合协议规范的 4 字节 bytes 对象。使用位运算将各个字段组合到字节中。Returns:bytes: 包含头部信息的 4 字节 bytes 对象。"""# 第 1 字节: (协议版本 << 4) | 头部大小byte1 = (self.protocol_version << 4) | self.header_size# 第 2 字节: (消息类型 << 4) | 消息类型特定标志byte2 = (self.message_type << 4) | self.message_type_specific_flags# 第 3 字节: (序列化方法 << 4) | 压缩方法byte3 = (self.serial_method << 4) | self.compression_type# 第 4 字节: 保留字段byte4 = self.reserved_data# 组合成 bytes 对象return bytes([byte1, byte2, byte3, byte4])class Optional:"""封装 WebSocket V3 协议的可选部分 (Optional)。这部分内容根据消息头中的标志位和事件类型决定是否存在及其具体内容,通常包含事件代码 (Event)、会话 ID (SessionId)、序列号 (Sequence) 等。"""def __init__(self, event: int = EVENT_NONE, sessionId: str = None, sequence: int = None):"""初始化 Optional 对象。Args:event (int, optional): 事件代码,默认为 EVENT_NONE。sessionId (str, optional): 会话 ID,默认为 None。sequence (int, optional): 序列号,默认为 None。"""self.event = event # 事件代码self.sessionId = sessionId # 会话 IDself.sequence = sequence # 序列号 (本示例中未使用,但协议支持)# 以下字段主要用于解析服务端响应,在客户端发送时通常不设置self.errorCode: int = 0 # 错误码 (来自服务端响应)self.connectionId: str | None = None # 连接 ID (来自服务端响应)self.response_meta_json: str | None = None # 响应元数据 (来自服务端响应, JSON 字符串)def as_bytes(self) -> bytes:"""将 Optional 对象中需要发送的字段(event, sessionId, sequence)转换为 bytes。字段顺序和格式需严格遵守协议规范。- Event: 4 字节大端整数 (如果 event 不是 EVENT_NONE)- SessionId: 4 字节长度 (大端) + SessionId 字符串的 UTF-8 bytes (如果 sessionId 不为 None)- Sequence: 4 字节大端整数 (如果 sequence 不为 None)Returns:bytes: 包含可选部分信息的 bytes 对象。"""option_bytes = bytearray() # 使用 bytearray 方便追加# 如果包含有效事件代码,则添加 Event 字段 (4 字节,大端序,有符号整数)if self.event != EVENT_NONE:# to_bytes(4, "big", signed=True) 将整数转为 4 字节大端序 bytesoption_bytes.extend(self.event.to_bytes(4, "big", signed=True))# 如果包含会话 ID,则添加 SessionId 字段if self.sessionId is not None:# 将 SessionId 字符串编码为 UTF-8 bytessession_id_bytes = str.encode(self.sessionId)# 获取 SessionId bytes 的长度size = len(session_id_bytes)# 添加长度字段 (4 字节,大端序,有符号整数)option_bytes.extend(size.to_bytes(4, "big", signed=True))# 添加 SessionId 的 bytes 内容option_bytes.extend(session_id_bytes)# 如果包含序列号,则添加 Sequence 字段 (4 字节,大端序,有符号整数)if self.sequence is not None:option_bytes.extend(self.sequence.to_bytes(4, "big", signed=True))# 返回最终构建的 bytes 对象return bytes(option_bytes)class Response:"""用于封装从服务器接收到的已解析的响应消息。包含解析后的 Header, Optional 部分,以及可能的 Payload (二进制或 JSON 字符串)。"""def __init__(self, header: Header, optional: Optional):"""初始化 Response 对象。Args:header (Header): 解析得到的 Header 对象。optional (Optional): 解析得到的 Optional 对象。"""self.header = header # 消息头部对象self.optional = optional # 可选部分对象self.payload: bytes | None = None # 负载的原始二进制数据 (如音频)self.payload_json: str | None = None # 负载的 JSON 字符串内容 (如句子开始/结束事件的元信息)def __str__(self):"""提供 Response 对象的基本字符串表示(可选,用于调试)。"""# 可以根据需要自定义更详细的字符串输出return f"Response(Header={self.header.__dict__}, Optional={self.optional.__dict__}, Payload_len={len(self.payload or b'')}, Payload_json={self.payload_json})"# --- 协议辅助函数 ---def get_payload_bytes(uid='1234', event=EVENT_NONE, text='', speaker='', audio_format='mp3',audio_sample_rate=24000) -> bytes:"""构建用于发送给服务器的 JSON Payload,并将其编码为 bytes。根据不同的事件类型和参数填充 JSON 结构。Args:uid (str, optional): 用户 ID。默认为 '1234'。event (int, optional): 当前请求关联的事件代码。默认为 EVENT_NONE。text (str, optional): 需要合成的文本 (仅在 EVENT_TaskRequest 时有效)。默认为 ''。speaker (str, optional): 发音人标识 (在 EVENT_StartSession 和 EVENT_TaskRequest 时需要)。默认为 ''。audio_format (str, optional): 请求的音频格式。默认为 'mp3'。audio_sample_rate (int, optional): 请求的音频采样率。默认为 24000。Returns:bytes: 编码后的 JSON Payload bytes。"""# 构建 Python 字典表示 JSON 结构payload_dict = {"user": {"uid": uid}, # 用户信息"event": event, # 事件代码"namespace": "BidirectionalTTS", # 命名空间,固定为 "BidirectionalTTS""req_params": { # 请求参数"text": text, # 合成文本"speaker": speaker, # 发音人"audio_params": { # 音频参数"format": audio_format, # 音频格式"sample_rate": audio_sample_rate # 音频采样率}}}# 使用 json.dumps 将字典转换为 JSON 字符串# 使用 str.encode 将 JSON 字符串编码为 UTF-8 bytesreturn str.encode(json.dumps(payload_dict))async def send_event(ws: ClientConnection, header: bytes, optional: bytes | None = None,payload: bytes | None = None):"""向 WebSocket 连接发送一个完整的 V3 协议消息。自动组装 Header, Optional (如果提供), Payload Size (如果提供), Payload。Args:ws (ClientConnection): 已建立的 websockets 客户端连接对象。header (bytes): 序列化后的 Header (4 字节)。optional (bytes | None, optional): 序列化后的 Optional 部分。如果为 None 则不发送。payload (bytes | None, optional): Payload 数据。如果为 None 则不发送。Raises:TypeError: 如果 ws 不是 ClientConnection 类型。"""# 检查 ws 类型,确保是 websockets 的 ClientConnectionif not isinstance(ws, ClientConnection):raise TypeError(f"Expected websockets.asyncio.client.ClientConnection, got {type(ws)}")# 使用 bytearray 构建完整的客户端请求消息full_client_request = bytearray(header)# 如果 Optional 部分存在,则追加到消息中if optional is not None:full_client_request.extend(optional)# 如果 Payload 部分存在if payload is not None:# 计算 Payload 的长度payload_size = len(payload)# 将长度转换为 4 字节大端序 bytes,并追加到消息中 (协议要求)full_client_request.extend(payload_size.to_bytes(4, 'big', signed=True))# 追加 Payload 的实际内容full_client_request.extend(payload)# 通过 WebSocket 连接异步发送完整的消息# print(f"Sending message: {full_client_request.hex()}") # 调试用:打印发送的消息内容await ws.send(full_client_request)def read_res_content(res: bytes, offset: int) -> tuple[str, int]:"""从响应字节流 `res` 的指定 `offset` 处读取一个 "内容块"。内容块格式为:4 字节长度 (大端) + UTF-8 编码的字符串内容。常用于读取 ConnectionId, SessionId, response_meta_json 等。Args:res (bytes): 包含响应数据的字节流。offset (int): 开始读取的偏移量。Returns:tuple[str, int]: 返回包含读取到的字符串内容和更新后的偏移量的元组。"""# 从 offset 处读取 4 字节,解析为内容长度 (大端序整数)content_size = int.from_bytes(res[offset: offset + 4], 'big') # 假设长度为非负,用 'big' 即可# 更新偏移量,跳过长度字段offset += 4# 根据解析出的长度,读取相应字节数的内容content_bytes = res[offset: offset + content_size]# 将读取到的 bytes 使用 UTF-8 解码为字符串content = str(content_bytes, encoding='utf8')# 更新偏移量,跳过内容字段offset += content_size# 返回解析到的字符串和新的偏移量return content, offsetdef read_res_payload(res: bytes, offset: int) -> tuple[bytes, int]:"""从响应字节流 `res` 的指定 `offset` 处读取 Payload 部分。Payload 格式为:4 字节长度 (大端) + Payload 二进制数据。Args:res (bytes): 包含响应数据的字节流。offset (int): 开始读取的偏移量。Returns:tuple[bytes, int]: 返回包含读取到的 Payload 二进制数据和更新后的偏移量的元组。"""# 从 offset 处读取 4 字节,解析为 Payload 长度 (大端序整数)payload_size = int.from_bytes(res[offset: offset + 4], 'big') # 假设长度为非负# 更新偏移量,跳过长度字段offset += 4# 根据解析出的长度,读取相应字节数的 Payload 数据payload = res[offset: offset + payload_size]# 更新偏移量,跳过 Payload 数据offset += payload_size# 返回读取到的 Payload bytes 和新的偏移量return payload, offsetdef parser_response(res) -> Response:"""解析从 WebSocket 收到的原始响应数据 `res`,填充并返回一个 `Response` 对象。根据 Header 中的信息(特别是消息类型和标志位)来决定如何解析 Optional 和 Payload 部分。Args:res (bytes | str): 从 WebSocket 收到的原始数据。期望是 bytes,如果是 str 则表示出错。Returns:Response: 包含解析后信息的 Response 对象。Raises:RuntimeError: 如果输入的 res 是字符串,表示可能接收到了错误消息或非预期数据。IndexError: 如果 res 的长度不足以解析预期字段。"""# 如果收到的是字符串,通常表示连接出错或服务器发送了非二进制错误信息if isinstance(res, str):raise RuntimeError(f"Received string message, expected bytes. Message: {res}")# 检查响应长度是否至少包含头部 (4 字节)if len(res) < 4:raise ValueError(f"Response too short to contain a header. Length: {len(res)}")# 初始化用于存储解析结果的 Response 对象response = Response(Header(), Optional())# --- 解析 Header (前 4 字节) ---header = response.header# 掩码,用于提取低 4 位num_mask = 0b00001111# 字节 0: 高 4 位是协议版本,低 4 位是头部大小header.protocol_version = (res[0] >> 4) & num_maskheader.header_size = res[0] & num_mask # 使用掩码提取低 4 位# 字节 1: 高 4 位是消息类型,低 4 位是特定标志header.message_type = (res[1] >> 4) & num_maskheader.message_type_specific_flags = res[1] & num_mask# 字节 2: 高 4 位是序列化方法,低 4 位是压缩方法header.serial_method = (res[2] >> 4) & num_mask # 注意:原文这里写反了,应为 >> 4header.compression_type = res[2] & num_mask # 注意:原文这里写反了,应为 & 0x0f# 字节 3: 保留字段header.reserved_data = res[3] # 注意:原文类属性名和这里变量名不一致,统一为 reserved_data# --- 解析 Optional 和 Payload ---# 初始化偏移量,指向 Header 之后的位置offset = 4 # Header 固定 4 字节optional = response.optional# 根据消息类型判断如何解析后续部分if header.message_type in [FULL_SERVER_RESPONSE, AUDIO_ONLY_RESPONSE]:# 对于服务端响应 (完整或仅音频),检查是否包含 Event 字段# MsgTypeFlagWithEvent (0b100) 标志位指示 Optional 部分存在 Eventif header.message_type_specific_flags & MsgTypeFlagWithEvent: # 使用位与操作检查标志位# 检查是否有足够字节读取 Event (4 字节)if offset + 4 > len(res):raise ValueError(f"Response too short for Event field. Offset: {offset}, Length: {len(res)}")# 读取 4 字节的 Event 代码 (大端序,有符号)optional.event = int.from_bytes(res[offset:offset + 4], "big", signed=True)# 更新偏移量offset += 4# --- 根据具体的 Event 代码解析 Optional 中的其他字段 ---# 注意:这里的解析逻辑需要严格对应协议文档中每个事件的响应格式if optional.event == EVENT_NONE:# 无事件,通常 Optional 部分结束pass # 可能后面还有 Payload,继续解析elif optional.event == EVENT_ConnectionStarted:# 连接成功事件,包含 ConnectionIdoptional.connectionId, offset = read_res_content(res, offset)elif optional.event == EVENT_ConnectionFailed:# 连接失败事件,包含错误信息的 JSONoptional.response_meta_json, offset = read_res_content(res, offset)elif optional.event in [EVENT_SessionStarted, EVENT_SessionFailed, EVENT_SessionFinished]:# 会话开始/失败/结束事件,包含 SessionId 和可能的元信息 JSONoptional.sessionId, offset = read_res_content(res, offset)optional.response_meta_json, offset = read_res_content(res, offset)elif optional.event == EVENT_TTSResponse:# TTS 响应事件 (通常是音频),包含 SessionId 和音频 Payloadoptional.sessionId, offset = read_res_content(res, offset)# 读取音频 Payloadresponse.payload, offset = read_res_payload(res, offset)elif optional.event in [EVENT_TTSSentenceStart, EVENT_TTSSentenceEnd]:# 句子开始/结束事件,包含 SessionId 和元信息 JSON Payloadoptional.sessionId, offset = read_res_content(res, offset)# 注意:这里读取的是 JSON 字符串 Payload,不是二进制 Payload# 需要一个类似 read_res_payload 但返回 str 的函数,或者调整 read_res_payload# 暂时假设 read_res_content 可以读取这种 JSON payload (如果其格式是 长度+内容)# 或者,如果 payload_json 本身就是 payload 部分的内容:payload_bytes, offset = read_res_payload(res, offset)try:response.payload_json = payload_bytes.decode('utf-8')except UnicodeDecodeError:print(f"Warning: Could not decode payload as UTF-8 for event {optional.event}")response.payload_json = repr(payload_bytes) # 存原始 repr# --- 解析可能存在的 Payload (即使没有 Event) ---# 对于 AUDIO_ONLY_RESPONSE,即使没有 Event 标志,也可能直接跟 Payload# 对于 FULL_SERVER_RESPONSE,在 Optional 之后可能还有 Payload# 检查是否还有剩余字节,并且剩余字节数大于等于 Payload 长度字段 (4 字节)if offset + 4 <= len(res):# 尝试读取 Payload (如果前面事件解析时未读取)# 注意:需要判断是否已经读取过 Payload,避免重复读取if response.payload is None and response.payload_json is None:# 简单假设剩余的就是 Payload (需要根据协议确认此逻辑是否总是正确)# 或者更严谨地,检查 Header 中的序列化方法等信息# 如果是 AUDIO_ONLY_RESPONSE,几乎总是音频 Payloadif header.message_type == AUDIO_ONLY_RESPONSE:response.payload, offset = read_res_payload(res, offset)# 对于 FULL_SERVER_RESPONSE,需要更明确的规则判断是否有 Payload# (例如,检查序列化方法是否为 NO_SERIALIZATION 或 JSON)# 此处简化处理,如果还有数据且未被解析,尝试按 Payload 读取elif header.message_type == FULL_SERVER_RESPONSE:# 检查剩余长度是否足够表示一个最小的payload(长度字段)if len(res) > offset + 4:# 尝试读取,如果失败(如长度超出范围)会抛出异常try:# 这里需要区分是二进制 payload 还是 json payloadif header.serial_method == JSON:payload_bytes, temp_offset = read_res_payload(res, offset)try:response.payload_json = payload_bytes.decode('utf-8')offset = temp_offsetexcept UnicodeDecodeError:print(f"Warning: Could not decode JSON payload as UTF-8 for event {optional.event}")# 不更新 offset,保留原始字节供调试elif header.serial_method == NO_SERIALIZATION:response.payload, offset = read_res_payload(res, offset)# else: 其他序列化方法暂不处理except Exception as e:print(f"Warning: Error trying to parse trailing payload: {e}")elif header.message_type == ERROR_INFORMATION:# 错误信息类型# 检查是否有足够字节读取 ErrorCode (4 字节)if offset + 4 > len(res):raise ValueError(f"Response too short for ErrorCode field. Offset: {offset}, Length: {len(res)}")# 读取 4 字节的错误码 (大端序,有符号)optional.errorCode = int.from_bytes(res[offset:offset + 4], "big", signed=True)offset += 4# 错误信息通常还包含一个描述性的 Payloadresponse.payload, offset = read_res_payload(res, offset)# 可以尝试将 payload 解码为字符串以获取错误描述try:response.payload_json = response.payload.decode('utf-8') # 假设错误描述是 UTF-8except (UnicodeDecodeError, TypeError):pass # 解码失败或 payload 为 None# 返回填充好的 Response 对象return responsedef print_response(res: Response, tag: str):"""打印 Response 对象的详细信息,用于调试。Args:res (Response): 要打印的 Response 对象。tag (str): 用于标识打印来源的前缀字符串。"""# 打印 Header 字典print(f'===>{tag} header:{res.header.__dict__}')# 打印 Optional 字典print(f'===>{tag} optional:{res.optional.__dict__}')# 打印二进制 Payload 的长度payload_len = 0 if res.payload is None else len(res.payload)print(f'===>{tag} payload len:{payload_len}')# 打印 JSON Payload 字符串print(f'===>{tag} payload_json:{res.payload_json}')# --- 核心 API 调用函数 ---async def start_connection(ws: ClientConnection):"""发送 "Start Connection" 事件 (EVENT_Start_Connection) 给服务器。Args:ws (ClientConnection): WebSocket 连接对象。"""# 构建 Header: 消息类型为客户端请求,带 Event 标志header = Header(message_type=FULL_CLIENT_REQUEST,message_type_specific_flags=MsgTypeFlagWithEvent).as_bytes()# 构建 Optional: 包含 Start Connection 事件代码optional = Optional(event=EVENT_Start_Connection).as_bytes()# 构建 Payload: 对于 Start Connection,通常为空 JSON 对象 "{}"payload = str.encode("{}")# 发送事件await send_event(ws, header, optional, payload)print("===> Sent Start Connection event")async def start_session(ws: ClientConnection, speaker: str, session_id: str):"""发送 "Start Session" 事件 (EVENT_StartSession) 给服务器。Args:ws (ClientConnection): WebSocket 连接对象。speaker (str): 请求使用的发音人标识。session_id (str): 本次会话的唯一 ID。"""# 构建 Header: 客户端请求,带 Event 标志,Payload 为 JSONheader = Header(message_type=FULL_CLIENT_REQUEST,message_type_specific_flags=MsgTypeFlagWithEvent,serial_method=JSON).as_bytes()# 构建 Optional: 包含 Start Session 事件代码和 Session IDoptional = Optional(event=EVENT_StartSession, sessionId=session_id).as_bytes()# 构建 Payload: 包含事件代码和发音人信息payload = get_payload_bytes(event=EVENT_StartSession, speaker=speaker)# 发送事件await send_event(ws, header, optional, payload)print(f"===> Sent Start Session event (Session ID: {session_id})")async def send_text(ws: ClientConnection, speaker: str, text: str, session_id: str):"""发送 "Task Request" 事件 (EVENT_TaskRequest) 给服务器,携带要合成的文本。Args:ws (ClientConnection): WebSocket 连接对象。speaker (str): 发音人标识。text (str): 需要合成的文本。session_id (str): 当前会话的 ID。"""# 构建 Header: 客户端请求,带 Event 标志,Payload 为 JSONheader = Header(message_type=FULL_CLIENT_REQUEST,message_type_specific_flags=MsgTypeFlagWithEvent,serial_method=JSON).as_bytes()# 构建 Optional: 包含 Task Request 事件代码和 Session IDoptional = Optional(event=EVENT_TaskRequest, sessionId=session_id).as_bytes()# 构建 Payload: 包含事件代码、文本和发音人信息payload = get_payload_bytes(event=EVENT_TaskRequest, text=text, speaker=speaker)# 发送事件await send_event(ws, header, optional, payload)print(f"===> Sent Task Request event (Text: '{text[:20]}...')")async def finish_session(ws: ClientConnection, session_id: str):"""发送 "Finish Session" 事件 (EVENT_FinishSession) 给服务器。Args:ws (ClientConnection): WebSocket 连接对象。session_id (str): 要结束的会话的 ID。"""# 构建 Header: 客户端请求,带 Event 标志,Payload 为 JSON (虽然内容为空)header = Header(message_type=FULL_CLIENT_REQUEST,message_type_specific_flags=MsgTypeFlagWithEvent,serial_method=JSON).as_bytes()# 构建 Optional: 包含 Finish Session 事件代码和 Session IDoptional = Optional(event=EVENT_FinishSession, sessionId=session_id).as_bytes()# 构建 Payload: 通常为空 JSON 对象 "{}"payload = str.encode('{}')# 发送事件await send_event(ws, header, optional, payload)print(f"===> Sent Finish Session event (Session ID: {session_id})")async def finish_connection(ws: ClientConnection):"""发送 "Finish Connection" 事件 (EVENT_FinishConnection) 给服务器。Args:ws (ClientConnection): WebSocket 连接对象。"""# 构建 Header: 客户端请求,带 Event 标志,Payload 为 JSON (虽然内容为空)header = Header(message_type=FULL_CLIENT_REQUEST,message_type_specific_flags=MsgTypeFlagWithEvent,serial_method=JSON).as_bytes()# 构建 Optional: 包含 Finish Connection 事件代码optional = Optional(event=EVENT_FinishConnection).as_bytes()# 构建 Payload: 通常为空 JSON 对象 "{}"payload = str.encode('{}')# 发送事件await send_event(ws, header, optional, payload)print("===> Sent Finish Connection event")# --- 主业务逻辑 ---async def run_demo(appId: str, token: str, speaker: str, text: str, output_path: str):"""运行 TTS WebSocket V3 示例的主函数。完成连接、认证、发送文本、接收音频、关闭连接的完整流程。Args:appId (str): 应用 ID (从火山引擎获取)。token (str): 访问令牌 (从火山引擎获取)。speaker (str): 发音人标识。text (str): 要合成的文本。output_path (str): 保存输出音频文件的路径。"""# WebSocket 连接地址url = 'wss://openspeech.bytedance.com/api/v3/tts/bidirection'# 构建 WebSocket 连接请求头,用于身份认证和资源指定ws_header = {"X-Api-App-Key": appId, # 应用 ID"X-Api-Access-Key": token, # 访问令牌"X-Api-Resource-Id": 'volc.service_type.10029', # 请求的资源 ID (TTS 服务)"X-Api-Connect-Id": str(uuid.uuid4()), # 本次连接的唯一 ID}print(f"Connecting to {url}...")# 使用 websockets.connect 建立异步 WebSocket 连接# additional_headers: 传递认证等头部信息# max_size: 设置接收消息的最大尺寸 (设置为较大值以处理长音频)async with websockets.connect(url, additional_headers=ws_header, max_size=1000000000) as ws:print("WebSocket connection established.")# 1. 发送 Start Connection 请求await start_connection(ws)# 接收服务器响应res_bytes = await ws.recv()# 解析响应res = parser_response(res_bytes)# 打印响应信息print_response(res, 'start_connection response:')# 检查响应事件是否为连接成功if res.optional.event != EVENT_ConnectionStarted:raise RuntimeError(f"Start connection failed. Response: {res.optional.response_meta_json or res.payload_json or 'Unknown error'}")print("Connection started successfully.")# 2. 发送 Start Session 请求# 生成一个唯一的 Session IDsession_id = uuid.uuid4().__str__().replace('-', '')await start_session(ws, speaker, session_id)# 接收响应并解析res = parser_response(await ws.recv())print_response(res, 'start_session response:')# 检查响应事件是否为会话开始成功if res.optional.event != EVENT_SessionStarted:raise RuntimeError(f"Start session failed! Response: {res.optional.response_meta_json or res.payload_json or 'Unknown error'}")print(f"Session started successfully (Session ID: {session_id}).")# 3. 发送文本 (Task Request)await send_text(ws, speaker, text, session_id)# 4. 发送 Finish Session 请求 (表明文本已发送完毕)await finish_session(ws, session_id)# 5. 循环接收音频数据并写入文件print(f"Receiving audio data and writing to {output_path}...")# 使用 aiofiles 异步打开文件进行二进制写入async with aiofiles.open(output_path, mode="wb") as output_file:while True:# 接收服务器响应res_bytes = await ws.recv()# 解析响应res = parser_response(res_bytes)# 打印响应信息 (调试)# print_response(res, 'audio_receive loop:')# 判断响应类型if res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE:# 如果是 TTS 音频响应,并且消息类型是仅音频if res.payload:# 将接收到的音频数据块写入文件await output_file.write(res.payload)# print(f"Received and wrote {len(res.payload)} bytes of audio.") # 调试信息else:print("Warning: Received EVENT_TTSResponse with empty payload.")elif res.optional.event in [EVENT_TTSSentenceStart, EVENT_TTSSentenceEnd]:# 如果是句子开始/结束事件,打印信息并继续接收print(f"Received event: {'Sentence Start' if res.optional.event == EVENT_TTSSentenceStart else 'Sentence End'}. Info: {res.payload_json}")continueelif res.optional.event == EVENT_SessionFinished:# 如果收到会话结束事件,表示音频流结束,退出循环print("Received Session Finished event. Audio stream ended.")print_response(res, 'session_finished response:')breakelif res.optional.event == EVENT_SessionFailed:# 如果收到会话失败事件,抛出异常print_response(res, 'session_failed response:')raise RuntimeError(f"Session failed during audio receive. Info: {res.optional.response_meta_json or res.payload_json}")elif header.message_type == ERROR_INFORMATION:# 如果收到错误信息print_response(res, 'error_information response:')error_desc = res.payload.decode('utf-8') if res.payload else 'Unknown error'raise RuntimeError(f"Received error information. Code: {res.optional.errorCode}, Desc: {error_desc}")else:# 收到其他未预期事件,打印并可能退出或忽略print(f"Warning: Received unexpected event or message type during audio receive.")print_response(res, 'unexpected_response:')# 根据需要决定是否退出 break# break # 暂时选择退出print(f"Audio saved to {output_path}")# 6. 发送 Finish Connection 请求await finish_connection(ws)# 接收最后的响应res = parser_response(await ws.recv())print_response(res, 'finish_connection response:')# 理论上应该收到 EVENT_ConnectionFinished 或类似确认print('===> Demo finished successfully.')# --- 主程序入口 ---
if __name__ == "__main__":# --- 用户配置 ---# 请替换为您的火山引擎应用 ID 和访问令牌# 重要提示:请勿将您的 AppID 和 Token 硬编码在代码中并公开分享。# 建议使用环境变量、配置文件或其他安全方式管理凭证。appId = "" # 替换为您自己的 AppIdtoken = "" # 替换为您自己的 Token# 检查 AppID 和 Token 是否已配置if not appId or not token:print("错误:请在代码中设置您的 appId 和 token。")print("请访问 https://console.volcengine.com/iam/keymanage/ 获取。")exit(1)# 要合成的文本text = '火山引擎,让智能增长。欢迎使用火山引擎实时语音合成服务。'text = ('我是松下机器人SuperMAX ジョジョの力は無限大だ!' )# 使用的发音人标识# 可选的发音人请参考文档: https://www.volcengine.com/docs/6561/1257544speaker = 'zh_female_shuangkuaisisi_moon_bigtts' # 示例发音人# 输出音频文件的路径 (请确保目录存在或有写入权限)output_path = './tts_output.mp3' # 将保存在当前目录下print("Starting TTS demo...")print(f" App ID: {appId[:4]}...") # 仅显示部分 AppIDprint(f" Speaker: {speaker}")print(f" Text: {text}")print(f" Output Path: {output_path}")try:# 使用 asyncio.run() 运行异步的 run_demo 函数asyncio.run(run_demo(appId, token, speaker, text, output_path))except RuntimeError as e:# 捕获并打印运行时错误print(f"\nAn error occurred: {e}")except websockets.exceptions.ConnectionClosedError as e:print(f"\nWebSocket connection closed unexpectedly: {e}")except Exception as e:# 捕获并打印其他未预料的异常import tracebackprint(f"\nAn unexpected error occurred: {e}")traceback.print_exc() # 打印详细的堆栈跟踪信息
相关文章:
火山引擎实时语音合成WebSocket V3协议Python实现demo
火山引擎语音整体特点 火山引擎(字节跳动旗下)的语音合成产品确实非常面向多媒体内容创作,特别是短视频、有声书和多人场景。 1. 音色多样性与场景细分 火山引擎提供了极其丰富的音色选择(100音色),并按以下场景精细分类: 多情感音色&…...
Apache SeaTunnel:新一代开源、高性能数据集成工具
Apache SeaTunnel 是一款开源、分布式、高性能的数据集成工具,可以通过配置快速搭建数据管道,支持实时海量数据同步。 Apache SeaTunnel 专注于数据集成和数据同步,主要旨在解决数据集成领域的常见问题: 数据源多样性:…...
【音视频】音频解码实战
音频解码过程 ⾳频解码过程如下图所示: FFmpeg流程 关键函数 关键函数说明: avcodec_find_decoder:根据指定的AVCodecID查找注册的解码器。av_parser_init:初始化AVCodecParserContext。avcodec_alloc_context3:为…...
学习思路分享---从0开始搭建基本web服务器
学习思路分享—从0开始搭建基本web服务器 为什么要搭建yum仓库? 下载系统软件,类似于应用商店,系统软件,podman,镜像,容器,镜像仓库,docker,集装箱, 作用:自动解决依赖关系 为什么要搭建web服务器? 提供网站 , nginx 第一步搭建yum仓库,本地离线仓库 挂载关盘驱动…...
开源模型应用落地-Podcastfy-从文本到声音的智能跃迁-Docker(二)
一、前言 在当今信息呈现方式越来越多样化的背景下,如何将文字、图片甚至视频高效转化为可听的音频体验,已经成为内容创作者、教育者和研究者们共同关注的重要话题。Podcastfy是一款基于Python的开源工具,它专注于将多种形式的内容智能转换成…...
docker 里面没有 wget 也 install 不了
docker 里面没有 wget 也 install 不了 如果你在Docker容器中发现没有安装wget工具,并且无法通过常规方法安装它,这通常是因为容器的基础镜像中缺少包管理工具,或者包源配置不正确。以下是一些可能的解决方案: 使用包管理工具安…...
【无人机】问题分析。查看电机转速时,四个电机转速不一致,QGC中检测到电机转速不均衡
1、问题描述 在组装完成无人机后,对无人机的电机进行测试 在MAVLink Inspector一栏中,点击SERVO_OUTPUT_RAW可以查看飞控输出的PWM信号。 无人机解锁(Armed)后,按照油门大小servo1/2/3/4_raw的值域为1000-2000&…...
Docker Compose常用命令
Docker Compose常用命令 安装docker-comosedocker-compose配置文件及常用指令yaml 文件级docker-compose.yml配置文件示例 docker compose常用命令启动服务停止服务重启服务查看运行容器列表查看服务日志构建镜像docker-compose rm删除 安装docker-comose # 声明版本 VER2.35.…...
生产环境大数据平台权限管理
引言:数据资产保护的生死线 在金融行业某头部企业发生的数据泄露事件中,由于权限管理漏洞导致千万级用户信息外泄,直接经济损失超过2.3亿元。这个案例揭示了生产环境大数据平台权限管理的重要性和复杂性。本文将深入探讨从权限模型设计到实施…...
【连载6】基础智能体的进展与挑战综述-奖励
基础智能体的进展与挑战综述 从类脑智能到具备可进化性、协作性和安全性的系统 【翻译团队】刘军(liujunbupt.edu.cn) 钱雨欣玥 冯梓哲 李正博 李冠谕 朱宇晗 张霄天 孙大壮 黄若溪 5. 奖励 奖励机制帮助智能体区分有益与有害的行为,塑造其学习过程并影响其决策…...
什么是CRM系统,它的作用是什么?CRM全面指南
CRM(Customer Relationship Management,客户关系管理)系统是一种专门用于集中管理客户信息、优化销售流程、提升客户满意度、支持精准营销、驱动数据分析决策、加强跨部门协同、提升客户生命周期价值的业务系统工具。其中,优化销售…...
【AI提示词】投资策略专家
提示说明 投资策略专家致力于帮助用户构建和优化投资组合,实现资产增值。专家通过深入分析市场趋势、经济数据和公司基本面,为用户制定符合其投资目标和风险偏好的策略。 提示词 # 角色 投资策略专家## 注意 1. 投资策略专家需要具备深入分析市场的能…...
川翔云电脑32G大显存集群机器上线!
川翔云电脑今日重磅推出32G 大显存机型,为游戏玩家、设计师、AI 开发者等提供极致云端算力体验! 一、两大核心配置,突破性能天花板 ✅ 32G 超大显存机型 行业领先:搭载 NVIDIA 专业显卡,单卡可分配 32G 独立显存&am…...
最新扣子(Coze)案例教程:飞书多维表格按条件筛选记录 + 读取分页Coze工作流,无限循环使用方法,手把手教学,完全免费教程
大家好,我是斜杠君。 👨💻 星球群里有同学想学习一下飞书多维表格的使用方法,关于如何通过按条件筛选飞书多维表格中的记录,以及如何使用分页解决最多一次只能读取500条的限制问题。 斜杠君今天就带大家一起搭建一…...
ProxySQL 的性能优化需结合实时监控数据与动态配置调整
ProxySQL 的性能优化需结合实时监控数据与动态配置调整,具体操作如下: 一、性能监控实现 内置监控模块配置 启用监控用户:在 global_variables 表中设置监控账号,用于检测节点健康状态: sql UPDATE global_variables SET variable_value=‘monitor’ WHERE va…...
前端框架的“快闪“时代:我们该如何应对技术迭代的洪流?
引言:前端开发者的"框架疲劳" “上周刚学完Vue 3的组合式API,这周SolidJS又火了?”——这恐怕是许多前端开发者2023年的真实心声。前端框架的迭代速度已经达到了令人目眩的程度,GitHub每日都有新框架诞生,n…...
准确--Tomcat更换证书
具体意思是: Starting Coyote HTTP/1.1 on http-8080: HTTP 连接器(端口 8080)启动成功了。严重: Failed to load keystore type PKCS12 with path conf/jlksearch.fzsmk.cn.pfx due to failed to decrypt safe contents entry: javax.crypt…...
数据库对象与权限管理-Oracle数据字典详解
1. 数据字典概念讲解 Oracle数据字典是数据库的核心组件,它存储了关于数据库结构、用户信息、权限设置和系统性能等重要的元数据信息。这些信息对于数据库的日常管理和维护至关重要。数据字典在数据库创建时自动生成,并随着数据库的运行不断更新。 数据…...
黑阈免激活版:智能管理后台,优化手机性能
在使用安卓手机的过程中,许多用户会遇到手机卡顿、电池续航不足等问题。这些问题通常是由于后台运行的应用程序过多,占用大量系统资源导致的。今天,我们要介绍的 黑阈免激活版,就是这样一款由南京简域网络科技工作室开发的手机辅助…...
C# foreach 循环中获取索引的完整方案
一、手动维护索引变量 实现方式: 在循环外部声明索引变量,每次迭代手动递增: int index 0; foreach (var item in collection) { Console.WriteLine($"{index}: {item}"); index; } 特点: 简单直接&#…...
刷题之路:C++ 解题分享与技术总结
目录 引言 ###刷题实战 (一)移除数组中的重复元素 (二)数组中出现次数超过一半的数字 (三)杨辉三角 (四)只出现一次的数字 (五)找出数组中两个只出现一…...
深入探讨:如何完美完成标签分类任务(数据治理中分类分级的分类思考)
文章目录 一、标签分类的核心价值与挑战1.1 标签分类的战略意义1.2 标签分类面临的主要挑战 二、标签分类方法论的系统设计2.1 多层级标签架构设计2.2 精准的标签匹配技术2.3 混合优化策略 三、标签分类的技术实现3.1 高维向量空间中的标签表示3.2 图数据库驱动的标签关系处理3…...
【解决 el-table 树形数据更新后视图不刷新的问题】
内容包含deepseek自动生成内容。第一种亲测可行。 本文章仅用于问题记录 解决 el-table 树形数据更新后视图不刷新的问题 在 Element Plus 的 el-table 中使用树形数据时,当数据更新后视图不自动刷新是一个常见问题。以下是几种解决方案: 问题原因 e…...
MuJoCo中的机器人状态获取
UR5e机器人xml文件模型 <mujoco model"ur5e"><compiler angle"radian" meshdir"assets" autolimits"true"/><option integrator"implicitfast"/><default><default class"ur5e">&…...
第五篇:linux之vim编辑器、用户相关
第五篇:linux之vim编辑器、用户相关 文章目录 第五篇:linux之vim编辑器、用户相关一、vim编辑器1、什么是vim?2、为什么要使用vim?3、vi和vim有什么区别?4、vim编辑器三种模式 二、用户相关1、什么是用户?2…...
taobao.trades.sold.get(淘宝店铺订单接口)
淘宝店铺提供了多种订单接口,可以用来获取订单信息、创建订单、修改订单等操作。 获取订单列表接口:可以使用该接口获取店铺的订单列表,包括订单号、买家信息、订单状态等。 获取单个订单信息接口:可以使用该接口获取指定订单的详…...
媒体发稿攻略,解锁新闻发稿成长新高度
新闻媒体发稿全攻略! 如何快速上稿主流权威央级媒体? 大家好!今天来聊聊媒体发稿的那些事儿,希望能帮到正在发稿或者准备发稿的小伙伴们。 ①明确目标媒体 首先,得搞清楚你要把稿子发给哪些媒体和。这一步非常关键,因为选择适合的媒体是发…...
WebRTC服务器Coturn服务器部署
1、概述 作为WebRTC服务器,只需要部署开源的coturn即可,coturn同时实现了STUN和TURN的协议 2、Coturn具体部署 2.1 Coturn简介 coturn是一个开源的STUN/TURN服务器,把STUN服务器跟TURN服务器都整合为一个服务器,主要提供一下几个功…...
lspci的资料
PCI即Peripheral Component Interconnect。 在 Linux 上使用 lspci 命令查看硬件情况 | Linux 中国 lspci 命令用于显示连接到 PCI 总线的所有设备,从而满足上述需求。该命令由 pciutils 包提供,可用于各种基于 Linux 和 BSD 的操作系统。 使用 lspci 和…...
GitLab 提交权限校验脚本
.git/hooks 目录详解与配置指南 一、什么是 .git/hooks? .git/hooks 是 Git 仓库中一个隐藏目录,用于存放 钩子脚本(Hook Scripts)。这些脚本会在 Git 执行特定操作(如提交、推送、合并)的前/后自动触发&…...
WebRTC服务器Coturn服务器相关测试工具
1、概述 在安装开源的webrtc服务器coturn服务器后,会附带安装coturn的相关工具,主要有以下几种工具 2、turnadmin工具 说明:服务器命令行工具,提供添加用户、添加管理员、生成TURN密钥等功能,turnadmin -h查看详细用…...
基于Python+Pytest实现自动化测试(全栈实战指南)
目录 第一篇:基础篇 第1章 自动化测试概述 1.1 什么是自动化测试 第2章 环境搭建与工具链配置 2.1 Python环境安装(Windows/macOS/Linux) 2.2 虚拟环境管理 2.3 Pytest基础配置(pytest.ini) 第3章 Pytest核心语…...
符号速率估计——小波变换法
[TOC]符号速率估计——小波变换法 一、原理 1.Haar小波变换 小波变换在信号处理领域被成为数学显微镜,不同于傅里叶变换,小波变换可以观测信号随时间变换的频谱特征,因此,常用于时频分析。 当小波变换前后位置处于同一个码元…...
SQLMesh隔离系统深度实践指南:动态模式映射与跨环境计算复用
在数据安全与开发效率的双重压力下,SQLMesh通过动态模式映射、跨环境计算复用和元数据隔离机制三大核心技术,完美解决了生产与非生产环境的数据壁垒问题。本文提供从环境配置到生产部署的完整实施框架,助您构建安全、高效、可扩展的数据工程体…...
调整IntelliJ IDEA中当前文件所在目录的显示位置
文章目录 1. 问题呈现2. 调整方法3. 更改后的界面 更多 IntelliJ IDEA 的使用技巧可查看 IntelliJ IDEA 专栏中的文章: IntelliJ IDEA 1. 问题呈现 在 IntelliJ IDEA 中,我们在浏览某个文件时,文件所在的目录会显示在下方的状态栏中&#x…...
关于ubuntu密码正确但是无法登录的情况
参考这个文章: https://blog.csdn.net/cuichongxin/article/details/117462494 检查一下是不是用户被lock了 输入passwd -s username 如果用户是L状态,那么就是lock了。 使用 passwd -u username 解锁 关于 .bashrc 不生效 有几点: ~/.…...
OpenCV中的透视变换方法详解
文章目录 引言1. 什么是透视变换2. 透视变换的数学原理3. OpenCV中的透视变换代码实现3.1 首先定义四个函数 3.1.1 cv_show() 函数 3.1.2 def resize() 函数 3.1.3 order_points() 函数 3.1.4 four_point_transform() 函数 3.2 读取图片并做预处理3.3 轮廓检测3.4 获取最大…...
基于DeepSeek的网络爬虫技术创新与实践应用
摘要 在人工智能迅猛发展的时代背景下,网络爬虫技术正经历着深刻变革。本文聚焦于融合DeepSeek大模型的开源爬虫框架Crawl4AI,深入探讨其在网络数据抓取与分析领域的技术原理、功能特性、应用实践及未来发展趋势。通过对Crawl4AI的异步操作、动态内容处理…...
Python基于语音识别的智能垃圾分类系统【附源码、文档说明】
博主介绍:✌Java老徐、7年大厂程序员经历。全网粉丝12w、csdn博客专家、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ 🍅文末获取源码联系🍅 👇🏻 精彩专栏推荐订阅👇&…...
关于RPC
1.什么是RPC RPC(Remote Procedure Call),即远程过程调用(协议)。它允许像调用本地服务一样调用远程服务,用于实现分布式系统中跨网络进行通信 的技术,是一种计算机通信协议。 RPC是一种服务器…...
前端频繁调用后端接口问题思考
针对前端频繁调用后端接口的问题,以下是一套分步解决方案,结合Java后端技术栈: 1. 实时数据场景:WebSocket/SSE 适用场景:实时股票行情、即时聊天、监控仪表盘 // Spring WebSocket 配置示例 Configuration EnableW…...
Mujoco robosuite 机器人模型
import ctypes import os# 获取当前脚本所在的目录 script_dir os.path.dirname(os.path.abspath(__file__))# 构建库文件的相对路径 lib_relative_path os.path.join(dynamic_models, UR5e, Jb.so)# 拼接成完整的路径 lib_path os.path.join(script_dir, lib_relative_path…...
进阶篇 第 7 篇 (终章):融会贯通 - 多变量、模型选择与未来之路
进阶篇 第 7 篇 (终章):融会贯通 - 多变量、模型选择与未来之路 (图片来源: Pixabay on Pexels) 我们已经一起走过了时间序列分析的进阶之旅!从深入经典统计模型 ETS、ARIMA、SARIMA,到探索现代利器 Prophet,再到拥抱机器学习和初…...
网络安全·第五天·TCP协议安全分析
一、传输层协议概述 1、功能 传输层负责建立端到端的连接,即应用进程之间的通信,负责数据在端到端之间的传输。与网络层不同的是,网络层负责主机与主机之间的通信。 同时,传输层还要对收到的报文进行差错检测(首部和…...
LX10-MDK的使用技巧
MDK5的使用技巧 查找匹配花括号 Ctrle table键的妙用 一次右缩进4个(个人偏好设置)空格shiftenter取消,即左缩进 快速注释/取消注释 先选代码→ 快速编辑一列 按住ALT键选择一列编辑(实用性极强) 窗口拆分 倒数第一个:按列拆分倒数第二个:按行拆分 查找与替换(一个超级…...
IDEA创建Gradle项目然后删除报错解决方法
根据错误信息,你的项目目录中缺少Gradle构建必需的核心文件(如settings.gradle/build.gradle),且IDEA可能残留了Gradle的配置。以下是具体解决方案: 一、问题根源分析 残留Gradle配置 你通过IDEA先创建了Gradle子模块…...
JavaScript性能优化实战(2):DOM操作优化策略
浏览器渲染原理与重排重绘机制 浏览器将HTML和CSS转换为用户可见页面的过程是前端开发的基础知识,也是理解DOM性能优化的关键。这个渲染过程大致可分为以下几个步骤: 渲染过程的核心步骤 解析HTML构建DOM树:浏览器解析HTML标记,转换为DOM树(Document Object Model),表…...
乐视系列玩机---乐视1s x500 x501 x502等系列线刷救砖以及刷写第三方twrp 卡刷第三方固件步骤解析
乐视乐1S(X500 x501 x502 等)采用联发科 Helio X10(MT6795T)Turbo 64位8核处理器 通过博文了解💝💝💝 1💝💝💝-----详细解析乐视1s x500 x501x502等系列黑砖线刷救砖的步骤 2💝💝💝----官方两种更新卡刷步骤以及刷写第三方twrp过程与资源 3💝💝…...
Spark-Streaming(1)
Spark Streaming概述: 用于流式计算,处理实时数据流。 数据流以DStream(Discretized Stream)形式表示,内部由一系列RDD组成。 Spark Streaming特点: 易用、容错、易整合到spark体系。 易用性:…...
【Git】Git Revert 命令详解
Git Revert 命令详解 1. Git Revert 的基本概念 Git Revert 是一个用于撤销特定提交的命令。与 Git Reset 不同,Git Revert 不会更改提交历史,而是会创建一个新的提交来撤销指定提交的更改。这意味着,使用 Git Revert 后,项目的…...