戰(zhàn)：BroadcastDomain 與 async_set_broadcast_domains 深度解析)
iTerm2 Python API 輸入廣播Broadcast實(shí)戰(zhàn)BroadcastDomain 與 async_set_broadcast_domains 深度解析【免費(fèi)下載鏈接】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異步函數(shù)。讀完本文你將掌握如何通過腳本將鍵盤輸入同時分發(fā)給同一窗口內(nèi)的多個會話Session、如何按標(biāo)簽頁構(gòu)造廣播域以及如何基于KeystrokeMonitor實(shí)現(xiàn)不對稱廣播等高級輸入分發(fā)方案。一、什么是輸入廣播Broadcast Input在 iTerm2 中輸入廣播指的是當(dāng)用戶在某個會話中鍵入內(nèi)容時這些按鍵會被同時發(fā)送給同一個**廣播域Broadcast Domain**內(nèi)的所有會話。這是多機(jī)運(yùn)維、批量操作、同步執(zhí)行命令時最常用的功能之一。從源碼 broadcast.py 的注釋可以明確廣播域的三條核心語義按鍵歸屬用戶在一個會話中鍵入按鍵會發(fā)送給該廣播域內(nèi)的全部會話域互斥disjoint廣播域之間是互斥的一個會話只可能屬于一個廣播域域與窗口綁定輸入廣播只發(fā)生在屬于同一個窗口Window的會話之間且每個窗口最多只能有一個廣播域。二、核心 API 總覽由 broadcast.rst 通過automodule/autoclass/autofunction指令導(dǎo)出的公開接口如下API類型說明iterm2.BroadcastDomain類描述一個廣播域即一組共享鍵盤輸入的會話集合BroadcastDomain.add_session(session)方法向廣播域添加一個iterm2.SessionBroadcastDomain.sessions屬性返回該廣播域當(dāng)前包含的所有會話列表iterm2.async_set_broadcast_domains(connection, broadcast_domains)異步函數(shù)將當(dāng)前廣播域集合整體替換為傳入的新集合這些接口所在的完整 Python 包位于 iterm2/broadcast.py是 iTerm2 官方 Python 腳本庫的一部分。三、BroadcastDomain廣播域的構(gòu)建BroadcastDomain是描述哪些會話共享輸入的容器其構(gòu)造與使用方式如下import iterm2 domain iterm2.BroadcastDomain() domain.add_session(session_a) domain.add_session(session_b)3.1 add_session加入會話add_session接受一個 iterm2.session.Session 對象將其追加到廣播域內(nèi)部列表中def add_session(self, session: iterm2.session.Session): self.__sessions.append(session)在收集會話對象時典型的來源是遍歷窗口與標(biāo)簽頁例如tab.sessions一個標(biāo)簽頁下的所有分割窗格會話、app.terminal_windows[i].tabs窗口內(nèi)的所有標(biāo)簽頁。3.2 sessions讀取域內(nèi)會話sessions是只讀屬性返回域內(nèi)所有有效會話的列表。其實(shí)現(xiàn)值得注意——它把已顯式添加的會話與尚未解析的會話合并并過濾掉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由內(nèi)部方法add_unresolved填充用于承載尚未最終確定的會話引用通過惰性解析閉包r()延遲求值。對普通使用者而言只需理解凡是成功add_session進(jìn)去的會話都會出現(xiàn)在sessions中最終調(diào)用設(shè)置函數(shù)時即以此列表為準(zhǔn)。四、async_set_broadcast_domains應(yīng)用廣播域async_set_broadcast_domains是廣播功能的入口它接受兩個參數(shù)connection與 iTerm2 進(jìn)程建立的 iterm2.connection.Connection 連接對象broadcast_domains一個List[BroadcastDomain]即新的廣播域集合調(diào)用后整體替換當(dāng)前配置。await iterm2.async_set_broadcast_domains(connection, [domain])4.1 底層 RPC 調(diào)用鏈從源碼結(jié)構(gòu)看該函數(shù)并非直接與 iTerm2 通信而是先把廣播域轉(zhuǎn)換為會話 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 中每個內(nèi)層列表會被包裝成一個 protobufBroadcastDomain消息session_ids字段填充對應(yīng)會話 ID最終組裝進(jìn)set_broadcast_domains_request發(fā)送給 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 錯誤處理調(diào)用返回后broadcast.py 會檢查響應(yīng)狀態(tài)。若 iTerm2 返回的狀態(tài)不是OK則拋出iterm2.rpc.RPCException異常信息中包含來自 protobuf 的SetBroadcastDomainsResponse.Status狀態(tài)名便于定位失敗原因如會話不存在、域內(nèi)跨窗口等非法配置。五、實(shí)戰(zhàn)示例一為每個標(biāo)簽頁開啟廣播官方示例 enable_broadcasting.rst可執(zhí)行腳本見同目錄 enable_broadcasting.its演示了最典型的用法把第一個窗口內(nèi)每個標(biāo)簽頁的第一個會話加入同一個廣播域。#!/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)要點(diǎn)拆解iterm2.async_get_app(connection)獲取應(yīng)用模型對象appapp.terminal_windows[0]指向第一個終端窗口遍歷該窗口下所有標(biāo)簽頁tabs每個標(biāo)簽頁取第一個會話tab.sessions[0]對于只有一個分割窗格的標(biāo)簽頁即唯一會話把所有這些會話加入同一個BroadcastDomain調(diào)用async_set_broadcast_domains(connection, [domain])完成設(shè)置——由于每個窗口最多只能有一個廣播域這里傳入的列表只包含一個域腳本以iterm2.run_until_complete(main)啟動異步主流程。運(yùn)行后在任一被廣播的會話中鍵入按鍵會同步出現(xiàn)在該窗口所有標(biāo)簽頁的首個會話中。若要關(guān)閉廣播只需傳入空列表await iterm2.async_set_broadcast_domains(connection, [])。六、實(shí)戰(zhàn)示例二不對稱廣播自定義按鍵分發(fā)內(nèi)置廣播是對稱的只要會話屬于同一廣播域輸入就會流向域內(nèi)所有會話。而官方示例 broadcast.rst腳本見 broadcast.its展示了如何繞過內(nèi)置廣播實(shí)現(xiàn)不對稱廣播創(chuàng)建四個分割窗格只把左下角窗格的輸入轉(zhuǎn)發(fā)給其他三個而其他窗格自身的輸入不會被廣播。#!/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)這段代碼綜合運(yùn)用了三個關(guān)鍵機(jī)制iterm2.KeystrokeMonitor監(jiān)聽指定會話的按鍵流async_get()逐個取出按鍵事件iterm2.KeystrokeFilterKeystrokePattern構(gòu)造一個匹配除帶 Command 修飾鍵外所有按鍵的模式交給KeystrokeFilter吞掉這些按鍵避免 iTerm2 在內(nèi)置層面處理它們forbidden_modifiers [iterm2.Modifier.COMMAND]保留了 Command 組合鍵給 iTerm2 正常使用Session.async_send_text將捕獲到的按鍵字符keystroke.characters以偽造鍵入的方式逐一發(fā)送給目標(biāo)會話列表實(shí)現(xiàn)單向、不對稱的輸入分發(fā)。按鍵在async_handle_keystroke中集中處理遇到 ESC 返回True結(jié)束監(jiān)聽并喚醒過濾器任務(wù)其余按鍵全部轉(zhuǎn)發(fā)給broadcast_to中的四個會話包含源會話自身。這正是一個自己實(shí)現(xiàn)廣播的范式——當(dāng)內(nèi)置廣播無法滿足不對稱、有選擇性的分發(fā)需求時可以借用本示例的思路。七、運(yùn)行環(huán)境與使用前提廣播功能由 iTerm2 主程序macOS 平臺的終端模擬器提供Python 腳本庫僅通過 API 連接下發(fā)指令廣播的實(shí)際生效依賴運(yùn)行中的 iTerm2 實(shí)例腳本庫源碼位于 api/library/python/iterm2/iterm2其 API 定義、protobuf 消息api_pb2.py與 RPC 層rpc.py共同構(gòu)成廣播功能的下行鏈路更多腳本示例可參考 docs/examples 目錄其中 broadcast.rst 與 enable_broadcasting.rst 均提供了可下載的.its腳本文件可直接運(yùn)行驗(yàn)證本指南描述的行為廣播域受同窗口與每窗口至多一個域約束構(gòu)造域時請勿跨窗口混入會話否則可能導(dǎo)致設(shè)置失敗對應(yīng)RPCException。八、小結(jié)iTerm2 的廣播 API 雖然只有兩個公開入口BroadcastDomain與async_set_broadcast_domains但配合窗口-標(biāo)簽頁-會話的層級模型足以覆蓋從整窗口批量廣播到自定義不對稱分發(fā)的各類輸入同步場景。理解其底層 RPC 與 protobuf 封裝有助于在遇到會話無效、跨窗口分組等邊界情況時快速定位問題而結(jié)合KeystrokeMonitor、KeystrokeFilter與async_send_text的組合拳則讓廣播能力從內(nèi)置固定模式擴(kuò)展為完全可控的編程式輸入路由。【免費(fèi)下載鏈接】iTerm2iTerm2 is a terminal emulator for Mac OS X that does amazing things.項目地址: https://gitcode.com/gh_mirrors/it/iTerm2創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 iTerm2 Python API 輸入廣播（Broadcast）實(shí)戰(zhàn)：BroadcastDomain 與 async_set_broadcast_domains 深度解析
iTerm2 Python API 輸入廣播Broadcast實(shí)戰(zhàn)BroadcastDomain 與 async_set_broadcast_domains 深度解析【免費(fèi)下載鏈接】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異步函數(shù)。讀完本文你將掌握如何通過腳本將鍵盤輸入同時分發(fā)給同一窗口內(nèi)的多個會話Session、如何按標(biāo)簽頁構(gòu)造廣播域以及如何基于KeystrokeMonitor實(shí)現(xiàn)不對稱廣播等高級輸入分發(fā)方案。一、什么是輸入廣播Broadcast Input在 iTerm2 中輸入廣播指的是當(dāng)用戶在某個會話中鍵入內(nèi)容時這些按鍵會被同時發(fā)送給同一個**廣播域Broadcast Domain**內(nèi)的所有會話。這是多機(jī)運(yùn)維、批量操作、同步執(zhí)行命令時最常用的功能之一。從源碼 broadcast.py 的注釋可以明確廣播域的三條核心語義按鍵歸屬用戶在一個會話中鍵入按鍵會發(fā)送給該廣播域內(nèi)的全部會話域互斥disjoint廣播域之間是互斥的一個會話只可能屬于一個廣播域域與窗口綁定輸入廣播只發(fā)生在屬于同一個窗口Window的會話之間且每個窗口最多只能有一個廣播域。二、核心 API 總覽由 broadcast.rst 通過automodule/autoclass/autofunction指令導(dǎo)出的公開接口如下API類型說明iterm2.BroadcastDomain類描述一個廣播域即一組共享鍵盤輸入的會話集合BroadcastDomain.add_session(session)方法向廣播域添加一個iterm2.SessionBroadcastDomain.sessions屬性返回該廣播域當(dāng)前包含的所有會話列表iterm2.async_set_broadcast_domains(connection, broadcast_domains)異步函數(shù)將當(dāng)前廣播域集合整體替換為傳入的新集合這些接口所在的完整 Python 包位于 iterm2/broadcast.py是 iTerm2 官方 Python 腳本庫的一部分。三、BroadcastDomain廣播域的構(gòu)建BroadcastDomain是描述哪些會話共享輸入的容器其構(gòu)造與使用方式如下import iterm2 domain iterm2.BroadcastDomain() domain.add_session(session_a) domain.add_session(session_b)3.1 add_session加入會話add_session接受一個 iterm2.session.Session 對象將其追加到廣播域內(nèi)部列表中def add_session(self, session: iterm2.session.Session): self.__sessions.append(session)在收集會話對象時典型的來源是遍歷窗口與標(biāo)簽頁例如tab.sessions一個標(biāo)簽頁下的所有分割窗格會話、app.terminal_windows[i].tabs窗口內(nèi)的所有標(biāo)簽頁。3.2 sessions讀取域內(nèi)會話sessions是只讀屬性返回域內(nèi)所有有效會話的列表。其實(shí)現(xiàn)值得注意——它把已顯式添加的會話與尚未解析的會話合并并過濾掉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由內(nèi)部方法add_unresolved填充用于承載尚未最終確定的會話引用通過惰性解析閉包r()延遲求值。對普通使用者而言只需理解凡是成功add_session進(jìn)去的會話都會出現(xiàn)在sessions中最終調(diào)用設(shè)置函數(shù)時即以此列表為準(zhǔn)。四、async_set_broadcast_domains應(yīng)用廣播域async_set_broadcast_domains是廣播功能的入口它接受兩個參數(shù)connection與 iTerm2 進(jìn)程建立的 iterm2.connection.Connection 連接對象broadcast_domains一個List[BroadcastDomain]即新的廣播域集合調(diào)用后整體替換當(dāng)前配置。await iterm2.async_set_broadcast_domains(connection, [domain])4.1 底層 RPC 調(diào)用鏈從源碼結(jié)構(gòu)看該函數(shù)并非直接與 iTerm2 通信而是先把廣播域轉(zhuǎn)換為會話 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 中每個內(nèi)層列表會被包裝成一個 protobufBroadcastDomain消息session_ids字段填充對應(yīng)會話 ID最終組裝進(jìn)set_broadcast_domains_request發(fā)送給 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 錯誤處理調(diào)用返回后broadcast.py 會檢查響應(yīng)狀態(tài)。若 iTerm2 返回的狀態(tài)不是OK則拋出iterm2.rpc.RPCException異常信息中包含來自 protobuf 的SetBroadcastDomainsResponse.Status狀態(tài)名便于定位失敗原因如會話不存在、域內(nèi)跨窗口等非法配置。五、實(shí)戰(zhàn)示例一為每個標(biāo)簽頁開啟廣播官方示例 enable_broadcasting.rst可執(zhí)行腳本見同目錄 enable_broadcasting.its演示了最典型的用法把第一個窗口內(nèi)每個標(biāo)簽頁的第一個會話加入同一個廣播域。#!/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)要點(diǎn)拆解iterm2.async_get_app(connection)獲取應(yīng)用模型對象appapp.terminal_windows[0]指向第一個終端窗口遍歷該窗口下所有標(biāo)簽頁tabs每個標(biāo)簽頁取第一個會話tab.sessions[0]對于只有一個分割窗格的標(biāo)簽頁即唯一會話把所有這些會話加入同一個BroadcastDomain調(diào)用async_set_broadcast_domains(connection, [domain])完成設(shè)置——由于每個窗口最多只能有一個廣播域這里傳入的列表只包含一個域腳本以iterm2.run_until_complete(main)啟動異步主流程。運(yùn)行后在任一被廣播的會話中鍵入按鍵會同步出現(xiàn)在該窗口所有標(biāo)簽頁的首個會話中。若要關(guān)閉廣播只需傳入空列表await iterm2.async_set_broadcast_domains(connection, [])。六、實(shí)戰(zhàn)示例二不對稱廣播自定義按鍵分發(fā)內(nèi)置廣播是對稱的只要會話屬于同一廣播域輸入就會流向域內(nèi)所有會話。而官方示例 broadcast.rst腳本見 broadcast.its展示了如何繞過內(nèi)置廣播實(shí)現(xiàn)不對稱廣播創(chuàng)建四個分割窗格只把左下角窗格的輸入轉(zhuǎn)發(fā)給其他三個而其他窗格自身的輸入不會被廣播。#!/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)這段代碼綜合運(yùn)用了三個關(guān)鍵機(jī)制iterm2.KeystrokeMonitor監(jiān)聽指定會話的按鍵流async_get()逐個取出按鍵事件iterm2.KeystrokeFilterKeystrokePattern構(gòu)造一個匹配除帶 Command 修飾鍵外所有按鍵的模式交給KeystrokeFilter吞掉這些按鍵避免 iTerm2 在內(nèi)置層面處理它們forbidden_modifiers [iterm2.Modifier.COMMAND]保留了 Command 組合鍵給 iTerm2 正常使用Session.async_send_text將捕獲到的按鍵字符keystroke.characters以偽造鍵入的方式逐一發(fā)送給目標(biāo)會話列表實(shí)現(xiàn)單向、不對稱的輸入分發(fā)。按鍵在async_handle_keystroke中集中處理遇到 ESC 返回True結(jié)束監(jiān)聽并喚醒過濾器任務(wù)其余按鍵全部轉(zhuǎn)發(fā)給broadcast_to中的四個會話包含源會話自身。這正是一個自己實(shí)現(xiàn)廣播的范式——當(dāng)內(nèi)置廣播無法滿足不對稱、有選擇性的分發(fā)需求時可以借用本示例的思路。七、運(yùn)行環(huán)境與使用前提廣播功能由 iTerm2 主程序macOS 平臺的終端模擬器提供Python 腳本庫僅通過 API 連接下發(fā)指令廣播的實(shí)際生效依賴運(yùn)行中的 iTerm2 實(shí)例腳本庫源碼位于 api/library/python/iterm2/iterm2其 API 定義、protobuf 消息api_pb2.py與 RPC 層rpc.py共同構(gòu)成廣播功能的下行鏈路更多腳本示例可參考 docs/examples 目錄其中 broadcast.rst 與 enable_broadcasting.rst 均提供了可下載的.its腳本文件可直接運(yùn)行驗(yàn)證本指南描述的行為廣播域受同窗口與每窗口至多一個域約束構(gòu)造域時請勿跨窗口混入會話否則可能導(dǎo)致設(shè)置失敗對應(yīng)RPCException。八、小結(jié)iTerm2 的廣播 API 雖然只有兩個公開入口BroadcastDomain與async_set_broadcast_domains但配合窗口-標(biāo)簽頁-會話的層級模型足以覆蓋從整窗口批量廣播到自定義不對稱分發(fā)的各類輸入同步場景。理解其底層 RPC 與 protobuf 封裝有助于在遇到會話無效、跨窗口分組等邊界情況時快速定位問題而結(jié)合KeystrokeMonitor、KeystrokeFilter與async_send_text的組合拳則讓廣播能力從內(nèi)置固定模式擴(kuò)展為完全可控的編程式輸入路由。【免費(fèi)下載鏈接】iTerm2iTerm2 is a terminal emulator for Mac OS X that does amazing things.項目地址: https://gitcode.com/gh_mirrors/it/iTerm2創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考