diff --git a/.clang-format b/.clang-format old mode 100644 new mode 100755 diff --git a/.gitignore b/.gitignore old mode 100644 new mode 100755 diff --git a/LICENSE b/LICENSE old mode 100644 new mode 100755 diff --git a/Makefile b/Makefile old mode 100644 new mode 100755 index ed3631a..04aa400 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -all: tikker run +all: plot process tikker: tikker.c sormc.h gcc tikker.c -Ofast -Wall -Werror -Wextra -o tikker -lsqlite3 @@ -9,8 +9,8 @@ run: PYTHON="./.venv/bin/python" ensure_env: - -@python3 -m venv .venv - $(PYTHON) -m pip install dataset matplotlib + -@python3.12 -m venv .venv + $(PYTHON) -m pip install dataset matplotlib openai requests merge: $(PYTHON) merge.py diff --git a/README.md b/README.md old mode 100644 new mode 100755 diff --git a/merge.py b/merge.py old mode 100644 new mode 100755 diff --git a/plot.py b/plot.py old mode 100644 new mode 100755 index ae23227..0135ee8 --- a/plot.py +++ b/plot.py @@ -1,279 +1,470 @@ +#!/usr/bin/env python3 +""" +Keyboard Analytics - A tool for analyzing keyboard usage patterns + +This script analyzes keyboard events stored in a SQLite database and generates +visualizations and reports based on the data. It can track key presses across +different time periods and create meaningful insights about typing patterns. +""" + import sqlite3 import time -import matplotlib.pyplot as plt import pathlib +import json +import logging +import requests +from typing import List, Dict, Tuple, Any, Set +import matplotlib.pyplot as plt from xmlrpc.client import ServerProxy -api = ServerProxy("https://api.molodetz.nl/rpc") - -connection = sqlite3.connect('tikker.db') - - -weekday_sql = ( - "CASE " - "WHEN strftime('%w', timestamp) = '0' THEN 'Sunday' " - "WHEN strftime('%w', timestamp) = '1' THEN 'Monday' " - "WHEN strftime('%w', timestamp) = '2' THEN 'Tuesday' " - "WHEN strftime('%w', timestamp) = '3' THEN 'Wednesday' " - "WHEN strftime('%w', timestamp) = '4' THEN 'Thursday' " - "WHEN strftime('%w', timestamp) = '5' THEN 'Friday' " - "WHEN strftime('%w', timestamp) = '6' THEN 'Saturday' " - "END" +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) +logger = logging.getLogger(__name__) -def query(sql): +# Initialize API connection +API_ENDPOINT = "https://api.molodetz.nl/rpc" +api = ServerProxy(API_ENDPOINT) + +# Database connection +DB_PATH = 'tikker.db' +connection = sqlite3.connect(DB_PATH) + +# Track processed items to avoid duplicate work +processed_files: Set[str] = set() +processed_weeks: Set[str] = set() + +# SQL helper for weekday names +WEEKDAY_SQL = """ + CASE + WHEN strftime('%w', timestamp) = '0' THEN 'Sunday' + WHEN strftime('%w', timestamp) = '1' THEN 'Monday' + WHEN strftime('%w', timestamp) = '2' THEN 'Tuesday' + WHEN strftime('%w', timestamp) = '3' THEN 'Wednesday' + WHEN strftime('%w', timestamp) = '4' THEN 'Thursday' + WHEN strftime('%w', timestamp) = '5' THEN 'Friday' + WHEN strftime('%w', timestamp) = '6' THEN 'Saturday' + END +""" + +def query(sql: str) -> List[Tuple]: + """ + Execute an SQL query and return the results. + + Args: + sql: SQL query to execute + + Returns: + List of result tuples + """ start = time.time() - + cursor = connection.cursor() - print(sql) + logger.debug(f"Executing SQL: {sql}") result = cursor.execute(sql).fetchall() - + cursor.close() duration = time.time() - start - - print("Duration: {}\n".format(duration)) + + logger.info(f"Query completed in {duration:.4f} seconds") return result -def render_per_hour(week): - week = f"{week}" +def setup_plot_defaults() -> None: + """Configure default matplotlib settings for all plots""" + plt.style.use('dark_background') + plt.figure(figsize=(10, 7)) - sql_presses = ( - "SELECT count(0) as total, strftime('%H', timestamp) as hour, " - "strftime('%U', timestamp) as week " - "FROM kevent WHERE event = 'PRESSED' AND week = '{week}' " - "GROUP BY week, hour ORDER BY hour" - ) +def save_figure(filename: str) -> None: + """Save the current matplotlib figure to a file""" + plt.tight_layout() + plt.savefig(filename) + plt.close() + logger.info(f"Figure saved: {filename}") + +def render_per_hour(week: str) -> None: + """ + Generate visualization of key presses per hour for a specific week + + Args: + week: Week number to analyze + """ + # Skip if already processed + if week in processed_weeks: + logger.info(f"Week {week} already processed for hourly analysis, skipping") + return + + processed_weeks.add(week) + + sql_presses = """ + SELECT count(0) as total, strftime('%H', timestamp) as hour, + strftime('%U', timestamp) as week + FROM kevent + WHERE event = 'PRESSED' AND week = '{week}' + GROUP BY week, hour + ORDER BY hour + """ - #sql_repeated = ( - # "SELECT count(0) as total, strftime('%H', timestamp) as hour, " - # "strftime('%U', timestamp) as week " - # "FROM kevent WHERE event = 'REPEATED' AND week = {week} " - # "GROUP BY week, hour ORDER BY hour" - #) - - #sql_released = ( - # "SELECT count(0) as total, strftime('%H', timestamp) as hour, " - # "strftime('%U', timestamp) as week " - # "FROM kevent WHERE event = 'RELEASED' AND week = {week} " - # "GROUP BY week, hour ORDER BY hour" - #) - - rows_presses = query(sql_presses.format(week=week)) - #rows_repeated = query(sql_repeated.format(week=week)) - #rows_released = query(sql_released.format(week=week)) - + + if not rows_presses: + logger.warning(f"No data found for week {week}") + return + totals = [row[0] for row in rows_presses] hours = [row[1] for row in rows_presses] - - #totals_repeated = [row[0] for row in rows_repeated] - #hours_repeated = [row[1] for row in rows_repeated] - - #totals_released = [row[0] for row in rows_released] - #hours_released = [row[1] for row in rows_released] - - plt.figure(figsize=(8, 6)) - #plt.plot(hours_repeated, totals_repeated, marker='o', label='Repeats per hour', color='green') - #plt.plot(hours_released, totals_released, marker='o', label='Releases per hour', color='orange') - plt.plot(hours, totals, marker='o', label=f'Presses per hour week {week}', color='red') - + + setup_plot_defaults() + plt.plot(hours, totals, marker='o', label=f'Presses per hour', color='red') + plt.xlabel('Hour') plt.ylabel('Event count') - plt.title(f'Key presses per hour. Week {week}') - plt.style.use('dark_background') + plt.title(f'Key presses per hour - Week {week}') plt.legend() + + save_figure(f"graph_week_{week.strip('\'')}_per_hour.png") - plt.savefig(f"graph_week_{week.strip('\'')}_per_hour.png") - -def render_per_day(): - sql_pressed_per_day = ( - "SELECT strftime('%Y-%m-%d', timestamp) as month_day,count(0) as total FROM kevent WHERE event = 'PRESSED' GROUP BY month_day ORDER BY month_day" - ) - plt.figure(figsize=(8,6)) - +def render_per_day() -> None: + """Generate visualization of key presses per day""" + sql_pressed_per_day = """ + SELECT strftime('%Y-%m-%d', timestamp) as month_day, + count(0) as total + FROM kevent + WHERE event = 'PRESSED' + GROUP BY month_day + ORDER BY month_day + """ + rows_pressed_per_day = query(sql_pressed_per_day) - - totals = [row[0] for row in rows_pressed_per_day] - dates = [row[1] for row in rows_pressed_per_day] - - plt.plot(totals, dates, marker='o', label='Presses per day', color='red') - + + if not rows_pressed_per_day: + logger.warning("No data found for daily analysis") + return + + dates = [row[0] for row in rows_pressed_per_day] + totals = [row[1] for row in rows_pressed_per_day] + + setup_plot_defaults() + plt.plot(dates, totals, marker='o', label='Presses per day', color='red') + plt.xlabel('Date') plt.ylabel('Event count') - plt.xticks(rotation=45) - plt.style.use('dark_background') - plt.title('Keyboard events') - plt.tight_layout() + plt.xticks(rotation=45) + plt.title('Keyboard events by day') plt.legend() - plt.savefig(f"graph_per_day.png") - -def render_per_week(): - sql_pressed_per_day = ( - f"SELECT strftime('%Y-%U', timestamp) as week,count(0) as total FROM kevent WHERE event = 'PRESSED' GROUP BY week ORDER BY week" - ) - plt.figure(figsize=(8,6)) - - rows_pressed_per_day = query(sql_pressed_per_day) - - totals = [row[0] for row in rows_pressed_per_day] - dates = [row[1] for row in rows_pressed_per_day] - - plt.plot(totals, dates, marker='o', label='Presses per day', color='red') + + save_figure("graph_per_day.png") +def render_per_week() -> None: + """Generate visualization of key presses per week""" + sql_pressed_per_week = """ + SELECT strftime('%Y-%U', timestamp) as week, + count(0) as total + FROM kevent + WHERE event = 'PRESSED' + GROUP BY week + ORDER BY week + """ + + rows_pressed_per_week = query(sql_pressed_per_week) + + if not rows_pressed_per_week: + logger.warning("No data found for weekly analysis") + return + + weeks = [row[0] for row in rows_pressed_per_week] + totals = [row[1] for row in rows_pressed_per_week] + + setup_plot_defaults() + plt.plot(weeks, totals, marker='o', label='Presses per week', color='red') + plt.xlabel('Week') plt.ylabel('Presses count') - plt.xticks(rotation=45) - plt.title(f'Presses per week') - plt.tight_layout() - plt.style.use('dark_background') + plt.xticks(rotation=45) + plt.title('Presses per week') plt.legend() + + save_figure("graph_per_week.png") - plt.savefig(f"graph_per_week.png") - - - - -def render_per_weekday(week): - - sql_presses = ( - f"SELECT count(0) as total, {weekday_sql} as weekday, " - "strftime('%w', timestamp) as day, strftime('%U', timestamp) as week " - "FROM kevent WHERE event = 'PRESSED' AND week = '{week}' " - "GROUP BY week, day ORDER BY day" - ) - - sql_repeated = ( - f"SELECT count(0) as total, {weekday_sql} as weekday, " - "strftime('%w', timestamp) as day, strftime('%U', timestamp) as week " - "FROM kevent WHERE event = 'REPEATED' AND week = '{week}' " - "GROUP BY week, day ORDER BY day" - ) - - sql_released = ( - f"SELECT count(0) as total, {weekday_sql} as weekday, " - "strftime('%w', timestamp) as day, strftime('%U', timestamp) as week " - "FROM kevent WHERE event = 'RELEASED' AND week = '{week}' " - "GROUP BY week, day ORDER BY day" - ) - - rows_presses = query(sql_presses.format(week=week)) - #rows_repeated = query(sql_repeated.format(week=week)) - #rows_released = query(sql_released.format(week=week)) - +def render_per_weekday(week: str) -> None: + """ + Generate visualization of key presses per weekday for a specific week + + Args: + week: Week number to analyze + """ + # Skip if already processed + if week in processed_weeks: + logger.info(f"Week {week} already processed for weekday analysis, skipping") + return + + processed_weeks.add(week) + + sql_presses = f""" + SELECT count(0) as total, {WEEKDAY_SQL} as weekday, + strftime('%w', timestamp) as day, strftime('%U', timestamp) as week + FROM kevent + WHERE event = 'PRESSED' AND week = '{week}' + GROUP BY week, day + ORDER BY day + """ + + rows_presses = query(sql_presses) + + if not rows_presses: + logger.warning(f"No data found for week {week} weekday analysis") + return + totals = [row[0] for row in rows_presses] days = [row[2] for row in rows_presses] - - #totals_repeated = [row[0] for row in rows_repeated] - #days_repeated = [row[2] for row in rows_repeated] - - #totals_released = [row[0] for row in rows_released] - #days_released = [row[2] for row in rows_released] - - plt.figure(figsize=(8, 6)) - #plt.plot(days_repeated, totals_repeated, marker='o', label='Repeats per weekday', color='green') - #plt.plot(days_released, totals_released, marker='o', label='Releases per weekday', color='orange') - plt.plot(days, totals, marker='o', label=f'Press count', color='red') - - plt.xlabel('Weekday (0 = Sunday, 6 = Saturday)') + weekday_names = [row[1] for row in rows_presses] + + setup_plot_defaults() + plt.plot(days, totals, marker='o', label='Press count', color='red') + + plt.xlabel('Weekday') plt.ylabel('Event count') - plt.title(f'Presses per weekday. Week {week}') - plt.style.use('dark_background') + plt.title(f'Presses per weekday - Week {week}') + plt.xticks(range(len(weekday_names)), weekday_names, rotation=45) plt.legend() + + save_figure(f"graph_week_{week.strip('\"')}_per_weekday.png") - plt.savefig(f"graph_week_{week.strip('\"')}_per_weekday.png") - -def get_weeks(): - sql = "SELECT strftime('%U', timestamp) as week FROM kevent GROUP BY week" +def get_weeks() -> List[str]: + """ + Get list of all weeks in the database + + Returns: + List of week numbers + """ + sql = "SELECT DISTINCT strftime('%U', timestamp) as week FROM kevent GROUP BY week" weeks = query(sql) return [record[0] for record in weeks] -def get_score_per_week(): - sql = ( - "SELECT strftime('%U', timestamp) as week, event, COUNT(0) as total " - "FROM kevent GROUP BY event, week" - ) - return query(sql) - -def get_score_per_day(): +def get_score_per_week() -> List[Tuple]: + """ + Get event counts grouped by week - sql ="SELECT count(0) as total, CASE WHEN strftime('%w', timestamp) = 0 THEN 'Sunday' WHEN strftime('%w', timestamp) = 1 THEN 'Monday' WHEN strftime('%w', timestamp) = 2 THEN 'Tuesday' WHEN strftime('%w', timestamp) = 3 THEN 'Wednesday' WHEN strftime('%w', timestamp) = 4 THEN 'Thursday' WHEN strftime('%w', timestamp) = 5 THEN 'Friday' WHEN strftime('%w', timestamp) = 6 THEN 'Saturday' END as weekday, strftime('%w', timestamp) as day, strftime('%U', timestamp) as week FROM kevent WHERE event = 'REPEATED' GROUP BY week, day ORDER BY day" - - sql = ( - f"SELECT strftime('%U',timestamp) as week, {weekday_sql} as wday, event, COUNT(0) as total " - f"FROM kevent WHERE event in ('PRESSED') GROUP BY week, event, wday ORDER BY week, event, wday" - ) + Returns: + List of (week, event_type, count) tuples + """ + sql = """ + SELECT strftime('%U', timestamp) as week, event, COUNT(0) as total + FROM kevent + GROUP BY event, week + """ return query(sql) -def get_totals(): - sql = "SELECT count(0) as total, event from kevent group by event" - return query(sql) - -# Main execution -if __name__ == "__main__": - time_start = time.time() - - render_per_day() - render_per_week() - for week in get_weeks(): - render_per_hour(week) - render_per_weekday(week) - - print("Score per week:") - for record in get_score_per_week(): - print(f"{record[0]} \t{record[1]} \t{record[2]}") - - print("Score per day:") - for record in get_score_per_day(): - print(f"{record[0]} \t{record[1]} \t{record[2]}") +def get_score_per_day() -> List[Tuple]: + """ + Get event counts grouped by day of week - print("Total events:") - totals = 0 - for record in get_totals(): - print(f"{record[1]}: {record[0]}") - totals += record[0] - print(totals) + Returns: + List of (week, weekday, event_type, count) tuples + """ + sql = f""" + SELECT strftime('%U', timestamp) as week, + {WEEKDAY_SQL} as wday, + event, COUNT(0) as total + FROM kevent + WHERE event in ('PRESSED') + GROUP BY week, event, wday + ORDER BY week, event, wday + """ + return query(sql) +def get_totals() -> List[Tuple]: + """ + Get total count of each event type + + Returns: + List of (count, event_type) tuples + """ + sql = "SELECT count(0) as total, event FROM kevent GROUP BY event" + return query(sql) + +def generate_keylog() -> Dict[str, str]: + """ + Generate a log of key presses grouped by date and hour + + Returns: + Dictionary of date-hour to concatenated key presses + """ result = {} - rows = query("SElECT strftime('%Y-%m-%d.%H', timestamp) as date_hour, GROUP_CONCAT(char,'') FROM kevent WHERE event = 'PRESSED' group by date_hour") + rows = query(""" + SELECT strftime('%Y-%m-%d.%H', timestamp) as date_hour, + GROUP_CONCAT(char,'') + FROM kevent + WHERE event = 'PRESSED' + GROUP BY date_hour + """) + for row in rows: result[row[0]] = row[1] - with open("keylog.txt","w") as f: - for day in result.keys(): - date, hour = day.split(".") + return result + +def write_keylog_files(keylog: Dict[str, str]) -> None: + """ + Write keylog data to files + + Args: + keylog: Dictionary of date-hour to concatenated key presses + """ + logs_dir = pathlib.Path("logs_plain") + logs_dir.mkdir(exist_ok=True) + + with open("keylog.txt", "w") as f: + for day in keylog.keys(): + date, hour = day.split(".") label = f"{date} {hour}:00" - if not pathlib.Path("logs_plain/"+day+".txt").exists(): - with open("logs_plain/"+day+".txt","w") as g: - - - g.write(f"**{label}**: ```{result[day]}```\n\n") - f.write(f"**{label}**: ```{result[day]}```\n\n") - - print("Duration: {}".format(time.time() - time_start)) - exit() - import json - for file in pathlib.Path(".").glob("logs_plain/*.txt"): - print("Working on: {}".format(file)) - dest_file = file.parent.parent.joinpath("logs_summaries").joinpath(file.name) - print("Dest file: ", dest_file) - if dest_file.exists(): - continue - with dest_file.open("w+") as f: - print("Requesting...") - param = file.read_text().replace("@","").replace("`","") - response = api.gpt4o_mini("The following data is key presses made by user. Describe what user could be working on using bulletpoints: "+param) - print("Done") - f.write(response) - print(response) - for file in pathlib.Path(".").glob("logs_summaries/*.txt"): - dest_file = file.parent.parent.joinpath("logs_lines").joinpath(file.name) - if dest_file.exists(): - print("One liner already exists for" + file.name) - continue - with dest_file.open("w+") as f: - source = file.read_text().replace("@","").replace("`","") - response = api.gpt4o_mini("The following data is a hour of work summarized from the user. Describe what user was doing in a onliner.: "+source) - f.write(response) - print("Made one liner for" + file.name) - print("Duration: {}".format(time.time() - time_start)) + log_file = logs_dir / f"{day}.txt" + if not log_file.exists(): + with open(log_file, "w") as g: + g.write(f"**{label}**: ```{keylog[day]}```\n\n") + + f.write(f"**{label}**: ```{keylog[day]}```\n\n") + +def ipa(prompt): + import requests + result = requests.post("https://retoor:retoorded@ipa.molodetz.nl/ai/prompt",json={"prompt": prompt, "model": "google/gemma-3-12b-it","json":False}).text + print(result) + return result + +def generate_summaries(dry_run: bool=False) -> None: + """Generate summaries for keylog files using AI API""" + logs_dir = pathlib.Path("logs_plain") + summary_dir = pathlib.Path("logs_summaries") + oneliner_dir = pathlib.Path("logs_lines") + + summary_dir.mkdir(exist_ok=True) + oneliner_dir.mkdir(exist_ok=True) + + # Process summaries + for file in logs_dir.glob("*.txt"): + # Skip if already processed + if str(file) in processed_files: + logger.info(f"File {file} already processed for summary, skipping") + continue + + processed_files.add(str(file)) + + dest_file = summary_dir / file.name + if dest_file.exists(): + logger.info(f"Summary already exists for {file.name}, skipping") + continue + + try: + logger.info(f"Generating summary for {file.name}") + if dry_run: + continue + param = file.read_text().replace("@", "").replace("`", "") + prompt = "The following data is key presses made by user. Describe what user could be working on using bulletpoints: " + param + response = ipa(prompt) + + with dest_file.open("w+") as f: + f.write(response) + + logger.info(f"Summary generated for {file.name}") + except Exception as e: + logger.error(f"Error generating summary for {file.name}: {e}") + + # Process one-liners + for file in summary_dir.glob("*.txt"): + # Skip if already processed + if str(file) in processed_files: + logger.info(f"File {file} already processed for one-liner, skipping") + continue + + processed_files.add(str(file)) + + dest_file = oneliner_dir / file.name + if dest_file.exists(): + logger.info(f"One-liner already exists for {file.name}, skipping") + continue + + try: + logger.info(f"Generating one-liner for {file.name}") + if dry_run: + continue + source = file.read_text().replace("@", "").replace("`", "") + prompt = "The following data is a hour of work summarized from the user. Describe what user was doing in a oneliner: " + source + response = ipa(prompt) + + with dest_file.open("w+") as f: + f.write(response) + + logger.info(f"One-liner generated for {file.name}") + except Exception as e: + logger.error(f"Error generating one-liner for {file.name}: {e}") + +def main() -> None: + + # Generate summaries + generate_summaries(False) + """Main function to execute all analytics tasks""" + time_start = time.time() + logger.info("Starting keyboard analytics process") + + # Load state if exists + state_file = pathlib.Path("analytics_state.json") + if state_file.exists(): + try: + state = json.loads(state_file.read_text()) + processed_files.update(state.get("processed_files", [])) + processed_weeks.update(state.get("processed_weeks", [])) + logger.info(f"Loaded state: {len(processed_files)} files and {len(processed_weeks)} weeks processed previously") + except Exception as e: + logger.error(f"Error loading state: {e}") + + # Generate visualizations + render_per_day() + render_per_week() + + weeks = get_weeks() + for week in weeks: + render_per_hour(week) + render_per_weekday(week) + + # Print statistics + logger.info("Score per week:") + for record in get_score_per_week(): + logger.info(f"{record[0]}\t{record[1]}\t{record[2]}") + + logger.info("Score per day:") + for record in get_score_per_day(): + logger.info(f"{record[0]}\t{record[1]}\t{record[2]}\t{record[3]}") + + logger.info("Total events:") + totals = 0 + for record in get_totals(): + logger.info(f"{record[1]}: {record[0]}") + totals += record[0] + logger.info(f"Total: {totals}") + + # Generate and write keylog + keylog = generate_keylog() + write_keylog_files(keylog) + + # Generate summaries + generate_summaries() + + # Save state + try: + state = { + "processed_files": list(processed_files), + "processed_weeks": list(processed_weeks), + "last_run": time.time() + } + state_file.write_text(json.dumps(state)) + logger.info("State saved successfully") + except Exception as e: + logger.error(f"Error saving state: {e}") + + duration = time.time() - time_start + logger.info(f"Process completed in {duration:.2f} seconds") + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt old mode 100644 new mode 100755 index f0b2593..1e80b22 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ matplotlib - +openai +requests diff --git a/review.md b/review.md old mode 100644 new mode 100755 diff --git a/sormc.h b/sormc.h old mode 100644 new mode 100755 diff --git a/tags.py b/tags.py old mode 100644 new mode 100755 diff --git a/tikker.c b/tikker.c old mode 100644 new mode 100755 diff --git a/tikker.c.md b/tikker.c.md old mode 100644 new mode 100755 diff --git a/tikker_viz.py b/tikker_viz.py new file mode 100644 index 0000000..7db5504 --- /dev/null +++ b/tikker_viz.py @@ -0,0 +1,1496 @@ +import sqlite3 +import json +from datetime import datetime, timedelta +from collections import defaultdict, Counter +import sys +import time +import logging + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') + +class TikkerVisualizer: + def __init__(self, database_path='tikker.db'): + self.database_path = database_path + self.connection = sqlite3.connect(database_path) + self.cursor = self.connection.cursor() + + def get_total_events(self): + start_time = time.time() + logging.info("Retrieving total events...") + self.cursor.execute("SELECT COUNT(*) FROM kevent") + result = self.cursor.fetchone()[0] + duration = time.time() - start_time + logging.info(f"Total events retrieved in {duration:.2f} seconds") + return result, duration + + def get_pressed_events(self): + start_time = time.time() + logging.info("Retrieving pressed events...") + self.cursor.execute("SELECT COUNT(*) FROM kevent WHERE event='PRESSED'") + result = self.cursor.fetchone()[0] + duration = time.time() - start_time + logging.info(f"Pressed events retrieved in {duration:.2f} seconds") + return result, duration + + def get_released_events(self): + start_time = time.time() + logging.info("Retrieving released events...") + self.cursor.execute("SELECT COUNT(*) FROM kevent WHERE event='RELEASED'") + result = self.cursor.fetchone()[0] + duration = time.time() - start_time + logging.info(f"Released events retrieved in {duration:.2f} seconds") + return result, duration + + def get_date_range(self): + start_time = time.time() + logging.info("Retrieving date range...") + self.cursor.execute("SELECT MIN(timestamp), MAX(timestamp) FROM kevent") + result = self.cursor.fetchone() + duration = time.time() - start_time + logging.info(f"Date range retrieved in {duration:.2f} seconds") + return result[0], result[1], duration + + def get_hourly_activity(self, limit=1000): + start_time = time.time() + logging.info("Retrieving hourly activity...") + self.cursor.execute(""" + SELECT strftime('%H', timestamp) as hour, COUNT(*) as count + FROM kevent + WHERE event='PRESSED' + GROUP BY hour + ORDER BY hour + """) + result = self.cursor.fetchall() + duration = time.time() - start_time + logging.info(f"Hourly activity retrieved in {duration:.2f} seconds") + return result, duration + + def get_daily_activity(self, limit=365): + start_time = time.time() + logging.info("Retrieving daily activity...") + self.cursor.execute(f""" + SELECT strftime('%Y-%m-%d', timestamp) as day, COUNT(*) as count + FROM kevent + WHERE event='PRESSED' + GROUP BY day + ORDER BY day DESC + LIMIT {limit} + """) + result = list(reversed(self.cursor.fetchall())) + duration = time.time() - start_time + logging.info(f"Daily activity retrieved in {duration:.2f} seconds") + return result, duration + + def get_weekday_activity(self): + start_time = time.time() + logging.info("Retrieving weekday activity...") + self.cursor.execute(""" + SELECT + CASE CAST(strftime('%w', timestamp) AS INTEGER) + WHEN 0 THEN 'Sunday' + WHEN 1 THEN 'Monday' + WHEN 2 THEN 'Tuesday' + WHEN 3 THEN 'Wednesday' + WHEN 4 THEN 'Thursday' + WHEN 5 THEN 'Friday' + WHEN 6 THEN 'Saturday' + END as weekday, + COUNT(*) as count + FROM kevent + WHERE event='PRESSED' + GROUP BY strftime('%w', timestamp) + ORDER BY strftime('%w', timestamp) + """) + result = self.cursor.fetchall() + duration = time.time() - start_time + logging.info(f"Weekday activity retrieved in {duration:.2f} seconds") + return result, duration + + def get_top_keys(self, limit=30): + start_time = time.time() + logging.info("Retrieving top keys...") + self.cursor.execute(f""" + SELECT char, COUNT(*) as count + FROM kevent + WHERE event='PRESSED' AND char IS NOT NULL AND char != '' + GROUP BY char + ORDER BY count DESC + LIMIT {limit} + """) + result = self.cursor.fetchall() + duration = time.time() - start_time + logging.info(f"Top keys retrieved in {duration:.2f} seconds") + return result, duration + + def get_keyboard_heatmap_data(self): + start_time = time.time() + logging.info("Retrieving keyboard heatmap data...") + self.cursor.execute(""" + SELECT char, COUNT(*) as count + FROM kevent + WHERE event='PRESSED' AND char IS NOT NULL AND char != '' + GROUP BY char + ORDER BY count DESC + """) + data = {} + for char, count in self.cursor.fetchall(): + data[char] = count + duration = time.time() - start_time + logging.info(f"Keyboard heatmap data retrieved in {duration:.2f} seconds") + return data, duration + + def get_hour_day_heatmap(self, days_back=30): + start_time = time.time() + logging.info("Retrieving hour-day heatmap...") + self.cursor.execute(f""" + SELECT + strftime('%Y-%m-%d', timestamp) as day, + strftime('%H', timestamp) as hour, + COUNT(*) as count + FROM kevent + WHERE event='PRESSED' + AND timestamp >= datetime('now', '-{days_back} days') + GROUP BY day, hour + ORDER BY day, hour + """) + result = self.cursor.fetchall() + duration = time.time() - start_time + logging.info(f"Hour-day heatmap retrieved in {duration:.2f} seconds") + return result, duration + + def get_typing_speed_data(self, sample_size=10000): + start_time = time.time() + logging.info("Retrieving typing speed data...") + self.cursor.execute(f""" + SELECT + strftime('%Y-%m-%d %H:00:00', timestamp) as hour_bucket, + COUNT(*) as keypresses + FROM kevent + WHERE event='PRESSED' + GROUP BY hour_bucket + ORDER BY hour_bucket DESC + LIMIT {sample_size} + """) + result = list(reversed(self.cursor.fetchall())) + duration = time.time() - start_time + logging.info(f"Typing speed data retrieved in {duration:.2f} seconds") + return result, duration + + def get_character_frequency(self): + start_time = time.time() + logging.info("Retrieving character frequency...") + letters = Counter() + numbers = Counter() + special = Counter() + + self.cursor.execute(""" + SELECT char, COUNT(*) as count + FROM kevent + WHERE event='PRESSED' AND char IS NOT NULL AND char != '' + GROUP BY char + """) + + for char, count in self.cursor.fetchall(): + if len(char) == 1: + if char.isalpha(): + letters[char.lower()] += count + elif char.isdigit(): + numbers[char] += count + else: + special[char] += count + + result = { + 'letters': dict(letters.most_common(26)), + 'numbers': dict(numbers.most_common(10)), + 'special': dict(special.most_common(20)) + } + duration = time.time() - start_time + logging.info(f"Character frequency retrieved in {duration:.2f} seconds") + return result, duration + + def get_monthly_stats(self): + start_time = time.time() + logging.info("Retrieving monthly stats...") + self.cursor.execute(""" + SELECT + strftime('%Y-%m', timestamp) as month, + COUNT(*) as count + FROM kevent + WHERE event='PRESSED' + GROUP BY month + ORDER BY month + """) + result = self.cursor.fetchall() + duration = time.time() - start_time + logging.info(f"Monthly stats retrieved in {duration:.2f} seconds") + return result, duration + + def get_peak_activity_times(self): + start_time = time.time() + logging.info("Retrieving peak activity times...") + self.cursor.execute(""" + SELECT + strftime('%Y-%m-%d %H:00:00', timestamp) as hour_block, + COUNT(*) as count + FROM kevent + WHERE event='PRESSED' + GROUP BY hour_block + ORDER BY count DESC + LIMIT 10 + """) + result = self.cursor.fetchall() + duration = time.time() - start_time + logging.info(f"Peak activity times retrieved in {duration:.2f} seconds") + return result, duration + + # NEW: Words Per Minute (WPM) and Performance Metrics + def get_wpm_stats(self): + start_time = time.time() + logging.info("Calculating WPM and performance metrics...") + + # Get total key presses and time span + self.cursor.execute("SELECT COUNT(*) FROM kevent WHERE event='PRESSED'") + total_presses = self.cursor.fetchone()[0] + + self.cursor.execute("SELECT MIN(timestamp), MAX(timestamp) FROM kevent WHERE event='PRESSED'") + min_time, max_time = self.cursor.fetchone() + + if not min_time or not max_time or total_presses == 0: + return { + 'wpm_average': 0, + 'wpm_peak_hour': 0, + 'wpm_peak_day': 0, + 'words_typed': 0, + 'characters_typed': 0, + 'total_typing_time_minutes': 0, + 'typing_efficiency': 0, + 'peak_hour': None, + 'peak_day': None + }, 0.0 + + min_dt = datetime.fromisoformat(min_time) + max_dt = datetime.fromisoformat(max_time) + total_minutes = (max_dt - min_dt).total_seconds() / 60.0 + + # Estimate words: 5 characters per word (standard WPM formula) + characters_typed = total_presses + words_typed = characters_typed / 5.0 + wpm_average = words_typed / total_minutes if total_minutes > 0 else 0 + + # Peak hour WPM + self.cursor.execute(""" + SELECT strftime('%Y-%m-%d %H:00:00', timestamp) as hour, COUNT(*) as presses + FROM kevent WHERE event='PRESSED' + GROUP BY hour + ORDER BY presses DESC + LIMIT 1 + """) + peak_hour_row = self.cursor.fetchone() + peak_hour_wpm = (peak_hour_row[1] / 5.0) if peak_hour_row else 0 + + # Peak day WPM + self.cursor.execute(""" + SELECT strftime('%Y-%m-%d', timestamp) as day, COUNT(*) as presses + FROM kevent WHERE event='PRESSED' + GROUP BY day + ORDER BY presses DESC + LIMIT 1 + """) + peak_day_row = self.cursor.fetchone() + peak_day_wpm = (peak_day_row[1] / 5.0 / 60.0) if peak_day_row else 0 # per minute + + # Typing efficiency: presses per minute during active time + # Approximate active time as sum of intervals with activity + self.cursor.execute(""" + SELECT COUNT(DISTINCT strftime('%Y-%m-%d %H', timestamp)) as active_hours + FROM kevent WHERE event='PRESSED' + """) + active_hours = self.cursor.fetchone()[0] + typing_efficiency = total_presses / (active_hours * 60) if active_hours > 0 else 0 + + result = { + 'wpm_average': round(wpm_average, 2), + 'wpm_peak_hour': round(peak_hour_wpm, 2), + 'wpm_peak_day': round(peak_day_wpm, 2), + 'words_typed': int(words_typed), + 'characters_typed': characters_typed, + 'total_typing_time_minutes': round(total_minutes, 1), + 'typing_efficiency': round(typing_efficiency, 2), + 'peak_hour': peak_hour_row[0] if peak_hour_row else None, + 'peak_day': peak_day_row[0] if peak_day_row else None + } + + duration = time.time() - start_time + logging.info(f"WPM stats calculated in {duration:.2f} seconds") + return result, duration + + def get_performance_insights(self): + start_time = time.time() + logging.info("Generating performance insights...") + + # === 1. Key type distribution and corrections === + self.cursor.execute(""" + SELECT + SUM(CASE WHEN char IS NOT NULL AND length(char) = 1 AND char GLOB '[a-zA-Z]' THEN 1 ELSE 0 END) as letters, + SUM(CASE WHEN char IS NOT NULL AND length(char) = 1 AND char GLOB '[0-9]' THEN 1 ELSE 0 END) as numbers, + SUM(CASE WHEN char IS NOT NULL AND length(char) = 1 AND char NOT GLOB '[a-zA-Z0-9]' THEN 1 ELSE 0 END) as special, + SUM(CASE WHEN char = ' ' THEN 1 ELSE 0 END) as spaces, + SUM(CASE WHEN char = 'Backspace' THEN 1 ELSE 0 END) as backspaces, + SUM(CASE WHEN char = 'Delete' THEN 1 ELSE 0 END) as deletes + FROM kevent WHERE event='PRESSED' + """) + row = self.cursor.fetchone() + letters = row[0] or 0 + numbers = row[1] or 0 + special = row[2] or 0 + spaces = row[3] or 0 + backspaces = row[4] or 0 + deletes = row[5] or 0 + + total_edits = backspaces + deletes + valid_keystrokes = letters + numbers + special + spaces + error_rate = (total_edits / valid_keystrokes * 100) if valid_keystrokes > 0 else 0 + + # === 2. Daily activity consistency (manual STDDEV) === + self.cursor.execute(""" + SELECT COUNT(*) as count + FROM kevent WHERE event='PRESSED' + GROUP BY strftime('%Y-%m-%d', timestamp) + """) + daily_counts = [row[0] for row in self.cursor.fetchall()] + + if daily_counts: + avg_daily = sum(daily_counts) / len(daily_counts) + variance = sum((x - avg_daily) ** 2 for x in daily_counts) / len(daily_counts) + std_daily = variance ** 0.5 + consistency_score = 100 - (std_daily / avg_daily * 100) if avg_daily > 0 else 0 + consistency_score = max(0, min(100, consistency_score)) + else: + avg_daily = std_daily = 0 + consistency_score = 0 + + result = { + 'error_rate_percent': round(error_rate, 2), + 'backspace_count': backspaces, + 'delete_count': deletes, + 'total_corrections': total_edits, + 'consistency_score': round(consistency_score, 1), + 'letters_typed': letters, + 'numbers_typed': numbers, + 'special_typed': special, + 'spaces_typed': spaces + } + + duration = time.time() - start_time + logging.info(f"Performance insights generated in {duration:.2f} seconds") + return result, duration + + def close(self): + self.connection.close() + + def generate_html(self, output_file='tikker_report.html'): + logging.info("Starting HTML generation...") + overall_start = time.time() + + total_events, total_events_duration = self.get_total_events() + pressed_events, pressed_events_duration = self.get_pressed_events() + released_events, released_events_duration = self.get_released_events() + date_range_min, date_range_max, date_range_duration = self.get_date_range() + date_range = (date_range_min, date_range_max) + + hourly_data, hourly_duration = self.get_hourly_activity() + daily_data, daily_duration = self.get_daily_activity() + weekday_data, weekday_duration = self.get_weekday_activity() + top_keys, top_keys_duration = self.get_top_keys() + keyboard_heatmap, keyboard_heatmap_duration = self.get_keyboard_heatmap_data() + hour_day_heatmap, hour_day_heatmap_duration = self.get_hour_day_heatmap() + typing_speed, typing_speed_duration = self.get_typing_speed_data() + char_frequency, char_frequency_duration = self.get_character_frequency() + monthly_stats, monthly_stats_duration = self.get_monthly_stats() + peak_times, peak_times_duration = self.get_peak_activity_times() + wpm_stats, wpm_duration = self.get_wpm_stats() + perf_insights, perf_duration = self.get_performance_insights() + + durations = { + 'total_events': total_events_duration, + 'pressed_events': pressed_events_duration, + 'released_events': released_events_duration, + 'date_range': date_range_duration, + 'hourly_activity': hourly_duration, + 'daily_activity': daily_duration, + 'weekday_activity': weekday_duration, + 'top_keys': top_keys_duration, + 'keyboard_heatmap': keyboard_heatmap_duration, + 'hour_day_heatmap': hour_day_heatmap_duration, + 'typing_speed': typing_speed_duration, + 'character_frequency': char_frequency_duration, + 'monthly_stats': monthly_stats_duration, + 'peak_activity_times': peak_times_duration, + 'wpm_calculation': wpm_duration, + 'performance_insights': perf_duration + } + + logging.info("Generating HTML content...") + html = f""" + + + + + Tikker Keyboard Performance Report + + + + +
+
+

