Record typing statistics

typing_stat.py

import os
import sqlite3
import time
import traceback
from datetime import datetime
from functools import wraps

from pynput.keyboard import Listener

DB_PATH = "./typing_stats"


def db_name():
    return datetime.now().strftime("%Y%m%d")


def clear():
    if os.name == "nt":
        os.system("cls")
    else:
        os.system("clear")
    print("Recording typing statistics... Press Ctrl+C to stop.")


def ignore_errors(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print(f"Error during {' '.join(func.__name__.split('_'))}: {str(e)}")
            print(traceback.format_exc())

    return wrapper



@ignore_errors
def create_db_if_not_exists(db_name_):
    if not os.path.exists(DB_PATH):
        os.makedirs(DB_PATH)

    db_file = os.path.join(DB_PATH, f"{db_name_}.db")

    if not os.path.exists(db_file):
        connection = sqlite3.connect(db_file)
        cursor = connection.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS typing_stats (
                time_slot TEXT,
                characters_per_minute_average REAL
            )
        ''')
        connection.commit()
        connection.close()


def print_minute_stats(db_name_):
    connection = sqlite3.connect(os.path.join(DB_PATH, f"{db_name_}.db"))
    cursor = connection.cursor()
    cursor.execute('SELECT * FROM typing_stats ORDER BY time_slot DESC LIMIT 15')
    rows = cursor.fetchall()
    connection.close()

    clear()

    print(f"┌─────────────────────────┬──────────────────────────┐")
    print(f"│{' Typing statistics for ':^{45}}│{db_name_:^{43}}│")
    print(f"├─────────────────────────┼──────────────────────────┤")
    print(f"│{' Time slot ':^{45}}         │{' Characters per minute average ':^{43}}│")
    print(f"├─────────────────────────┼──────────────────────────┤")

    for row in rows:
        color = '\033[32m' if row[1] > 120 else '\033[31m' if row[1] < 40 else '\033[0m'
        print(f"│{row[0]:^{45}}│{color}{row[1]:.2f}\033[0m│")

    print(f"└─────────────────────────┴──────────────────────────┘")


def record_typing_statistics():
    start_time = time.time()
    typing_data_ = []
    characters_typed = 0

    def on_press(_):
        nonlocal characters_typed
        characters_typed += 1

    listener = Listener(on_press=on_press)
    listener.start()

    try:
        while True:
            time.sleep(1)
            time_elapsed = time.time() - start_time

            if time_elapsed >= 60:  # 1 minute
                characters_per_minute = (characters_typed / time_elapsed) * 60
                typing_data_.append(characters_per_minute)
                start_time = time.time()
                characters_typed = 0
                print(".", end="", flush=True)
            if len(typing_data_) >= 15:
                create_db_if_not_exists(db_name())
                write_to_database(db_name(), typing_data_)
                typing_data_ = []

    except KeyboardInterrupt:
        listener.stop()
        return typing_data_


@ignore_errors
def write_to_database(db_name_, typing_data_):
    if typing_data_:
        connection = sqlite3.connect(os.path.join(DB_PATH, f"{db_name_}.db"))
        cursor = connection.cursor()
        time_slot_end = int(time.time())
        time_slot_start = time_slot_end - 60 * len(typing_data_)
        time_slot = f"{datetime.fromtimestamp(time_slot_start).strftime('%H:%M')} - {datetime.fromtimestamp(time_slot_end).strftime('%H:%M')}"
        characters_per_minute_average = sum(typing_data_) / len(typing_data_)
        cursor.execute('INSERT INTO typing_stats VALUES (?, ?)',
                       (time_slot, characters_per_minute_average))
        connection.commit()
        connection.close()
    print_minute_stats(db_name_)

if __name__ == "__main__":
    clear()
    create_db_if_not_exists(db_name())
    typing_data = record_typing_statistics()
    create_db_if_not_exists(db_name())
    write_to_database(db_name(), typing_data)