stats

#!/usr/bin/env python3
#

import sys
from argparse                           import ArgumentParser, RawDescriptionHelpFormatter
from datetime                           import datetime, timezone, timedelta
from lib.bugzdb                         import BugzDB
from lib.bug                            import Bug, BugHelper
import lib.colored
import yaml


def pro(*args, **kwargs):
    print(*args, file=sys.stdout, **kwargs)

def style(text, color):
    return lib.colored.stylize(text, color)

def CS(txt, clr):
    return style(txt, clr)


epoch = datetime(2009, 3, 2, 0, 0, tzinfo=timezone.utc)

class BadDelta(Exception):
    def __init__(self, left, right):
        self.left = left
        self.right = right

def date_of_timestamp(ts):
    date = epoch + timedelta(seconds=ts)
    return date

def duration(end_time, start_time):
    if end_time == 0 or start_time == 0:
        pretty_time = '-'
        total_seconds = 0
    else:
        delta = date_of_timestamp(end_time) - date_of_timestamp(start_time)
        pretty_time = ptd(int(delta.total_seconds()))
        total_seconds = delta.total_seconds()
    return pretty_time, total_seconds

def find_all_cycles():
    results = []
    q = 'select distinct cycle from bugs where cycle is not null and cycle != \'None\' and cycle <> \'\' order by cycle;'
    for r in BugHelper().query(q):
        results.append(r['cycle'])
    return results

# ptd
#   Print Time Delta - pretty print the delta interval (integer).
#
def ptd(seconds):
    sign_string = '-' if seconds < 0 else ''
    seconds = abs(int(seconds))
    days, seconds = divmod(seconds, 86400)
    hours, seconds = divmod(seconds, 3600)
    minutes, seconds = divmod(seconds, 60)
    if days > 0:
        return '%s%dd %2dh %2dm' % (sign_string, days, hours, minutes)
    elif hours > 0:
        return '%s%dh %2dm' % (sign_string, hours, minutes)
    elif minutes > 0:
        return '%s%dm' % (sign_string, minutes)
    elif seconds > 0:
        return '%s%2ds' % (sign_string, seconds)
    else:
        return '-'

def compensate_for_weekend(right_edge, left_edge):
    diff = right_edge - left_edge
    if diff < 0:
        raise BadDelta(left_edge, right_edge)

    if diff < 86400:
        return diff

    le = epoch + timedelta(seconds=left_edge)
    re = epoch + timedelta(seconds=right_edge)
    for x in range((re - le).days + 1):
        day = le + timedelta(x)
        if day.weekday() > 4:
            diff = diff - 86400
    return diff