Tikker Keyboard Performance Report

+

Comprehensive Analysis of Typing Speed, Accuracy, and Productivity Metrics

+
+ +
+
+

Total Events

+
{total_events:,}
+
Recorded Actions
+
+
+

Key Presses

+
{pressed_events:,}
+
Keys Pressed
+
+
+

Average WPM

+
{wpm_stats['wpm_average']}
+
Words Per Minute
+
+
+

Peak Hour WPM

+
{wpm_stats['wpm_peak_hour']}
+
Max Speed
+
+
+

Tracking Period

+
{(datetime.fromisoformat(date_range[1]) - datetime.fromisoformat(date_range[0])).days if date_range[0] and date_range[1] else 0}
+
Days of Data
+
+
+

Words Typed

+
{wpm_stats['words_typed']:,}
+
Total Output
+
+
+ +
+
+

Data Retrieval Performance

+
    + {''.join(f'
  • {key.replace("_", " ").title()}: {value:.3f}s
    • ' for key, value in durations.items())} +
+
+ + +
+

Typing Speed & Performance

+

Detailed breakdown of words per minute, efficiency, and accuracy metrics.

+
+
+

Average WPM

+
{wpm_stats['wpm_average']}
+
Over {wpm_stats['total_typing_time_minutes']} minutes
+
+
+

Peak Hour WPM

+
{wpm_stats['wpm_peak_hour']}
+
{wpm_stats['peak_hour'] or 'N/A'}
+
+
+

Typing Efficiency

+
{wpm_stats['typing_efficiency']}
+
Presses per active minute
+
+
+

Consistency Score

+
{perf_insights['consistency_score']}%
+
Daily activity stability
+
+
+

Error Rate

+
{perf_insights['error_rate_percent']}%
+
{perf_insights['total_corrections']:,} corrections
+
+
+

Characters Typed

+
{wpm_stats['characters_typed']:,}
+
{perf_insights['letters_typed']:,} letters • {perf_insights['spaces_typed']:,} spaces
+
+
+
+ +
+

Hourly Activity Distribution

+

Average key press volume by hour of day (00:00–23:00) across the complete dataset.

+
+ +
+
+ +
+

Daily Activity Timeline (Last 365 Days)

+

Daily key press totals over the past year, illustrating long-term usage patterns.

+
+ +
+
+ +
+
+

Activity by Day of Week

+

Distribution of typing activity across weekdays, highlighting weekly patterns.

+
+ +
+
+ +
+

Monthly Activity Trends

+

Monthly key press totals showing seasonal and long-term productivity trends.

+
+ +
+
+
+ +
+

Top 30 Most Frequently Used Keys

+

Ranking of the most pressed keys by total count, excluding system keys.

+
+
+
+
+ +
+

Keyboard Usage Heatmap

+

Visual intensity map of key usage. Color intensity corresponds to press frequency.

+
+
+
+ Low Activity +
+ High Activity +
+
+
+ +
+

Hour × Day Activity Matrix (Last 30 Days)

+

Detailed view of typing patterns by hour and day. Each cell represents one hour of activity.

+
+
+
+ 00:00 + 06:00 + 12:00 + 18:00 + 23:00 +
+
+
+ +
+

Keypresses Per Hour Timeline

+

Hourly key press volume over time, highlighting peak productivity periods.

+
+ +
+
+ +
+
+

Letter Frequency Analysis

+

Relative frequency of letter usage (A-Z), case-insensitive.

+
+ +
+
+ +
+

Number Key Usage

+

Frequency distribution of numeric key usage (0-9).

+
+ +
+
+
+ +
+

Peak Activity Periods

+

Top 10 one-hour periods with highest key press volume.

+
+ + + + + + + + + + {''.join(f'' for i, (time, count) in enumerate(peak_times))} + +
RankTime PeriodKey Presses
{i+1}{time}{count:,}
+
+
+
+ + +
+ +
+ + + +""" + + with open(output_file, 'w', encoding='utf-8') as f: + f.write(html) + + overall_duration = time.time() - overall_start + logging.info(f"Enterprise HTML report generated in {overall_duration:.2f} seconds: {output_file}") + logging.info(f"Total events analyzed: {total_events:,}") + print(f"Enterprise performance report generated: {output_file}") + +if __name__ == '__main__': + db_path = sys.argv[1] if len(sys.argv) > 1 else 'tikker.db' + output_path = sys.argv[2] if len(sys.argv) > 2 else 'tikker_report.html' + + visualizer = TikkerVisualizer(db_path) + visualizer.generate_html(output_path) + visualizer.close()