【python项目推荐】键盘监控--统计打字频率

2024-04-22 19:20
文章标签 python 统计 项目 推荐 键盘 监控 打字 频率

本文主要是介绍【python项目推荐】键盘监控--统计打字频率，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

原文：https://greptime.com/blogs/2024-03-19-keyboard-monitoring
代码：https://github.com/GreptimeTeam/demo-scene/tree/main/keyboard-monitor

项目简介

该项目实现了打字频率统计及可视化功能。
在这里插入图片描述

主要使用的库

pynput：允许您控制和监视输入设备。 这里我们用来获取键盘输入。
SQLAlchemy：数据库操作。 这里我们用来保存键盘输入。
streamlit：提供可视化界面。

项目组成

agent.py ：获得键盘输入
display.py：可视化

补充说明

如果你不想用原文的数据库，也可以替换为本地的数据库，如免安装的sqlite，

agent.py

# agent.py
from dotenv import load_dotenv
from pynput import keyboard
from pynput.keyboard import Keyimport concurrent.futures
import logging
import os
import queue
import sqlalchemy
import sqlalchemy.exc
import sys
import timeMODIFIERS = {Key.shift, Key.shift_l, Key.shift_r,Key.alt, Key.alt_l, Key.alt_r, Key.alt_gr,Key.ctrl, Key.ctrl_l, Key.ctrl_r,Key.cmd, Key.cmd_l, Key.cmd_r,
}TABLE = sqlalchemy.Table('keyboard_monitor',sqlalchemy.MetaData(),sqlalchemy.Column('hits', sqlalchemy.String),sqlalchemy.Column('ts', sqlalchemy.DateTime),
)if __name__ == '__main__':load_dotenv()log = logging.getLogger("agent")log.setLevel(logging.DEBUG)formatter = logging.Formatter('%(asctime)s %(levelname)s %(funcName)s %(message)s')file_handler = logging.FileHandler(f'agent-{time.time_ns()}.log', encoding='utf-8')file_handler.setLevel(logging.DEBUG)file_handler.setFormatter(formatter)stdout_handler = logging.StreamHandler(sys.stdout)stdout_handler.setLevel(logging.INFO)stdout_handler.setFormatter(formatter)log.addHandler(file_handler)log.addHandler(stdout_handler)#engine = sqlalchemy.create_engine(os.environ['DATABASE_URL'], #                                  echo_pool=True, #                                  isolation_level='AUTOCOMMIT')engine = sqlalchemy.create_engine("sqlite:///keyboard.db")current_modifiers = set()pending_hits = queue.Queue()cancel_signal = queue.Queue()def on_press(key):if key in MODIFIERS:current_modifiers.add(key)else:hits = sorted([ str(key) for key in current_modifiers ]) + [ str(key) ]hits = '+'.join(hits)pending_hits.put(hits)log.debug(f'{key} pressed, current_modifiers: {current_modifiers}')def on_release(key):if key in MODIFIERS:try:current_modifiers.remove(key)except KeyError:log.warning(f'Key {key} not in current_modifiers {current_modifiers}')log.debug(f'{key} released, current_modifiers: {current_modifiers}')#with engine.connect() as connection:#    connection.execute(sqlalchemy.sql.text("""#        CREATE TABLE IF NOT EXISTS keyboard_monitor (#            hits STRING NULL,#            ts TIMESTAMP(3) NOT NULL,#            TIME INDEX ("ts")#        ) ENGINE=mito WITH( regions = 1, ttl = '3months')#    """))# ...from sqlalchemy import create_engine, Table, Column, String, TIMESTAMP, MetaData, Indexmetadata = MetaData()keyboard_monitor = Table('keyboard_monitor', metadata,Column('hits', String, nullable=True),Column('ts', TIMESTAMP, nullable=False),)metadata.create_all(engine)def sender_thread():retries = 0while True:hits = pending_hits.get()log.debug(f'got: {hits}')if hits is None:log.info("Exiting...")breakwith engine.connect() as connection:try:log.debug(f'sending: {hits}')connection.execute(TABLE.insert().values(hits=hits, ts=sqlalchemy.func.now()))connection.commit()# ...log.info(f'sent: {hits}')retries = 0except sqlalchemy.exc.OperationalError as e:if retries >= 10:log.error(f'Retry exceeds. Operational error: {e}')pending_hits.put(hits)continueif e.connection_invalidated:log.warning(f'Connection invalidated: {e}')pending_hits.put(hits)continuemsg = str(e)if "(1815, 'Internal error: 1000')" in msg:# TODO 1815 - should not handle internal error;# see https://github.com/GreptimeTeam/greptimedb/issues/3447log.warning(f'Known operational error: {e}')pending_hits.put(hits)continueelif '2005' in msg and 'Unknown MySQL server host' in msg:log.warning(f'DNS temporary unresolved: {e}')pending_hits.put(hits)continueraise efinally:retries += 1def listener_thread():with keyboard.Listener(on_press=on_press, on_release=on_release) as listener:log.info("Listening...")cancel_signal.get()pending_hits.put(None)log.info("Exiting...")with concurrent.futures.ThreadPoolExecutor() as executor:sender = executor.submit(sender_thread)listener = executor.submit(listener_thread)try:f = concurrent.futures.wait([sender, listener], return_when=concurrent.futures.FIRST_EXCEPTION)for fut in f.done:log.error(f'Unhandled exception for futures: {fut.exception(timeout=0)}')except KeyboardInterrupt as e:log.info("KeyboardInterrupt. Exiting...")except Exception as e:log.error(f'Unhandled exception: {e}')finally:cancel_signal.put(True)

