#!/usr/bin/env python3
"""
activity_cli.py - command line tool to display activity tracking data from rewinddb.
this script provides information about computer usage activity, application usage,
and calendar meetings from the rewind.ai database.
call flow:
1. parse command line arguments using argparse
2. connect to rewinddb database
3. determine time range based on arguments:
- if --relative is provided, calculate relative time range from now
- if --from and --to are provided, use specific time range
4. query activity data for the specified time range
5. format and display results:
- active hours - when the computer was being used
- apps - list of applications used with time spent
- meetings - calendar events during the specified time period
6. close database connection
the cli supports two main query modes:
- relative time queries (e.g., "1 hour", "5h", "30m", "2d", "1w")
- specific time range queries with --from and --to timestamps
examples:
python activity_cli.py --relative "1 day"
python activity_cli.py --relative "5h"
python activity_cli.py --relative "3m"
python activity_cli.py --relative "10d"
python activity_cli.py --relative "2w"
python activity_cli.py --from "2023-05-11 13:00:00" --to "2023-05-11 17:00:00"
python activity_cli.py --from "2023-05-11" --to "2023-05-12" # uses 00:00:00 and 23:59:59
python activity_cli.py --from "13:00:00" --to "17:00:00" # uses today's date
python activity_cli.py --relative "7 days" --debug
python activity_cli.py --relative "1 hour" --env-file /path/to/.env
"""
import argparse
import datetime
import logging
import re
import sys
import time
from datetime import timezone
import rewinddb
from tabulate import tabulate
# configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)
def parse_relative_time(time_str):
"""parse a relative time string into timedelta components.
args:
time_str: string like "1 hour", "5 hours", "30 minutes" or short form "5h", "3m", "10d", "2w"
returns:
dict with keys for days, hours, minutes, seconds
raises:
ValueError: if the time string format is invalid
"""
time_str = time_str.lower().strip()
time_components = {"days": 0, "hours": 0, "minutes": 0, "seconds": 0}
# short form pattern (e.g., "5h", "3m", "10d", "2w")
short_patterns = {
r"^(\d+)w$": lambda x: {"days": int(x) * 7},
r"^(\d+)d$": lambda x: {"days": int(x)},
r"^(\d+)h$": lambda x: {"hours": int(x)},
r"^(\d+)m$": lambda x: {"minutes": int(x)},
r"^(\d+)s$": lambda x: {"seconds": int(x)}
}
# check for short form patterns first
for pattern, handler in short_patterns.items():
match = re.search(pattern, time_str)
if match:
component_values = handler(match.group(1))
for component, value in component_values.items():
time_components[component] = value
return time_components
# long form patterns
patterns = {
r"(\d+)\s*(?:day|days)": "days",
r"(\d+)\s*(?:hour|hours|hr|hrs)": "hours",
r"(\d+)\s*(?:minute|minutes|min|mins)": "minutes",
r"(\d+)\s*(?:second|seconds|sec|secs)": "seconds",
r"(\d+)\s*(?:week|weeks)": "weeks"
}
# try to match each pattern
found_match = False
for pattern, component in patterns.items():
match = re.search(pattern, time_str)
if match:
if component == "weeks":
time_components["days"] += int(match.group(1)) * 7
else:
time_components[component] = int(match.group(1))
found_match = True
if not found_match:
raise ValueError(f"invalid time format: {time_str}. use format like '1 hour', '5h', '30m', '2d', '1w'.")
return time_components
def get_activity_relative(db, time_str, debug=False):
"""get activity data from a relative time period.
args:
db: rewinddb instance
time_str: relative time string (e.g., "1 hour", "5 hours")
debug: whether to print debug information
returns:
dictionary with active hours, app usage, and meetings data
"""
try:
time_components = parse_relative_time(time_str)
if debug:
logger.info(f"getting activity data for the past {time_str}")
logger.info(f"time components: {time_components}")
# get active hours data
active_hours = db.get_active_hours(**time_components)
# get app usage data
app_usage = db.get_app_usage(**time_components)
# get meetings data
try:
meetings = db.get_meetings(**time_components)
except Exception as e:
if debug:
logger.error(f"error retrieving meetings data: {e}")
meetings = None
return {
'active_hours': active_hours,
'app_usage': app_usage,
'meetings': meetings
}
except ValueError as e:
logger.error(f"error: {e}")
sys.exit(1)
def get_activity_absolute(db, from_time_str, to_time_str, debug=False):
"""get activity data from a specific time range.
args:
db: rewinddb instance
from_time_str: start time string in format "YYYY-MM-DD HH:MM:SS", "YYYY-MM-DD HH:MM", "YYYY-MM-DD", "HH:MM:SS", or "HH:MM"
to_time_str: end time string in format "YYYY-MM-DD HH:MM:SS", "YYYY-MM-DD HH:MM", "YYYY-MM-DD", "HH:MM:SS", or "HH:MM"
debug: whether to print debug information
returns:
dictionary with active hours, app usage, and meetings data
"""
def normalize_time_string(time_str, is_end_time=False):
"""normalize time string to handle various formats."""
# check if time_str is time-only format (HH:MM or HH:MM:SS)
if len(time_str) <= 8 and ':' in time_str:
today = datetime.datetime.now().strftime("%Y-%m-%d")
# if it's HH:MM format, add :00 for seconds
if time_str.count(':') == 1:
time_str = f"{time_str}:00"
time_str = f"{today} {time_str}"
# check if it's date-only format (YYYY-MM-DD)
elif len(time_str) == 10 and time_str.count('-') == 2:
if is_end_time:
time_str = f"{time_str} 23:59:59"
else:
time_str = f"{time_str} 00:00:00"
# check if it's date with HH:MM format
elif ' ' in time_str and time_str.split(' ')[1].count(':') == 1:
time_str = f"{time_str}:00"
return time_str
try:
# get local timezone for proper conversion
local_tz = datetime.datetime.now().astimezone().tzinfo
# normalize time strings to handle various formats
from_time_str = normalize_time_string(from_time_str, is_end_time=False)
to_time_str = normalize_time_string(to_time_str, is_end_time=True)
# parse as naive datetime first
from_time_naive = datetime.datetime.strptime(from_time_str, "%Y-%m-%d %H:%M:%S")
to_time_naive = datetime.datetime.strptime(to_time_str, "%Y-%m-%d %H:%M:%S")
# add local timezone info and convert to UTC for database query
from_time = from_time_naive.replace(tzinfo=local_tz).astimezone(timezone.utc)
to_time = to_time_naive.replace(tzinfo=local_tz).astimezone(timezone.utc)
if debug:
logger.info(f"getting activity data from {from_time} to {to_time}")
# get active hours data
active_hours = db.get_active_hours(start_time=from_time, end_time=to_time)
# get app usage data
app_usage = db.get_app_usage(start_time=from_time, end_time=to_time)
# get meetings data
try:
meetings = db.get_meetings(start_time=from_time, end_time=to_time)
except Exception as e:
if debug:
logger.error(f"error retrieving meetings data: {e}")
meetings = None
return {
'active_hours': active_hours,
'app_usage': app_usage,
'meetings': meetings
}
except ValueError as e:
logger.error(f"error: invalid time format. use format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD', 'HH:MM:SS', or 'HH:MM'.")
sys.exit(1)
def convert_to_local_time(dt):
"""convert a utc datetime to local time.
args:
dt: datetime object in utc
returns:
datetime object in local time
"""
if dt is None:
return None
# if datetime has no timezone info, assume it's utc
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
# convert to local time
return dt.astimezone()
def display_active_hours(active_hours, use_utc=False):
"""display active hours data in a formatted way.
args:
active_hours: dictionary with active hours data from get_active_hours()
use_utc: whether to display times in utc (default: false, use local time)
"""
print("\n" + "=" * 80)
print(" ACTIVE HOURS ")
print("=" * 80)
print(f"\nTotal Active Time: {active_hours['total_active_hours']} hours")
print(f"Number of Active Sessions: {active_hours['session_count']}")
print(f"Average Session Length: {active_hours['avg_session_minutes']} minutes")
# display hourly activity
print("\nHourly Activity:")
hourly_data = []
for hour_data in active_hours['hourly_activity']:
hour = hour_data['hour']
hours_active = hour_data['hours']
# convert hour from UTC to local time if needed
if not use_utc:
# create a UTC datetime for this hour to convert to local time
utc_dt = datetime.datetime.now(timezone.utc).replace(hour=hour, minute=0, second=0, microsecond=0)
local_dt = convert_to_local_time(utc_dt)
display_hour = local_dt.hour
else:
display_hour = hour
# create a simple bar chart
bar = "█" * int(hours_active * 4) if hours_active > 0 else ""
hour_str = f"{display_hour:02d}:00"
hourly_data.append([hour_str, f"{hours_active:.2f}", bar])
print(tabulate(hourly_data, headers=["Hour", "Hours Active", "Activity"], tablefmt="simple"))
# display daily activity if available
if active_hours['daily_activity']:
print("\nDaily Activity:")
daily_data = []
for day_data in active_hours['daily_activity']:
date_str = day_data['date']
hours_active = day_data['hours']
# convert date string to local time if needed
if not use_utc and '-' in date_str:
try:
# parse the date string
date_obj = datetime.datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
# convert to local time
local_date = convert_to_local_time(date_obj)
# format as date string
date_str = local_date.strftime("%Y-%m-%d")
except ValueError:
# if parsing fails, keep original string
pass
# create a simple bar chart
bar = "█" * int(hours_active) if hours_active > 0 else ""
daily_data.append([date_str, f"{hours_active:.2f}", bar])
print(tabulate(daily_data, headers=["Date", "Hours Active", "Activity"], tablefmt="simple"))
def display_app_usage(app_usage):
"""display application usage data in a formatted way.
args:
app_usage: dictionary with app usage data from get_app_usage()
"""
print("\n" + "=" * 80)
print(" APPLICATION USAGE ")
print("=" * 80)
# filter out rewind's own app
filtered_apps = [app for app in app_usage['top_apps'] if app['name'] != "ai.rewind.audiorecorder"]
# recalculate total hours for filtered apps
filtered_total_hours = sum(app['hours'] for app in filtered_apps)
# recalculate percentages based on filtered total
if filtered_total_hours > 0:
for app in filtered_apps:
app['percentage'] = (app['hours'] / filtered_total_hours) * 100
# adjust total apps count (subtract 1 if we filtered out rewind)
total_apps = app_usage['total_apps']
if len(filtered_apps) < len(app_usage['top_apps']):
total_apps -= 1
print(f"\nTotal Apps Used: {total_apps}")
print(f"Total Usage Time: {filtered_total_hours:.2f} hours")
# display top apps
print("\nTop Applications by Usage Time:")
app_data = []
for app in filtered_apps:
name = app['name']
hours = app['hours']
percentage = app['percentage']
# create a simple bar chart
bar = "█" * int(percentage / 5) if percentage > 0 else ""
app_data.append([name, f"{hours:.2f}", f"{percentage:.1f}%", bar])
print(tabulate(app_data, headers=["Application", "Hours", "Percentage", "Usage"], tablefmt="simple"))
def display_meetings(meetings, use_utc=False):
"""display meetings data in a formatted way.
args:
meetings: dictionary with meetings data from get_meetings()
use_utc: whether to display times in utc (default: false, use local time)
"""
print("\n" + "=" * 80)
print(" CALENDAR MEETINGS ")
print("=" * 80)
if meetings is None:
print("\nMeetings data could not be retrieved. This may be due to a database schema mismatch.")
return
if meetings['total_events'] == 0:
print("\nNo calendar events found for the specified time period.")
return
print(f"\nTotal Meetings: {meetings['total_events']}")
print(f"Total Meeting Time: {meetings['total_hours']} hours")
print(f"Average Meeting Length: {meetings['avg_meeting_minutes']} minutes")
# display calendar stats
if meetings['calendar_stats']:
print("\nMeetings by Calendar:")
calendar_data = []
for cal in meetings['calendar_stats']:
name = cal['calendar']
hours = cal['hours']
count = cal['event_count']
percentage = cal['percentage']
calendar_data.append([name, count, f"{hours:.2f}", f"{percentage:.1f}%"])
print(tabulate(calendar_data, headers=["Calendar", "Count", "Hours", "Percentage"], tablefmt="simple"))
# display daily meeting hours
if meetings['daily_meeting_hours']:
print("\nMeeting Hours by Day:")
daily_data = []
for day_data in meetings['daily_meeting_hours']:
date_str = day_data['date']
hours = day_data['hours']
# convert date string to local time if needed
if not use_utc and '-' in date_str:
try:
# parse the date string
date_obj = datetime.datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
# convert to local time
local_date = convert_to_local_time(date_obj)
# format as date string
date_str = local_date.strftime("%Y-%m-%d")
except ValueError:
# if parsing fails, keep original string
pass
# create a simple bar chart
bar = "█" * int(hours * 2) if hours > 0 else ""
daily_data.append([date_str, f"{hours:.2f}", bar])
print(tabulate(daily_data, headers=["Date", "Hours", "Meetings"], tablefmt="simple"))
# display hourly distribution
print("\nMeeting Hours by Time of Day:")
hourly_data = []
for hour_data in meetings['hourly_distribution']:
hour = hour_data['hour']
hours = hour_data['hours']
# convert hour from UTC to local time if needed
if not use_utc:
# create a UTC datetime for this hour to convert to local time
utc_dt = datetime.datetime.now(timezone.utc).replace(hour=hour, minute=0, second=0, microsecond=0)
local_dt = convert_to_local_time(utc_dt)
display_hour = local_dt.hour
else:
display_hour = hour
# create a simple bar chart
bar = "█" * int(hours * 4) if hours > 0 else ""
hour_str = f"{display_hour:02d}:00"
hourly_data.append([hour_str, f"{hours:.2f}", bar])
print(tabulate(hourly_data, headers=["Hour", "Hours", "Meetings"], tablefmt="simple"))
def parse_arguments():
"""parse command line arguments.
returns:
parsed argument namespace
"""
parser = argparse.ArgumentParser(
description="display activity tracking data from rewinddb",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
examples:
%(prog)s --relative "1 day"
%(prog)s --relative "5h"
%(prog)s --relative "3m"
%(prog)s --relative "10d"
%(prog)s --relative "2w"
%(prog)s --from "2023-05-11 13:00:00" --to "2023-05-11 17:00:00"
%(prog)s --from "2023-05-11" --to "2023-05-12" # uses 00:00:00 and 23:59:59
%(prog)s --from "13:00:00" --to "17:00:00" # uses today's date
%(prog)s --from "13:00" --to "17:00" # uses today's date, HH:MM format
%(prog)s --relative "7 days" --debug
%(prog)s --relative "1 hour" --env-file /path/to/.env
%(prog)s --relative "1 day" --utc # display times in UTC instead of local time
"""
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("-r", "--relative", metavar="TIME", help="relative time period (e.g., '1 hour', '5h', '3m', '10d', '2w')")
group.add_argument("--from", dest="from_time", metavar="DATETIME",
help="start time in format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD' (uses 00:00:00), 'HH:MM:SS', or 'HH:MM' (uses today's date)")
parser.add_argument("--to", dest="to_time", metavar="DATETIME",
help="end time in format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD' (uses 23:59:59), 'HH:MM:SS', or 'HH:MM' (uses today's date)")
parser.add_argument("--debug", action="store_true", help="enable debug output")
parser.add_argument("--env-file", metavar="FILE", help="path to .env file with database configuration")
parser.add_argument("--utc", action="store_true", help="display times in UTC instead of local time")
args = parser.parse_args()
# validate that if --from is provided, --to is also provided
if args.from_time and not args.to_time:
parser.error("--to is required when --from is provided")
return args
def main():
"""main function for the activity cli tool."""
args = parse_arguments()
# set debug level if requested
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
logger.debug("debug mode enabled")
start_time = time.time()
try:
# connect to the database using rewinddb library
logger.info("connecting to rewind database...")
with rewinddb.RewindDB(args.env_file) as db:
# get activity data based on the specified time range
if args.relative:
logger.info(f"retrieving activity data for the past {args.relative}...")
activity_data = get_activity_relative(db, args.relative, args.debug)
else:
logger.info(f"retrieving activity data from {args.from_time} to {args.to_time}...")
activity_data = get_activity_absolute(db, args.from_time, args.to_time, args.debug)
# display the results
display_active_hours(activity_data['active_hours'], args.utc)
display_app_usage(activity_data['app_usage'])
# display meetings if available
if activity_data['meetings'] is not None:
display_meetings(activity_data['meetings'], args.utc)
else:
print("\n" + "=" * 80)
print(" CALENDAR MEETINGS ")
print("=" * 80)
print("\nMeetings data could not be retrieved. This may be due to a database schema mismatch.")
elapsed_time = time.time() - start_time
logger.info(f"activity data retrieved and displayed in {elapsed_time:.2f} seconds")
except FileNotFoundError as e:
logger.error(f"error: {e}")
logger.error("check your DB_PATH setting in .env file")
sys.exit(1)
except ConnectionError as e:
logger.error(f"error: {e}")
logger.error("check your DB_PASSWORD setting in .env file")
sys.exit(1)
except Exception as e:
logger.error(f"unexpected error: {e}")
logger.error(f"error type: {type(e).__name__}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
