199 lines
7.4 KiB
Python
199 lines
7.4 KiB
Python
|
from pynput import keyboard
|
|||
|
|
from PySide2.QtCore import QMutex
|
|||
|
|
from time import time
|
|||
|
|
|
|||
|
|
from umi_log import logger
|
|||
|
|
from ...platform import Platform
|
|||
|
|
from ..pubsub_service import PubSubService
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 按键转换器
|
|||
|
|
class _KeyTranslator:
|
|||
|
|
# 回调函数的 KeyCode 类型 转为键名字符串
|
|||
|
|
@staticmethod
|
|||
|
|
def key2name(key):
|
|||
|
|
return Platform.getKeyName(key)
|
|||
|
|
|
|||
|
|
# 键名字符串 转为KeyCode
|
|||
|
|
@staticmethod
|
|||
|
|
def name2key(char):
|
|||
|
|
if hasattr(keyboard.Key, char): # 控制键返回Code
|
|||
|
|
return getattr(keyboard.Key, char).value
|
|||
|
|
else: # 非控制键返回自己
|
|||
|
|
return char
|
|||
|
|
|
|||
|
|
# 多个键名的字符串 转为集合
|
|||
|
|
@staticmethod
|
|||
|
|
def names2set(char):
|
|||
|
|
return set(char.split("+"))
|
|||
|
|
|
|||
|
|
# 集合转键名
|
|||
|
|
def set2names(keys):
|
|||
|
|
keys = keys.copy()
|
|||
|
|
# 优先级键名
|
|||
|
|
highPriority = ("win", "cmd", "shift", "ctrl", "alt")
|
|||
|
|
names = ""
|
|||
|
|
# 添加高优先级键
|
|||
|
|
for h in highPriority:
|
|||
|
|
if h in keys:
|
|||
|
|
if names != "":
|
|||
|
|
names += "+"
|
|||
|
|
names += h
|
|||
|
|
keys.discard(h)
|
|||
|
|
# 添加剩下的键
|
|||
|
|
for k in keys:
|
|||
|
|
if names != "":
|
|||
|
|
names += "+"
|
|||
|
|
names += k
|
|||
|
|
return names
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 热键控制器类
|
|||
|
|
class _HotkeyController:
|
|||
|
|
# ========================= 【接口】 =========================
|
|||
|
|
|
|||
|
|
# 添加一组快捷键，对应触发事件为title。 press 为0时按下触发，1抬起触发
|
|||
|
|
def addHotkey(self, keysName, title, press=0):
|
|||
|
|
self._start()
|
|||
|
|
if press != 0 and press != 1:
|
|||
|
|
return f"[Error] press只能为0按下或1抬起，不能为 {press} 。"
|
|||
|
|
keySet = _KeyTranslator.names2set(keysName)
|
|||
|
|
self._hotkeyMutex.lock()
|
|||
|
|
kl = self._hotkeyList[press]
|
|||
|
|
for k in kl: # 检测重复
|
|||
|
|
if k["keySet"] == keySet: # 键集合相同
|
|||
|
|
self._hotkeyMutex.unlock()
|
|||
|
|
msg = "[Success] 注册事件相同的重复快捷键。"
|
|||
|
|
if k["title"] != title: # 事件标题不同
|
|||
|
|
msg = f'[Warning] Registering same hotkey. The existing event for {keysName} is {k["title"]}, new event is {title} .'
|
|||
|
|
return msg
|
|||
|
|
# 加入列表
|
|||
|
|
kl.append({"keySet": keySet, "title": title})
|
|||
|
|
self._hotkeyMutex.unlock()
|
|||
|
|
return "[Success]"
|
|||
|
|
|
|||
|
|
# 移除一组快捷键，传入键名或事件之一
|
|||
|
|
def delHotkey(self, keysName="", title="", press=0):
|
|||
|
|
if press != 0 and press != 1:
|
|||
|
|
logger.error(f"press只能为0按下或1抬起，不能为 {press} 。")
|
|||
|
|
return
|
|||
|
|
keySet = _KeyTranslator.names2set(keysName)
|
|||
|
|
self._hotkeyMutex.lock()
|
|||
|
|
kListOld = self._hotkeyList[press] # 旧列表
|
|||
|
|
kListNew = [] # 新列表
|
|||
|
|
for k in kListOld:
|
|||
|
|
# 忽略键集合相同或标题相同
|
|||
|
|
if (keysName and k["keySet"] == keySet) or (title and title == k["title"]):
|
|||
|
|
pass
|
|||
|
|
# 其余键写入新列表
|
|||
|
|
else:
|
|||
|
|
kListNew.append(k)
|
|||
|
|
self._hotkeyList[press] = kListNew
|
|||
|
|
self._hotkeyMutex.unlock()
|
|||
|
|
|
|||
|
|
# 开始录制快捷键。过程发送事件为runningTitle，完毕发送事件为finishTitle
|
|||
|
|
def readHotkey(
|
|||
|
|
self, runningTitle="<<readHotkeyRunning>>", finishTitle="<<readHotkeyFinish>>"
|
|||
|
|
):
|
|||
|
|
self._start()
|
|||
|
|
if self._status == 1:
|
|||
|
|
return (
|
|||
|
|
"[Warning] Recording is running. 当前快捷键录制已在进行，不能同时录制！"
|
|||
|
|
)
|
|||
|
|
self._status = 1
|
|||
|
|
self._readRunningTitle = runningTitle
|
|||
|
|
self._readFinishTitle = finishTitle
|
|||
|
|
return "[Success]"
|
|||
|
|
|
|||
|
|
# ========================= 【实现】 =========================
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self._listener = None # 监听器
|
|||
|
|
# 热键列表，[0]存放按下触发，[1]存放抬起触发
|
|||
|
|
# 每个元素为：{"keySet":按键集合, "title":事件标题}
|
|||
|
|
self._hotkeyList = [[], []]
|
|||
|
|
self._hotkeyMutex = QMutex() # 热键列表的锁
|
|||
|
|
self._status = 0 # 状态，0正常，1录制中
|
|||
|
|
self._pressSet = set() # 当前已按下的按键集合
|
|||
|
|
self._strict = True # 键集合相等的判定，T为严格，F为宽松
|
|||
|
|
self._ttl = 30 # 长按键超时忽略时间，秒
|
|||
|
|
self._ttlDict = {} # 存放当前已按下按键的超时时间
|
|||
|
|
self._readRunningTitle = ""
|
|||
|
|
self._readFinishTitle = ""
|
|||
|
|
# 注册默认监听的热键
|
|||
|
|
self.addHotkey("esc", "<<esc>>")
|
|||
|
|
|
|||
|
|
# 第一次注册热键时，启动监听
|
|||
|
|
def _start(self):
|
|||
|
|
if not self._listener:
|
|||
|
|
self._listener = keyboard.Listener(
|
|||
|
|
on_press=self._onPress, on_release=self._onRelease
|
|||
|
|
)
|
|||
|
|
self._listener.start()
|
|||
|
|
|
|||
|
|
# 按键按下的回调
|
|||
|
|
def _onPress(self, key):
|
|||
|
|
keyName = _KeyTranslator.key2name(key)
|
|||
|
|
# 禁止重复触发
|
|||
|
|
if keyName in self._pressSet:
|
|||
|
|
return
|
|||
|
|
self._checkTTL() # 检查超时
|
|||
|
|
self._pressSet.add(keyName) # 加入集合
|
|||
|
|
self._ttlDict[keyName] = time() + self._ttl # 记录超时时间
|
|||
|
|
if self._status == 0: # 正常运行
|
|||
|
|
self._checkKeyEvent(0, keyName) # 检查按下模式的快捷键
|
|||
|
|
elif self._status == 1: # 录制中
|
|||
|
|
self._readRunning()
|
|||
|
|
|
|||
|
|
# 按键抬起的回调
|
|||
|
|
def _onRelease(self, key):
|
|||
|
|
keyName = _KeyTranslator.key2name(key)
|
|||
|
|
if not keyName in self._pressSet:
|
|||
|
|
return
|
|||
|
|
self._checkTTL() # 检查超时
|
|||
|
|
if self._status == 0: # 正常运行
|
|||
|
|
self._checkKeyEvent(1, keyName) # 检查抬起模式的快捷键
|
|||
|
|
elif self._status == 1: # 录制结束
|
|||
|
|
self._readFinish()
|
|||
|
|
if keyName in self._pressSet:
|
|||
|
|
self._pressSet.discard(keyName) # 从集合中删除
|
|||
|
|
del self._ttlDict[keyName] # 删除超时时间
|
|||
|
|
|
|||
|
|
# 检查并触发按键事件
|
|||
|
|
def _checkKeyEvent(self, press, nowKey):
|
|||
|
|
# 对比每组按键集合。一致触发，则发送事件
|
|||
|
|
if self._strict: # 严格模式，要求完全一致
|
|||
|
|
for k in self._hotkeyList[press]:
|
|||
|
|
if k["keySet"] == self._pressSet:
|
|||
|
|
PubSubService.publish(k["title"])
|
|||
|
|
else: # 宽松模式，只要求当前组合中包含指定按键，且当前按下的按键在指定按键中
|
|||
|
|
for k in self._hotkeyList[press]:
|
|||
|
|
if k["keySet"] <= self._pressSet and nowKey in k["keySet"]:
|
|||
|
|
PubSubService.publish(k["title"])
|
|||
|
|
|
|||
|
|
# 更新录制
|
|||
|
|
def _readRunning(self):
|
|||
|
|
names = _KeyTranslator.set2names(self._pressSet)
|
|||
|
|
PubSubService.publish(self._readRunningTitle, names)
|
|||
|
|
|
|||
|
|
# 录制结束
|
|||
|
|
def _readFinish(self):
|
|||
|
|
self._status = 0
|
|||
|
|
if "esc" in self._pressSet: # 含esc，则为退出
|
|||
|
|
PubSubService.publish(self._readFinishTitle, "")
|
|||
|
|
else:
|
|||
|
|
names = _KeyTranslator.set2names(self._pressSet)
|
|||
|
|
PubSubService.publish(self._readFinishTitle, names)
|
|||
|
|
|
|||
|
|
# 检查已按键的超时时间。若超时，则删除该键
|
|||
|
|
def _checkTTL(self):
|
|||
|
|
nowTime = time()
|
|||
|
|
for k in self._pressSet.copy():
|
|||
|
|
if nowTime >= self._ttlDict[k]:
|
|||
|
|
logger.debug(f"忽略超时按键 {k}")
|
|||
|
|
del self._ttlDict[k]
|
|||
|
|
self._pressSet.discard(k)
|
|||
|
|
|
|||
|
|
|
|||
|
|
HotkeyCtrl = _HotkeyController()
|