display.py

# display.py
import datetime
import os
from dotenv import load_dotenv
import pytz
import streamlit as st
import tzlocal
import pandasst.title("Keyboard Monitor")load_dotenv()
#conn = st.connection(
##    type="sql",
#    url="sqlite:///keyboard.db",
#)conn = st.connection('keyboard', type='sql', url="sqlite:///keyboard.db")df = conn.query("SELECT COUNT(*) AS total_hits FROM keyboard_monitor")
st.metric("Total hits", df.total_hits[0])most_frequent_key, most_frequent_combo = st.columns(2)
df = conn.query("""
SELECT hits, COUNT(*) as times
FROM keyboard_monitor
WHERE hits NOT LIKE '%+%'
GROUP BY hits
ORDER BY times DESC limit 1;
""")
most_frequent_key.metric("Most frequent key", df.hits[0])
df = conn.query("""
SELECT hits, COUNT(*) as times
FROM keyboard_monitor
WHERE hits LIKE '%+%'
GROUP BY hits
ORDER BY times DESC limit 1;
""")
most_frequent_combo.metric("Most frequent combo", df.hits[0])top_frequent_keys, top_frequent_combos = st.columns(2)
df = conn.query("""
SELECT hits, COUNT(*) as times
FROM keyboard_monitor
WHERE hits NOT LIKE '%+%'
GROUP BY hits
ORDER BY times DESC limit 10;
""")
top_frequent_keys.subheader("Top 10 keys")
top_frequent_keys.dataframe(df)
df = conn.query("""
SELECT hits, COUNT(*) as times
FROM keyboard_monitor
WHERE hits LIKE '%+%'
GROUP BY hits
ORDER BY times DESC limit 10;
""")
top_frequent_combos.subheader("Top 10 combos")
top_frequent_combos.dataframe(df)st.header("Find your inputs frequency of day")
local_tz = tzlocal.get_localzone()
hours = int(local_tz.utcoffset(datetime.datetime.now()).total_seconds() / 3600)
if hours > 0:offset = f" + INTERVAL '{hours} hours'"
elif hours < 0:offset = f" - INTERVAL '{hours} hours'"
else:offset = ''
d = st.date_input("Pick a day:", value=datetime.date.today())
query = f"""
SELECT ts,COUNT(1) AS times
FROM keyboard_monitor
WHERE strftime('%Y-%m-%d', ts, 'localtime') = '{d}'
GROUP BY strftime('%Y-%m-%d %H:00:00', ts)
ORDER BY ts ASC
LIMIT 10;
"""df = conn.query(query)
#print(df.keys())
df['ts'] = pandas.to_datetime(df['ts'])
df['ts'] = df['ts'].dt.tz_localize(pytz.utc).dt.tz_convert(local_tz)
st.dataframe(df)

这篇关于【python项目推荐】键盘监控--统计打字频率的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/926644。

相关文章

部署Vue项目到服务器后404错误的原因及解决方案

