iTerm2 Python API 输入广播Broadcast实战BroadcastDomain 与 async_set_broadcast_domains 深度解析【免费下载链接】iTerm2iTerm2 is a terminal emulator for Mac OS X that does amazing things.项目地址: https://gitcode.com/gh_mirrors/it/iTerm2本指南围绕 iTerm2 Python 脚本 API 中的广播功能展开核心讲解 broadcast.rst 定义的iterm2.BroadcastDomain类与iterm2.async_set_broadcast_domains异步函数。读完本文你将掌握如何通过脚本将键盘输入同时分发给同一窗口内的多个会话Session、如何按标签页构造广播域以及如何基于KeystrokeMonitor实现不对称广播等高级输入分发方案。一、什么是输入广播Broadcast Input在 iTerm2 中输入广播指的是当用户在某个会话中键入内容时这些按键会被同时发送给同一个**广播域Broadcast Domain**内的所有会话。这是多机运维、批量操作、同步执行命令时最常用的功能之一。从源码 broadcast.py 的注释可以明确广播域的三条核心语义按键归属用户在一个会话中键入按键会发送给该广播域内的全部会话域互斥disjoint广播域之间是互斥的一个会话只可能属于一个广播域域与窗口绑定输入广播只发生在属于同一个窗口Window的会话之间且每个窗口最多只能有一个广播域。二、核心 API 总览由 broadcast.rst 通过automodule/autoclass/autofunction指令导出的公开接口如下API类型说明iterm2.BroadcastDomain类描述一个广播域即一组共享键盘输入的会话集合BroadcastDomain.add_session(session)方法向广播域添加一个iterm2.SessionBroadcastDomain.sessions属性返回该广播域当前包含的所有会话列表iterm2.async_set_broadcast_domains(connection, broadcast_domains)异步函数将当前广播域集合整体替换为传入的新集合这些接口所在的完整 Python 包位于 iterm2/broadcast.py是 iTerm2 官方 Python 脚本库的一部分。三、BroadcastDomain广播域的构建BroadcastDomain是描述哪些会话共享输入的容器其构造与使用方式如下import iterm2 domain iterm2.BroadcastDomain() domain.add_session(session_a) domain.add_session(session_b)3.1 add_session加入会话add_session接受一个 iterm2.session.Session 对象将其追加到广播域内部列表中def add_session(self, session: iterm2.session.Session): self.__sessions.append(session)在收集会话对象时典型的来源是遍历窗口与标签页例如tab.sessions一个标签页下的所有分割窗格会话、app.terminal_windows[i].tabs窗口内的所有标签页。3.2 sessions读取域内会话sessions是只读属性返回域内所有有效会话的列表。其实现值得注意——它把已显式添加的会话与尚未解析的会话合并并过滤掉Noneproperty def sessions(self) - typing.List[iterm2.session.Session]: return list(filter( lambda x: x is not None, self.__sessions list(map(lambda r: r(), self.__unresolved))))其中__unresolved由内部方法add_unresolved填充用于承载尚未最终确定的会话引用通过惰性解析闭包r()延迟求值。对普通使用者而言只需理解凡是成功add_session进去的会话都会出现在sessions中最终调用设置函数时即以此列表为准。四、async_set_broadcast_domains应用广播域async_set_broadcast_domains是广播功能的入口它接受两个参数connection与 iTerm2 进程建立的 iterm2.connection.Connection 连接对象broadcast_domains一个List[BroadcastDomain]即新的广播域集合调用后整体替换当前配置。await iterm2.async_set_broadcast_domains(connection, [domain])4.1 底层 RPC 调用链从源码结构看该函数并非直接与 iTerm2 通信而是先把广播域转换为会话 ID 的二维列表再交给 RPC 层处理见 broadcast.pyresponse await iterm2.rpc.async_set_broadcast_domains( connection, list(map(lambda d: list(map(lambda s: s.session_id, d.sessions)), broadcast_domains)))在 RPC 层 rpc.py 中每个内层列表会被包装成一个 protobufBroadcastDomain消息session_ids字段填充对应会话 ID最终组装进set_broadcast_domains_request发送给 iTerm2 主程序proto iterm2.api_pb2.BroadcastDomain() proto.session_ids.extend(list_of_session_ids) domains_protos.append(proto) request.set_broadcast_domains_request.broadcast_domains.extend(domains_protos)4.2 错误处理调用返回后broadcast.py 会检查响应状态。若 iTerm2 返回的状态不是OK则抛出iterm2.rpc.RPCException异常信息中包含来自 protobuf 的SetBroadcastDomainsResponse.Status状态名便于定位失败原因如会话不存在、域内跨窗口等非法配置。五、实战示例一为每个标签页开启广播官方示例 enable_broadcasting.rst可执行脚本见同目录 enable_broadcasting.its演示了最典型的用法把第一个窗口内每个标签页的第一个会话加入同一个广播域。#!/usr/bin/env python3 import iterm2 async def main(connection): app await iterm2.async_get_app(connection) domain iterm2.broadcast.BroadcastDomain() for tab in app.terminal_windows[0].tabs: domain.add_session(tab.sessions[0]) await iterm2.async_set_broadcast_domains(connection, [domain]) iterm2.run_until_complete(main)要点拆解iterm2.async_get_app(connection)获取应用模型对象appapp.terminal_windows[0]指向第一个终端窗口遍历该窗口下所有标签页tabs每个标签页取第一个会话tab.sessions[0]对于只有一个分割窗格的标签页即唯一会话把所有这些会话加入同一个BroadcastDomain调用async_set_broadcast_domains(connection, [domain])完成设置——由于每个窗口最多只能有一个广播域这里传入的列表只包含一个域脚本以iterm2.run_until_complete(main)启动异步主流程。运行后在任一被广播的会话中键入按键会同步出现在该窗口所有标签页的首个会话中。若要关闭广播只需传入空列表await iterm2.async_set_broadcast_domains(connection, [])。六、实战示例二不对称广播自定义按键分发内置广播是对称的只要会话属于同一广播域输入就会流向域内所有会话。而官方示例 broadcast.rst脚本见 broadcast.its展示了如何绕过内置广播实现不对称广播创建四个分割窗格只把左下角窗格的输入转发给其他三个而其他窗格自身的输入不会被广播。#!/usr/bin/env python3 import asyncio import iterm2 async def main(connection): app await iterm2.async_get_app(connection) # Create four split panes and make the bottom left one active. bottomLeft app.current_terminal_window.current_tab.current_session bottomRight await bottomLeft.async_split_pane(verticalTrue) topLeft await bottomLeft.async_split_pane(verticalFalse, beforeTrue) topRight await bottomRight.async_split_pane(verticalFalse, beforeTrue) await bottomLeft.async_activate() broadcast_to [ topLeft, bottomLeft, topRight, bottomRight ] async def async_handle_keystroke(keystroke): if keystroke.keycode iterm2.Keycode.ESCAPE: # User pressed escape. Terminate script. return True for session in broadcast_to: await session.async_send_text(keystroke.characters) return False # Construct a pattern that matches all keystrokes except those with a Command modifier. pattern iterm2.KeystrokePattern() pattern.keycodes [keycode for keycode in iterm2.Keycode] pattern.forbidden_modifiers [iterm2.Modifier.COMMAND] future asyncio.Future() # Swallow all keystrokes matching the pattern async def filter_all_keystrokes(): async with iterm2.KeystrokeFilter(connection, [pattern], bottomLeft.session_id) as mon: await asyncio.wait([future]) task asyncio.create_task(filter_all_keystrokes()) # This will block until async_handle_keystroke returns True. async with iterm2.KeystrokeMonitor(connection, bottomLeft.session_id) as mon: done False while not done: keystroke await mon.async_get() done await async_handle_keystroke(keystroke) if done: break future.set_result(True) await task iterm2.run_until_complete(main)这段代码综合运用了三个关键机制iterm2.KeystrokeMonitor监听指定会话的按键流async_get()逐个取出按键事件iterm2.KeystrokeFilterKeystrokePattern构造一个匹配除带 Command 修饰键外所有按键的模式交给KeystrokeFilter吞掉这些按键避免 iTerm2 在内置层面处理它们forbidden_modifiers [iterm2.Modifier.COMMAND]保留了 Command 组合键给 iTerm2 正常使用Session.async_send_text将捕获到的按键字符keystroke.characters以伪造键入的方式逐一发送给目标会话列表实现单向、不对称的输入分发。按键在async_handle_keystroke中集中处理遇到 ESC 返回True结束监听并唤醒过滤器任务其余按键全部转发给broadcast_to中的四个会话包含源会话自身。这正是一个自己实现广播的范式——当内置广播无法满足不对称、有选择性的分发需求时可以借用本示例的思路。七、运行环境与使用前提广播功能由 iTerm2 主程序macOS 平台的终端模拟器提供Python 脚本库仅通过 API 连接下发指令广播的实际生效依赖运行中的 iTerm2 实例脚本库源码位于 api/library/python/iterm2/iterm2其 API 定义、protobuf 消息api_pb2.py与 RPC 层rpc.py共同构成广播功能的下行链路更多脚本示例可参考 docs/examples 目录其中 broadcast.rst 与 enable_broadcasting.rst 均提供了可下载的.its脚本文件可直接运行验证本指南描述的行为广播域受同窗口与每窗口至多一个域约束构造域时请勿跨窗口混入会话否则可能导致设置失败对应RPCException。八、小结iTerm2 的广播 API 虽然只有两个公开入口BroadcastDomain与async_set_broadcast_domains但配合窗口-标签页-会话的层级模型足以覆盖从整窗口批量广播到自定义不对称分发的各类输入同步场景。理解其底层 RPC 与 protobuf 封装有助于在遇到会话无效、跨窗口分组等边界情况时快速定位问题而结合KeystrokeMonitor、KeystrokeFilter与async_send_text的组合拳则让广播能力从内置固定模式扩展为完全可控的编程式输入路由。【免费下载链接】iTerm2iTerm2 is a terminal emulator for Mac OS X that does amazing things.项目地址: https://gitcode.com/gh_mirrors/it/iTerm2创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