class Stats():

    def __init__(self):
        self.bdb = BugzDB()
        self.row1 = '    '
        self.row2 = '    '
        with open('config.yaml', 'r') as f:
            cfg = yaml.safe_load(f.read())
        self.clr_title             = lib.colored.fg(cfg['colors']['title'])
        self.clr_delta             = lib.colored.fg(cfg['colors']['delta'])
        self.clr_exceeds_threshold = lib.colored.fg(cfg['colors']['exceeds_threshold']) + lib.colored.attr('bold')
        self.clr_default           = lib.colored.fg(cfg['colors']['default'])
        self.clr_series            = lib.colored.fg(cfg['colors']['series']) + lib.colored.attr('bold')
        self.clr_package           = lib.colored.fg(cfg['colors']['package'])
        self.clr_bg_odd_rows       = lib.colored.bg(cfg['colors']['bg_odd_rows'])
        self.clr_bg_even_rows      = lib.colored.bg(cfg['colors']['bg_even_rows'])
        self.clr_skew              = lib.colored.fg(cfg['colors']['skew'])

        self.th_total              = cfg['thresholds']['total']
        self.th_ready              = cfg['thresholds']['ready']
        self.th_waiting            = cfg['thresholds']['waiting']
        self.th_crank              = cfg['thresholds']['crank']
        self.th_build              = cfg['thresholds']['build']
        self.th_proposed           = cfg['thresholds']['proposed']
        self.th_review_start       = cfg['thresholds']['review_start']
        self.the_review            = cfg['thresholds']['review']
        self.th_release            = cfg['thresholds']['release']
        self.th_bt                 = cfg['thresholds']['bt']
        self.th_rt                 = cfg['thresholds']['rt']
        self.th_at                 = cfg['thresholds']['at']
        self.th_sru                = cfg['thresholds']['sru']
        self.th_new                = cfg['thresholds']['new']

    def h1(self, text):
        self.row1 += CS('{:<22s}'.format(text),      self.clr_title)
        self.row2 += CS('{:<22s}'.format('-' * 20),  self.clr_title)

    def h2(self, text):
        self.row1 += CS('{:>22s}'.format(text),      self.clr_title)
        self.row2 += CS('{:>22s}'.format('-' * 18),  self.clr_title)

    def cell(self, text):
        pass

    def delta_threshold(self, right_edge, left_edge, threshold):
        oc = self.clr_delta
        if right_edge == 0 or left_edge == 0:
            val = '-'
            diff = 0
        else:
            diff = compensate_for_weekend(right_edge, left_edge)
            if diff > threshold:
                oc = self.clr_exceeds_threshold
            val = ptd(diff)
        return val, oc, diff

    def render_header(self):
        SPACE = ' '
        self.row1 += f'{SPACE:26s}'
        self.row2 += f'{SPACE:26s}'

        self.h1('Bug ID (spin)')
        if args.crank:
            self.h2('Crank')
        if args.build:
            self.h2('Build')
        if args.bt:
            self.h2('Boot')
        if args.sru:
            self.h2('SRU Review')
        if args.new:
            self.h2('New Review')
        if args.proposed:
            self.h2('-proposed')
        if args.rt:
            self.h2('Regression')
        # if args.ct:
        #     self.h2('Certification')
        if args.at:
            self.h2('ADT')
        if args.updates:
            self.h2('-updates')

        pro(self.row1)
        pro(self.row2)

    def render(self, args):
        self.render_header()
        row_odd = True
        for cycle in args.cycles:
            for series in BugHelper().series_in_cycle_ex(cycle):
                if args.series is not None and series not in args.series:
                    continue
                pro(CS(series, self.clr_series))
                for stats in BugHelper().stats_in_cycle_and_series(cycle, series):
                    oc = self.clr_delta
                    if args.packages is not None and stats.package not in args.packages:
                        continue
                    if row_odd:
                        COLOR_BG = self.clr_bg_odd_rows
                        row_odd = False
                    else:
                        COLOR_BG = self.clr_bg_even_rows
                        row_odd = True
                    o  = '    '
                    o += CS(f'{stats.package:26s}', self.clr_package + COLOR_BG)
                    bug = Bug().load(stats.id)
                    s = f'lp: #{stats.id} ({bug.spin})'
                    if not bug.master_bug_id:
                        s += ' (m)'
                    o += CS(f'{s:<22s}', self.clr_default + COLOR_BG)

                    if args.crank:
                        # Crank Time
                        #
                        # pro(f'prepare-package.date_in_progress: {str(date_of_timestamp(bug.tasks["prepare-package"].date_in_progress))}')
                        # pro(f'prepare-package.date_fix_released: {str(date_of_timestamp(bug.tasks["prepare-package"].date_fix_released))}')
                        # pro(f'duration: {duration(bug.tasks["prepare-package"].date_fix_released, bug.tasks["prepare-package"].date_in_progress)[0]}')
                        # val, oc, diff = self.delta_threshold(bug.tasks['prepare-package'].date_fix_released, bug.tasks['prepare-package'].date_in_progress, self.th_crank)
                        val = duration(bug.tasks["prepare-package"].date_fix_released, bug.tasks["prepare-package"].date_in_progress)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.build:
                        # Build Time
                        try:
                            # val, oc, diff = self.delta_threshold(bug.tasks['boot-testing'].date_triaged, bug.tasks['prepare-package'].date_fix_committed, self.th_build)
                            val = duration(bug.tasks['boot-testing'].date_triaged, bug.tasks['prepare-package'].date_fix_committed)[0]
                        except BadDelta:
                            # How this happens is that someone resets the prepare-package status to New and then has SWM do it's thing. This happens after promote-to-proposed
                            # has been set to to something after confirmed and so there is skew in the delta.
                            #
                            val = '* date skew *'
                            oc  = self.clr_skew
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.bt:
                        # Boot Testing
                        #
                        # val, oc, diff = self.delta_threshold(bug.tasks['boot-testing'].date_fix_released, bug.tasks['boot-testing'].date_triaged, self.th_bt)
                        val = duration(bug.tasks['boot-testing'].date_fix_released, bug.tasks['boot-testing'].date_triaged)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.sru:
                        # val, oc, diff = self.delta_threshold(bug.tasks['sru-review'].date_fix_released, bug.tasks['sru-review'].date_triaged, self.th_sru)
                        val = duration(bug.tasks['sru-review'].date_fix_released, bug.tasks['sru-review'].date_confirmed)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.new:
                        # val, oc, diff = self.delta_threshold(bug.tasks['new-review'].date_fix_released, bug.tasks['new-review'].date_triaged, self.th_new)
                        val = duration(bug.tasks['new-review'].date_fix_released, bug.tasks['new-review'].date_confirmed)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.proposed:
                        # Time until -proposed
                        #
                        # val, oc, diff = self.delta_threshold(bug.tasks['promote-to-proposed'].date_fix_released, bug.tasks['prepare-package'].date_in_progress, self.th_proposed)
                        val = duration(bug.tasks['promote-to-proposed'].date_fix_released, bug.tasks['prepare-package'].date_in_progress)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.rt:
                        try:
                            # val, oc, diff = self.delta_threshold(bug.tasks['regression-testing'].date_fix_released, bug.tasks['regression-testing'].date_incomplete, self.th_rt)
                            val = duration(bug.tasks['regression-testing'].date_fix_released, bug.tasks['regression-testing'].date_incomplete)[0]
                        except BadDelta:
                            # How this happens is that someone resets the prepare-package status to New and then has SWM do it's thing. This happens after promote-to-proposed
                            # has been set to to something after confirmed and so there is skew in the delta.
                            #
                            val = '* date skew *'
                            oc  = self.clr_skew
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    # if args.ct:
                    #     val, oc, diff = self.delta_threshold(bug.tasks['certification-testing'].date_fix_released, bug.tasks['certification-testing'].date_opinion, self.th_proposed)
                    #     o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.at:
                        # val, oc, diff = self.delta_threshold(bug.tasks['automated-testing'].date_fix_released, bug.tasks['automated-testing'].date_incomplete, self.th_rt)
                        val = duration(bug.tasks['automated-testing'].date_fix_released, bug.tasks['automated-testing'].date_incomplete)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    if args.updates:
                        # Time until -updates
                        #
                        # val, oc, diff = self.delta_threshold(bug.tasks['promote-to-updates'].date_fix_released, bug.tasks['prepare-package'].date_in_progress, self.th_updates)
                        val = duration(bug.tasks['promote-to-updates'].date_fix_released, bug.tasks['prepare-package'].date_in_progress)[0]
                        o += style(f'{val:>22s}', oc + COLOR_BG)

                    pro(o)

        legend = '''
Legend:
------------------------------------------------------------------------------------------------------------
'''
        if args.crank:
            legend += 'Crank        - Time from when the prepare-package was confirmed until the prepare-package was committed.\n'
        if args.build:
            legend += 'Build        - Time from prepare-package set to Fix Committed until boot-testing set to Triaged\n'
        if args.proposed:
            legend += '-proposed    - Time from prepare-package set to In Progress until promote-to-proposed set to Fix Released\n'
        if args.proposed:
            legend += '-updates     - Time from promote-to-proposed set to In Progress until promote-to-proposed set to Fix Released\n'
        if args.bt:
            legend += 'Boot         - Time from boot-testing set to Incomplete until boot-testing set to Fix Released\n'
        if args.rt:
            legend += 'Regression   - Time from regression-testing set to Incomplete until regression-testing set to Fix Released\n'
        if args.at:
            legend += 'ADT          - Time from automated-testing set to Incomplete until automated-testing set to Fix Released\n'
        if args.sru:
            legend += 'SRU Review   - Time from sru-review set to Triaged until sru-review set to Fix Released\n'
        if args.new:
            legend += 'New Review   - Time from new-review set to Triaged until new-review set to Fix Released\n'

        pro(legend)