部署Vue项目到服务器后404错误的原因及解决方案

《部署Vue项目到服务器后404错误的原因及解决方案》文章介绍了Vue项目部署步骤以及404错误的解决方案,部署步骤包括构建项目、上传文件、配置Web服务器、重启Nginx和访问域名,404错误通常是... 目录一、vue项目部署步骤二、404错误原因及解决方案错误场景原因分析解决方案一、Vue项目部署步骤
阅读更多...
python使用fastapi实现多语言国际化的操作指南

python使用fastapi实现多语言国际化的操作指南

《python使用fastapi实现多语言国际化的操作指南》本文介绍了使用Python和FastAPI实现多语言国际化的操作指南,包括多语言架构技术栈、翻译管理、前端本地化、语言切换机制以及常见陷阱和... 目录多语言国际化实现指南项目多语言架构技术栈目录结构翻译工作流1. 翻译数据存储2. 翻译生成脚本
阅读更多...
如何通过Python实现一个消息队列

如何通过Python实现一个消息队列

《如何通过Python实现一个消息队列》这篇文章主要为大家详细介绍了如何通过Python实现一个简单的消息队列,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录如何通过 python 实现消息队列如何把 http 请求放在队列中执行1. 使用 queue.Queue 和 reque
阅读更多...
Python如何实现PDF隐私信息检测

Python如何实现PDF隐私信息检测

《Python如何实现PDF隐私信息检测》随着越来越多的个人信息以电子形式存储和传输,确保这些信息的安全至关重要,本文将介绍如何使用Python检测PDF文件中的隐私信息,需要的可以参考下... 目录项目背景技术栈代码解析功能说明运行结php果在当今，数据隐私保护变得尤为重要。随着越来越多的个人信息以电子形
阅读更多...
使用Python快速实现链接转word文档

使用Python快速实现链接转word文档

《使用Python快速实现链接转word文档》这篇文章主要为大家详细介绍了如何使用Python快速实现链接转word文档功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 演示代码展示from newspaper import Articlefrom docx import
阅读更多...
Python Jupyter Notebook导包报错问题及解决

Python Jupyter Notebook导包报错问题及解决

《PythonJupyterNotebook导包报错问题及解决》在conda环境中安装包后,JupyterNotebook导入时出现ImportError,可能是由于包版本不对应或版本太高,解决方... 目录问题解决方法重新安装Jupyter NoteBook 更改Kernel总结问题在conda上安装了
阅读更多...
golang内存对齐的项目实践

golang内存对齐的项目实践

《golang内存对齐的项目实践》本文主要介绍了golang内存对齐的项目实践,内存对齐不仅有助于提高内存访问效率,还确保了与硬件接口的兼容性,是Go语言编程中不可忽视的重要优化手段,下面就来介绍一下... 目录一、结构体中的字段顺序与内存对齐二、内存对齐的原理与规则三、调整结构体字段顺序优化内存对齐四、内
阅读更多...
Python如何计算两个不同类型列表的相似度

Python如何计算两个不同类型列表的相似度

《Python如何计算两个不同类型列表的相似度》在编程中,经常需要比较两个列表的相似度,尤其是当这两个列表包含不同类型的元素时,下面小编就来讲讲如何使用Python计算两个不同类型列表的相似度吧... 目录摘要引言数字类型相似度欧几里得距离曼哈顿距离字符串类型相似度Levenshtein距离Jaccard相
阅读更多...
Python安装时常见报错以及解决方案

Python安装时常见报错以及解决方案

《Python安装时常见报错以及解决方案》：本文主要介绍在安装Python、配置环境变量、使用pip以及运行Python脚本时常见的错误及其解决方案,文中介绍的非常详细,需要的朋友可以参考下... 目录一、安装 python 时常见报错及解决方案（一）安装包下载失败（二）权限不足二、配置环境变量时常见报错及
阅读更多...
Python中顺序结构和循环结构示例代码

Python中顺序结构和循环结构示例代码

《Python中顺序结构和循环结构示例代码》：本文主要介绍Python中的条件语句和循环语句,条件语句用于根据条件执行不同的代码块,循环语句用于重复执行一段代码,文章还详细说明了range函数的使... 目录一、条件语句（1）条件语句的定义（2）条件语句的语法（a）单分支 if（b）双分支 if-else（
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2