Enhance code quality with linting and type hints

[PabloLION]

This update introduces a comprehensive linting script, improves code style, and adds type hints across multiple files. It also updates dependencies and addresses various style issues to ensure better maintainability and readability of the codebase.

[H1DDENADM1N]

流式输出用作预览，语音结束后再用非流式完整转录一遍

我一直以为是大模型根据语义修正的呢。那下图这种蓝色下划线的快速纠正的数据源是怎么来的？语音识别的话那同一段录音识别到的token不应该是确定的同样的字吗？

[PabloLION]

@HaujetZhao

iPhone 默认输入法实时语音输入还能更正前文，用过就忘不掉，馋死我了，好像电脑端也用上。😭

我感觉iPhone的不是很好用啊，macOS自带的也很难用，我感觉这个项目比苹果自带的好太多了。苹果的必须吐字非常清晰才能识别出来。

---

我目前水平有点菜，习惯有些不一样，例如我更喜欢用单引号（因为我懒得按shift），代码可读性都是以中文注释实现的（用CapsWriter写中文注释很方便），FFmpeg 的命令列表分行是按照我理解的意义组织的，有些地方用了 type hint 是为了帮助 vscode 解析，各种地方全都用上 type hint 反而让我觉得可读性更差了。

稍微说下吧。

• 单引号的都是代码风格的问题，项目内一样就无所谓，不过双引号确实更常用（由于 shell 中的引号规则，我反而倾向单引号）。这个看个人便好。
• 可读性不一定需要注释，这个项目中有些注释没什么大用。我习惯用注释说明代码意图，类似于总结（如在一段代码前写明用的算法，或者这段代码的目的）。因为具体每一句的作用可以直接看文档。我个人倾向多写 docstring，少写注释。不过注释总归是要写”代码中不易看懂的“部分，这个”不易看懂“因人而异”。后面具体给你举个例子。
• FFmpeg 的命令列表分行：可能我水平不够，反正没看出来这有什么问题。大概是-f的后面我没看懂。反正能用就行，当常量处理的话可以把它包在一个函数中。另，说到文件处理，最好搞一个 interface（python 中的 protocol），统一有无ffmpeg安装的两种情况。另外 ffmpeg 的检测有点问题。我一开始跑不通这个代码，查找了很久才发现是 ffmpeg 的问题，具体我还没看。
• type hint：我认为是个人风格吧。我比较喜欢静态类型语言，看习惯就没啥了。其实想了想，VSCode比较笨，加上之后还是方便点。

例子

此处创建的这两个变量的目的不是很清晰，注释虽然写了但是我也没看懂。我最后把它改成了一个`class AudioFileManger`，然后只有在`save_audio`的时候才会创建一个实例。这样可以把三个文件和在一起。

        # 保存音频文件
        file_path: str | Path = ""
        file: Popen[bytes] | Wave_write | None = None

---

我阅读的障碍主要是变量有点混乱。例如client_cosmic.py的queue_out好像从来没用过，file_cosmic.py也是。就会感觉逻辑上很复杂。逻辑好的代码读起来应该比较简单，因为最后大体上的范式都差不多。

---

我自己在使用的过程中，经常是按下按键开始说话了，然后发现光标还没在输入框里面。这个时候我有足够的时间调整光标，然后说完松手一下子输上去。如果说是边说话边实时输出，反而还失去这优势呢。

这个好像是我想加的取消功能啊

---

@H1DDENADM1N 我前两天看了你的fork，感觉做的很棒

我一直以为是大模型根据语义修正的呢。那下图这种蓝色下划线的快速纠正的数据源是怎么来的？语音识别的话那同一段录音识别到的token不应该是确定的同样的字吗？

他没有依赖LLM，具体实现我也不知道。
我想做成类似的效果，加一个类似打字时候的选词框的东西。然后还能从用户反馈来学习热词。现在苹果没有这个功能，我也没见过哪家有。

[H1DDENADM1N]

@PabloLION

iPhone的语音输入非常好用，特别是对比windows自带的Win+H快捷键启动语音输入来说，在准确度，速度，交互体验上全面领先。我平时聊天或问ai用capswriter，但大段文字是用iPhone配合实时同步剪贴板 etherpad-lite 非常折中的实现windows流式语音输入。之前用过vscode的流式语音输入插件，但不如iPhone好用，大概是我电脑这个老麦克风问题。你觉得iPhone的不好用可能你口音比较重吧，我这里用起来iPhone的语音输入非常好使。

