personal-data/git_time_tracker/__main__.py

import argparse
import logging
import sys
from pathlib import Path

from personal_data.activity import (
    ActivitySample,
    heuristically_realize_samples,
    merge_adjacent_samples,
)

from .format import cli, icalendar
from .source import csv_file, git_repo

logger = logging.getLogger(__name__)


def filter_samples(
    samples: list[ActivitySample],
    sample_filter: set[str],
) -> list[ActivitySample]:
    assert len(sample_filter) > 0
    return [s for s in samples if set(s.labels).intersection(sample_filter)]


def parse_arguments():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--git-repo',
        action='extend',
        nargs='+',
        type=Path,
        dest='repositories',
        default=[],
    )
    parser.add_argument(
        '--csv-file',
        action='extend',
        nargs='+',
        type=Path,
        dest='csv_files',
        default=[],
    )
    parser.add_argument(
        '--filter',
        action='extend',
        nargs='+',
        type=str,
        dest='sample_filter',
        default=[],
    )
    parser.add_argument(
        '--format',
        action='store',
        type=str,
        dest='format_mode',
        default='cli_report',
        choices=['cli_report', 'icalendar'],
    )
    parser.add_argument(
        '--merge',
        action='store',
        type=str,
        dest='merge',
        default=None,
    )
    parser.add_argument(
        '--out',
        action='store',
        type=Path,
        dest='output_file',
        default='output/samples.ics',
    )
    return parser.parse_args()


def load_samples(args) -> set[ActivitySample]:
    shared_time_stamps_set: set[ActivitySample] = set()

    # Git repositories
    for repo_path in args.repositories:
        logger.warning('Determine commits from %s', repo_path)
        shared_time_stamps_set |= set(
            git_repo.iterate_samples_from_git_repository(repo_path),
        )
        del repo_path

    # CSV Files
    for csv_path in args.csv_files:
        logger.warning('Load samples from %s', csv_path)
        shared_time_stamps_set |= set(
            csv_file.iterate_samples_from_csv_file(csv_path),
        )
        del csv_path

    return shared_time_stamps_set


def main():
    logging.basicConfig()
    logging.getLogger('git_time_tracker').setLevel('INFO')

    args = parse_arguments()

    # Determine samples
    shared_time_stamps_set = load_samples(args)

    # Sort samples
    shared_time_stamps = sorted(shared_time_stamps_set, key=lambda s: s.end_at)
    del shared_time_stamps_set

    # Filter samples
    sample_filter = args.sample_filter
    if len(sample_filter) != 0:
        logger.warning('Filtering %s samples', len(shared_time_stamps))
        shared_time_stamps = filter_samples(shared_time_stamps, sample_filter)
        logger.warning('Filtered down to %s samples', len(shared_time_stamps))

    # Heuristic samples
    logger.warning('Realizing %s samples', len(shared_time_stamps))
    shared_time_stamps = list(heuristically_realize_samples(shared_time_stamps))

    # Merge adjacent
    if args.merge:
        shared_time_stamps = merge_adjacent_samples(shared_time_stamps, args.merge)
        logger.warning('Merged to %s samples', len(shared_time_stamps))

    # Output format
    if args.format_mode == 'cli_report':
        for t in cli.generate_report(shared_time_stamps):
            sys.stdout.write(t)
    elif args.format_mode == 'icalendar':
        icalendar.generate_icalendar_file(
            shared_time_stamps,
            file=args.output_file,
        )


if __name__ == '__main__':
    main()