A Python-based Continuous Flow Diagram (CFD) for Trello

trello_continuous_flow.py

import os
import sys
import pytz
import numpy as np
import pandas as pd
import plotly.express as px
from termcolor import cprint
from datetime import datetime, timedelta
from collections import namedtuple, defaultdict
from trello import TrelloClient

# Set Up the Environment
api_key = os.environ.get('TRELLO_API_KEY')
api_secret = os.environ.get('TRELLO_API_SECRET')
api_token = os.environ.get('TRELLO_API_TOKEN')

client = TrelloClient(api_key=api_key, api_secret=api_secret, token=api_token)

board_name = sys.argv[1]

board_id = None
for board in client.list_boards():
    if board_name.lower() in board.name.lower():
        board_id = board.id
        break

# Function to fetch data from Trello
def fetch_data(board_id):
    board = client.get_board(board_id)
    actions = board.fetch_actions(action_filter='updateCard:closed,updateCard:idList,createCard,deleteCard', action_limit=1000)
    
    data = []
    for action in actions:
        action_type = action['type']
        date = action['date']
        card_id = action['data']['card']['id']
        card_name = action['data']['card']['name']

        if action_type == 'createCard':
            source_list = None
            destination_list = action['data']['list']['name']
            data.append({'date': date, 'card_id': card_id, 'card_name': card_name, 'action': 'create', 'destination_list': destination_list})
        elif action_type == 'updateCard':
            if 'listBefore' in action['data']:
                source_list = action['data']['listBefore']['name']
                destination_list = action['data']['listAfter']['name']
                data.append({'date': date, 'card_id': card_id, 'card_name': card_name, 'action': 'move', 'source_list': source_list, 'destination_list': destination_list})
            elif 'closed' in action['data']['card']:
                closed = action['data']['card']['closed']
                if closed:
                    source_list = action['data']['list']['name']
                    destination_list = None
                    data.append({'date': date, 'card_id': card_id, 'card_name': card_name, 'action': 'archive', 'source_list': source_list, 'destination_list': destination_list})
                else:
                    source_list = None
                    destination_list = action['data']['list']['name']
                    data.append({'date': date, 'card_id': card_id, 'card_name': card_name, 'action': 'unarchive', 'source_list': source_list, 'destination_list': destination_list})

        elif action_type == 'deleteCard':
            source_list = action['data']['list']['name']
            destination_list = None
            data.append({'date': date, 'card_id': card_id, 'card_name': card_name, 'action': 'delete', 'source_list': source_list, 'destination_list': destination_list})
        
    df = pd.DataFrame(data)

    # Convert date to datetime
    df['date'] = pd.to_datetime(df['date'], utc=True)
    
    # Sort by date ascending
    df = df.sort_values(by=['date'])

    return df

def rename_list(x):
    if isinstance(x, float):
        return ''
    
    if x is None:
        return ''

    if 'backlog' in x.lower():
        return 'backlog'
    elif 'doing' in x.lower():
        return 'doing'
    elif 'done' in x.lower():
        return 'done'
    return 'other'

# Cumulative Total (for each list)
totals = {}

# Cumulative Totals (at each point in time)
totals_entry = namedtuple('totals_entry', ['date', 'backlog', 'doing', 'done'])
cumulative_totals = []

def process_cumulative_flow(df):
    df['date'] = pd.to_datetime(df['date']).dt.date
    
    df['source_list'] = df['source_list'].apply(rename_list)
    df['destination_list'] = df['destination_list'].apply(rename_list)

    for index, row in df.iterrows():
        if row['action'] == 'create':
            totals[row['destination_list']] = totals.get(row['destination_list'], 0) + 1

        elif row['action'] == 'unarchive':
            totals[row['destination_list']] = totals.get(row['destination_list'], 0) + 1

        elif row['action'] == 'delete':
            totals[row['source_list']] = totals.get(row['source_list'], 0) - 1

        elif row['action'] == 'archive':
            totals[row['source_list']] = totals.get(row['source_list'], 0) - 1

        elif row['action'] == 'move':
            totals[row['source_list']] = totals.get(row['source_list'], 0) - 1
            totals[row['destination_list']] = totals.get(row['destination_list'], 0) + 1

        cumulative_totals.append(totals_entry(row['date'], totals.get('backlog', 0), totals.get('doing', 0), totals.get('done', 0)))

    totals_df = pd.DataFrame(cumulative_totals)

    # Keep only the last entry for each date
    totals_df = totals_df.drop_duplicates(subset=['date'], keep='last')

    # Fill in missing dates
    min_date = totals_df['date'].min()
    max_date = totals_df['date'].max()
    all_dates = pd.date_range(min_date, max_date)
    totals_df = totals_df.set_index('date').reindex(all_dates).fillna(method='ffill').fillna(0).reset_index()

    return totals_df