[PabloLION]

你觉得iPhone的不好用可能你口音比较重吧，我这里用起来iPhone的语音输入非常好使。

我感觉他没微信转语音识别的准确，很多生活化的表达他都不知道，而且经常识别不上（我说了一大堆，没有出字）。

我刚换的16PM，用的iOS 18.1.1。用了十几年iPhone在这方面还是有点羡慕安卓用户的。

而且iOS不让第三方做语音输入。第三方键盘我试过：谷歌家的会跳转到另一个全屏幕app。微软家的键盘干脆不支持中文。我目前没什么好办法。

vscode的copilot里我一直说的英语，尴尬的是我一停下来思考他就会自动发送。有那么点烦人。

你如果有什么好用的工具请给我推荐一下！谢谢！

[H1DDENADM1N]

@PabloLION

试试这个， VS Code Speech 支持流式语音输入，而且也是离线的，只是可惜只能用在 vs code

微信的交互必须按着按钮，而且只能用在微信，不方便编辑，我几乎不用，毕竟都开微信了，还不如直接发语音或视频

[PabloLION]

@H1DDENADM1N 你说的这个我知道，他前身应该是 https://githubnext.com/projects/copilot-voice/ 我很早用过。现在也在用，咱俩说的应该一样。
另外，微软家的翻译软件应该是流式语音识别里做的最好的。识别的很清晰，断句也很合理。但是不开源，我挺想知道他们是怎么做的。

[H1DDENADM1N]

@PabloLION

https://www.bilibili.com/video/av616320040?unique_k=114514

这个听风转录可以实时语音识别并翻译，用的sensevoice加qwen，但他主要是实时字幕，也不方便编辑

[HaujetZhao]

回答下问题。

Queue_in 和 Queue_out 方面：

1. 主要是用于 server 端的，因为识别任务是计算密集，需要把模型放到一个另外的进程，主进程与模型进程之间就需要两个队列进行通讯，主进程把音频给模型用 Queue_in ，模型把结果给主进程用 Queue_out。用的是多进程 Queue
2. Client 里的 Queue 是用于传递任务信息和音频数据，用的是协程 Queue ，因为是单向投喂音频数据，没有来回沟通的需求，就没用到 queue_out，这样的变量名完全是为了形式上的统一，不想再给客户端的 queue 单独再起一个名了。
3. file_cosmic.py 是忘记删的一个遗漏哈

关于你的举例：

1. send_audio() 里面，设计的是，收到音频数据，要流式地发送、保存，比如我当录音机用（假设正在和产品经理对话，要保留证据），我不希望录了1分钟，电脑一断电，前面的数据白毁了，所以一定要流式保存，收一秒，存一秒。保存和发送部分，是用一个 while 从 queue 里面读数据的，读完后就要写，写前要判断文件有没有被创建。如果不提前新建 file_path 变量，我就得这样判断：

# 开始取数据
while task := await Cosmic.queue_in.get():
    Cosmic.queue_in.task_done()
    
    ...

    # 创建音频文件
    if Config.save_audio and 'file_path' not in locals():    # 如果之后我想重命名 file_path 变量，还得回到这里修改字符串
        file_path, file = create_file(task['data'].shape[1], time_start)
        Cosmic.audio_files[task_id] = file_path

所以我干脆在 while 之前新建这个变量，把它赋值为空，在 while 里面用 if not file_path 就可以得知需要初始化，后面重命名变量，IDE 能自己修改：

# 保存音频文件
file_path, file = '', None

# 开始取数据
while task := await Cosmic.queue_in.get():
    ...

    # 创建音频文件
    if Config.save_audio and not file_path:
        file_path, file = create_file(task['data'].shape[1], time_start)
        Cosmic.audio_files[task_id] = file_path

[H1DDENADM1N]

所以我干脆在 while 之前新建这个变量，把它赋值为空

我遇到过几次录音文件被放到1970年的情况，困扰了我很久，但又没见有其他人提issue，我注意到那是UTC时间戳起始点，但这个bug较难复现，可能是这里造成的吗？还是什么其他？

[PabloLION]

录音文件被放到1970年的情况

我也遇见过 @H1DDENADM1N 你既然有源代码，可以加一个条件，用 log 看一下。

[PabloLION]

@HaujetZhao Server Cosmic 的 Queue_in 和 Queue_out 还挺容易读懂的。 client 的花了点时间也读懂了。我感觉用locals()有点危险，确实不如现在的方法。