if __name__ == '__main__':
    # Command line argument setup and initial processing
    #
    app_description = '''
This util takes the data in the database created and updated via db-update and produces Ubuntu Kernel SRU statistics for the various kernels for the
specified cycles and series.
    '''
    app_epilog = '''
Examples:
    ./stats --cycles 2024.04.29 --series noble --packages linux-nvidia
    '''
    crank_help    = 'How long to "crank": prepare-package(in progress) -> prepare-package(fix committed)'
    build_help    = 'Build time duration: prepare-package(in progress) -> prepare-package(fix committed)'
    boot_help     = 'Boot testing duration: prepare-package(in progress) -> prepare-package(fix committed)'
    proposed_help = 'Show the times for each stage between when a crank starts and when the kernel reaches the -proposed pocket and a total time as well.'
    parser = ArgumentParser(description=app_description, epilog=app_epilog, formatter_class=RawDescriptionHelpFormatter)
    parser.add_argument('--cycles',   nargs="?", default=None, help='The SRU cycle tag.')
    parser.add_argument('--series',   nargs='?', help='One or more Ubuntu series (noble,jammy,focal) separated by commas.')
    parser.add_argument('--packages', nargs='?', help='One or more kernel source packages separated by commas.')
    parser.add_argument('--crank', action='store_true', default=False, help=crank_help)
    parser.add_argument('--build', action='store_true', default=False, help=build_help)
    parser.add_argument('--proposed', action='store_true', default=False, help=proposed_help)
    parser.add_argument('--updates', action='store_true', default=False, help='Show the time it takes to go from the start of a crank until the kernel reaches the -updates pocket.')
    parser.add_argument('--bt', action='store_true', default=False, help='Show the time it takes to perform boot testing.')
    parser.add_argument('--rt', action='store_true', default=False, help='Show the time it takes to perform regression testing.')
    # parser.add_argument('--ct', action='store_true', default=False, help='certification-testing:Opinion -> certification-testing:Fix Released')
    parser.add_argument('--at', action='store_true', default=False, help='Show how long to perform ADT testing.')
    parser.add_argument('--sru', action='store_true', default=False, help='Show how long it takes for an sru review.')
    parser.add_argument('--new', action='store_true', default=False, help='Show how long it takes for a new package review.')
    parser.add_argument('--all', action='store_true', default=False, help='Show duarations for all stages.')

    args = parser.parse_args()
    try:
        args.cycles = args.cycles.split(',')
    except AttributeError:
        args.cycles = find_all_cycles()
        pro(args.cycles)
    try:
        args.series = args.series.split(',')
    except AttributeError:
        pass
    try:
        args.packages = args.packages.split(',')
    except AttributeError:
        pass
    if args.proposed:
        args.crank = True
        args.build = True
        args.new   = True
        args.sru   = True
        args.bt    = True
    if args.updates:
        args.proposed = True
    if args.all:
        args.crank = True
        args.build = True
        args.new   = True
        args.sru   = True
        args.bt    = True
        args.proposed = True
        args.rt = True
        args.at = True
        args.updates = True
    Stats().render(args)