def time_in_destination(df):
    '''
    For each card, calculate the time spent it its current destination
    '''

    out_data = {}

    # For each unique card_id
    card_ids = df['card_id'].unique()

    for card_id in card_ids:
        # Get all the rows for that card_id
        card_df = df[df['card_id'] == card_id]

        # Get the last row where destination_list is not null
        last_row = card_df[card_df['destination_list'].notnull()].iloc[-1]
        # print(last_row)

        # Get the time spent in that destination
        time_spent = datetime.now(pytz.utc) - last_row['date']

        # Add to out_data
        out_data[card_id] = {'destination_list': last_row['destination_list'], 'time_spent': time_spent}

    return out_data

def average_time_in_destination(t_dest_dict):
    '''
    For each destination, calculate the average time a card spends in that destination
    '''

    out_data = {}
    data = defaultdict(list)

    for card_id, card_data in t_dest_dict.items():
        list_name = rename_list(card_data['destination_list'])
        data[list_name].append(card_data['time_spent'])

    for destination_list, time_spent_list in data.items():
        average_time_spent = np.mean([x.total_seconds() for x in time_spent_list])
        
        # Convert back to a timedelta
        out_data[destination_list] = timedelta(seconds=average_time_spent)

    return out_data

# Fetch and process the data
df = fetch_data(board_id)
print(df)
times_in_destination_by_card = time_in_destination(df)
times_by_destination = average_time_in_destination(times_in_destination_by_card)
print(times_by_destination)

totals_df = process_cumulative_flow(df)

cprint('Cumulative Totals:', 'blue')
print(totals_df)

deltas_df = totals_df.diff()
# set the first row of deltas_df to be the same as the first row of totals_df
deltas_df.iloc[0] = totals_df.iloc[0]
# drop the index column from deltas_df
deltas_df = deltas_df.drop(columns=['index'])

cprint('Additions and Deletions:', 'blue')
print(deltas_df)

def get_backlog_colour(x):
    if x > 30:
        return 'red'
    elif x > 7:
        return 'yellow'
    return 'green'

def get_doing_colour(x):
    if x > 7:
        return 'red'
    elif x > 3:
        return 'yellow'
    return 'green'

# Statistics

# Average Work in Progress
wip = totals_df['doing'].mean()
cprint(f'Average Work in Progress: {wip:.2f} tasks at a time', 'green' if wip < 3 else 'red')

# Average Time in Backlog
time_in_backlog = times_by_destination['backlog']
print('Average Time in:')
cprint(f'\t Backlog: {time_in_backlog.days} days', get_backlog_colour(time_in_backlog.days))

# Average Time in Doing
time_in_doing = times_by_destination['doing']
cprint(f'\t Doing: {time_in_doing.days} days', get_doing_colour(time_in_doing.days))

# Backlog Growth (tasks per day)
backlog_growth = deltas_df['backlog'].mean()
median_backlog_growth = deltas_df['backlog'].median()
cprint(f'Backlog Growth: {backlog_growth:.2f} tasks per day', 'blue')

# Done Growth (tasks per day)
done_growth = deltas_df['done'].mean()
median_done_growth = deltas_df['done'].median()
cprint(f'Done Growth: {done_growth:.2f} tasks per day', 'blue')

# Estimated Project Completion
margin_of_safety = 0.3
most_recent_backlog_size = totals_df['backlog'].iloc[-1]
estimated_completion = most_recent_backlog_size / done_growth
cprint(f'Estimated Project Completion: {estimated_completion:.0f} days (assuming fixed backlog, no margin of safety added)', 'blue')
estimated_completion_with_margin = estimated_completion * (1 + margin_of_safety)


# Ratio of Growth (done/backlog)
ratio = done_growth / backlog_growth
cprint(f'{ratio:.2f} tasks done for every task added to the backlog', 'green' if ratio > 1 else 'red')

# Visualization with Plotly Express
# fig = px.area(totals_df, x='index', y=['done', 'doing', 'backlog'], title='Cumulative Flow Diagram')

# green:done, yellow:doing, red:backlog
fig = px.area(totals_df, x='index', y=['done', 'doing', 'backlog'], title='Cumulative Flow Diagram', color_discrete_sequence=['#33d9b2', '#ff793f', '#706fd3'])

fig.show()