mouse_tracker.py

#!/usr/bin/python

# TODO: https://stackoverflow.com/a/58840987/3112139

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'
    
    @staticmethod
    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:
            f.write(stats.csv_header())
            os.chmod(file_name, stat.S_IROTH)

    if stats.changed():
        with open(file_name, 'a') as f:
            f.write(stats.to_csv())

    stats.reset()

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'
            else:
                continue
            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()
    try:
        loop.create_task(persist(stats))
        loop.run_until_complete(main(dev, stats))
    except KeyboardInterrupt:
        dump(stats)
        print('Finished')