我还有个问题想请教：send_audio最后一部分有一个写死的 15 和 2，这个是忘改了吗？

message={
    # ...
    "seg_duration": 15,
    "seg_overlap": 2,
 }

[PabloLION]

1. 流式保存，收一秒，存一秒。

这个功能可能放在 client_stream.py 的 record_callback 里面可能好一点，可以存的更完整。不过我看到要改文件名，这个可能不太好办。我没仔细考虑过，可能mktemp有点多余？

[HaujetZhao]

我还有个问题想请教：send_audio最后一部分有一个写死的 15 和 2，这个也是忘改了吗？

我想是的 haha

我遇到过几次录音文件被放到1970年的情况，困扰了我很久，但又没见有其他人提issue，我注意到那是UTC时间戳起始点，但这个bug较难复现，可能是这里造成的吗？还是什么其他？

我知道原因了。但懒得改了。复现概率较小。位置在 send_audio()

async def send_audio():
    try:
        ...
        # 任务起始时间
        time_start = 0
        ...

        # 开始取数据
        while task := await Cosmic.queue_in.get():
            ...
            if task['type'] == 'begin':
                time_start = task['time']

修改 time_start 初始化为 0，只在一处修改过。如果某一个任务的 start 在队列中排错位置了，就有可能在应有的一次任务中取不到它，就改不了时间了。

[PabloLION]

我想把创建文件的放到 if task['type'] == 'begin': 里面，这么做安全不？

[HaujetZhao]

可能mktemp有点多余

不多余的。要确保文件名的唯一性，就得用它。另外，识别完成后，还会重命名为有意义的文件。

[PabloLION]

如果某一个任务的 start 在队列中排错位置了

我感觉是因为queue_in.task_done()没有放在try的最后导致的，try如果报错，可能不应该调用他。

[HaujetZhao]

我想把创建文件的放到 if task['type'] == 'begin': 里面，这么做安全不？

不安全。如你所见，当队列出错时，time_start 没有被正确的赋值，导致了 1970 年的 bug。所以这个队列顺序可能不可靠。具体什么情况下队列顺序出错了，我也不知道怎么复现。

queue_in.task_done() 应该不会出错，它只是一个计数作用，甚至加不加它都不影响，加上它只是用 Queue 的一个传统而已，取完登记一下已取，缓解强迫症。

[PabloLION]

是我记错了。你说的对。我看了文档，GPT说他还有一个作用是block下一次 await 的 queue.join()。

[PabloLION]

我遇到过几次录音文件被放到1970年的情况，困扰了我很久，但又没见有其他人提issue，我注意到那是UTC时间戳起始点，但这个bug较难复现，可能是这里造成的吗？还是什么其他？

我估计我找到原因了。录音时长不到threshold时候，因为 cancel task 直接取消了“收集并发送数据的任务”（client_send_audio.py 中的 send_audio）。此时 queue 中可能还有一些数据残留，所以新的send_audio就会用 time=0创建文件。 修改建议：cancel的时候发送一个type='cancel'的事件，取代task.cancel()。

我没测试能不能复现，我现在有点乱，还在看其他部分。你们有想法可以试试 @HaujetZhao @H1DDENADM1N

[PabloLION]

这个地方有问题

[PabloLION]

此处引入了一个Bug。这个with在退出时会关闭Popen，导致后面的file不可写。需要在merge前修复这个bug。

[HaujetZhao]

😂

[HaujetZhao]

我估计我找到原因了。录音时长不到threshold时候，因为 cancel task 直接取消了“收集并发送数据的任务”（client_send_audio.py 中的 send_audio）。此时 queue 中可能还有一些数据残留，所以新的send_audio就会用 time=0创建文件。 修改建议：cancel的时候发送一个type='cancel'的事件，取代task.cancel()。

可行

[HaujetZhao]

想请教一下@HaujetZhao ：貌似send_audio只是在管理不同task的任务起始时间，cache，filepath之类的变量。这用个dict[task_id, [resources...]]应该能更节省资源。那么此处用async的初衷是为了用asyncio的轮询机制吗？
我想了下还有键盘的按键事件是在新的thread里，不过asyncio.Queue 是 thread-safe 的，好像也没有必要。

因为有一个发送和接收。我的策略并不是发送完音频，然后去建立接收任务等待或轮询，因为我不能保证服务端在零点几秒内就把结果立马给我，我不能一直等（在这个期间 可能我又会建立新的语音任务）。所以等待不能阻塞，我需要左手随便发，右手一直等，只要等到一个回复就立马输出，不管这个回复是不是左手刚发出的任务结果。

能够一直等，不需要轮询，一有结果就能立马输出的，只有 asyncio。

[PabloLION]

@HaujetZhao 我回来了，感谢你的解释。我发现这么写确实有好处，本来决定暂时不改这个逻辑。后来有了第二个担心的点，可能还是需要改一下。

我主要担心的有两点：

    recording_task = asyncio.run_coroutine_threadsafe(
        send_mic_recording(),
        client_app.loop,
    )

第一是在上面的launch task片段中，global recording_task（此前叫task）会被覆盖，所以cancel_task可能会取消错误的recording_task。我们之前讨论的方法（下面的引用）应该能解决这个问题。

我估计我找到原因了。录音时长不到threshold时候，因为 cancel task 直接取消了“收集并发送数据的任务”（client_send_audio.py 中的 send_audio）。此时 queue 中可能还有一些数据残留，所以新的send_audio就会用 time=0创建文件。 修改建议：cancel的时候发送一个type='cancel'的事件，取代task.cancel()。

可行

---

第二：假设我们应用了上面的更改，现在是 Python 在管理 send_mic_recording 的多个call，他们会在收到 queue中有 type=cancel或者type=finished事件时结束。
正常情况下，只会有一个call，但是一旦有两个call（launch_task被触发两次），两个send_mic_recording会在同一个queue_in中抢夺(racing)type=cancel和type=finished事件，也能辨应该停止哪个call。所以我准备直接改成，强制只用一个call，用之前client cosmic 中的on来检查是否有send_mic_recording 正在执行，如果有就不运行新的send_mic_recording

想问一下你的建议

[HaujetZhao]

我使用一个队列是基于以下的逻辑：用户在使用的时候一定是先按下大写锁定键，再松开。再次按下之前必定松开，松开之前必定有按下。那队列里面就一定是这样的状态：启动之后必有对应的结束（或cancel），队列里不会有连续的启动。

启动之前必有结束，就是新的 task 覆盖旧的 task 之前，结束命令必然已经被发出应用到旧的task上。

[PabloLION]

你的意思是，正常情况下，至多有一个 send recording 在运行，我刚才也是这么想的。我的意思是，让python也确认一下这个逻辑，有两种做法：

1. 在 launch, cancel, finish 时，都检测一下 cosmic.on 是否满足条件。小改一下，就能增加稳定性。
2. 实现每次只有一个 send recording 在运行，把 recording task 和 queue in 通过 一个class 绑定在一起，让他们同时创建和销毁。但是这个改动量很大。我感觉不太划算。

[PabloLION]

另外，这么改其实还有一个目的：我想把 click mode 和 hold mode 的快捷键分开。这样比切换 config 要方便的多。但是这就会引入 “有两个 send recording 同时运行” 的问题。
但是我没想好用户交互应该怎么做：假如用户已经开始了 click 识别，又开始了 hold 识别，应该怎么做：

1. 同时执行两个识别，这种情况下用 async 特别好
2. 提示用户不能这么做，并且：2.1 停止 click 模式，并开始 hold 模式。 2.2 继续 click 模式，忽略 hold 模式。但是因为我们没有视觉上的反馈，所以挺难的。顺便说下 UI 方面，我看 better dictation 里会在输入框写上 “listening”， 并且用三个点提示正在输入，可能可以参考。具体我没想好怎么办。

[HaujetZhao]

我的建议是 奥卡姆剃刀 如无必要 勿增实体，从第一性原理出发做开发。架构是服务于需求的。

你为什么想做这个功能，是你真的需要吗，还是做出来能增加你的收入，是真的有用户会在生活中这样干吗？这个需求真的很强烈吗？

你当然可以去改进架构呀，现在的录音数据是往一个队列里面灌，需要的时候就往里灌，不需要的时候就停止，接受者只能有一个，只能有一个正在进行的任务。

如果你真的想做，那你可以改进，抽象点设计，让这个录音数据像喷泉一样不断往外喷，有一个任务的时候就拿一个水管去接，这个任务停止的时候就把水管拔掉拿走，有多个任务的时候就插入多个水管。

[PabloLION]

嗯，有道理，确实是个伪需求（连我自己都没这个需求）。学习了。