Skip to content

Instantly share code, notes, and snippets.

Last active April 14, 2022 11:43
What would you like to do?
import os
import stat
import time
import logging
import asyncio
from dataclasses import dataclass
from evdev import InputDevice, categorize, ecodes
dev = InputDevice('/dev/input/event2') # mouse0
@dataclass(init=True, repr=True)
class Stats:
mouse_x_cnt: int = 0
mouse_y_cnt: int = 0
mouse_clicks_left_cnt: int = 0
mouse_clicks_right_cnt: int = 0
def get_all(self):
return self.mouse_x_cnt, self.mouse_y_cnt, self.mouse_clicks_left_cnt, self.mouse_clicks_right_cnt
def changed(self):
return any([x > 0 for x in self.get_all()])
def reset(self):
self.mouse_x_cnt = 0
self.mouse_y_cnt = 0
self.mouse_clicks_left_cnt = 0
self.mouse_clicks_right_cnt = 0
def to_csv(self):
return ','.join([str(x) for x in [time.time(), *self.get_all()]]) + '\n'
def csv_header():
return ','.join(['time', 'mouse_x_cnt', 'mouse_y_cnt', 'mouse_clicks_left_cnt', 'mouse_clicks_right_cnt']) + '\n'
def dump(stats):
file_name = 'mouse_stats.csv'
if not os.path.exists(file_name):
with open(file_name, 'w') as f:
os.chmod(file_name, stat.S_IROTH)
if stats.changed():
with open(file_name, 'a') as f:
async def persist(stats):
while True:
await asyncio.sleep(60)
await asyncio.get_event_loop().run_in_executor(None, dump, stats)
async def main(dev, stats):
async for ev in dev.async_read_loop():
if ev.type == ecodes.EV_REL:
if ev.code == ecodes.REL_X:
stats.mouse_x_cnt += abs(ev.value)
if ev.value > 0: direction = 'right'
else: direction = 'left'
elif ev.code == ecodes.REL_Y:
stats.mouse_y_cnt += abs(ev.value)
if ev.value > 0: direction = 'down'
else: direction = 'up'
logging.debug(f'Mouse moved {ev.value} pixels {direction}')
elif ev.type == ecodes.EV_KEY:
if ev.value == 1:
if ev.code == ecodes.BTN_LEFT:
stats.mouse_clicks_left_cnt += 1
logging.debug('Left button clicked')
elif ev.code == ecodes.BTN_RIGHT:
stats.mouse_clicks_right_cnt += 1
logging.debug('Right button clicked')
if __name__ == '__main__':
loop = asyncio.get_event_loop()
stats = Stats()
loop.run_until_complete(main(dev, stats))
except KeyboardInterrupt:
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment