fosdem_video_progress/fosdem_video_progress.py

#!/usr/bin/env python3
import pickle
from matplotlib import pyplot as plt
import requests
import json
from datetime import datetime
from dataclasses import dataclass
from collections import defaultdict
from pathlib import Path
import traceback
import os

api_url = r'https://review.video.fosdem.org/api/v1/event/1/overview'
poll_rate = 10*60 # seconds
pickle_file = Path(r'./talk_data.pickle')

@dataclass
class TimeSeries:
	"The object to pickle and unpickle"
	states: set[str]
	jobstates: set[str]
	data: dict[datetime, tuple[dict[str, int], dict[str, int]]] # time → (states(name → count), jobstates(ditto))

def get_data() -> tuple[datetime, dict[str, int], dict[str, int]]:
	resp = requests.get(api_url)
	if resp.status_code != 200: raise RuntimeError(f'bad status code: {resp.status_code}')
	time = datetime.now().astimezone()
	data = json.loads(resp.content.decode())
	states = defaultdict(lambda: 0)
	jobstates = defaultdict(lambda: 0)
	# should be an array/list
	for talk in data:
		states[talk['state']] += 1
		jobstates[talk['progress']] += 1
	return time, dict(states), dict(jobstates)


def main():
	if pickle_file.exists():
		with open(pickle_file, 'rb') as f:
			ts = pickle.load(f)
	else:
		print("New time series.")
		ts = TimeSeries(states=set(), jobstates=set(), data=dict())

	plt.ion()
	plt.clf()
	plt.show(block=False)
	
	while True:
		try:
			time, states, jobstates = get_data()
			print(f'data got at {time}')
			assert time not in ts.data
			ts.data[time] = (states, jobstates)
			ts.states |= states.keys()
			ts.jobstates |= jobstates.keys()
		except RuntimeError as e:
			print('Could not get/save data.')
			traceback.print_exc()

		# Save the pickle
		# Keep the previous pickle in case we fail (e.g. ENOSPC)
		os.rename(pickle_file, pickle_file.with_suffix('.bak'))
		with open(pickle_file, 'wb') as f:
			pickle.dump(ts, f)
		print('pickle saved')

		# Show the plot
		x = sorted(ts.data.keys()) # times
		# first is lowest
		states_low = (
			'lost',
			'ignored',
			'uninteresting',
			'broken',
			'preview',
			'cutting',
			'transcoding',
			'publishing',
			)
		states_high = (
			'done',
			'waiting_for_files',
			)
		states_other = tuple(sorted(ts.states - (set(states_low) | set(states_high))))
		states = states_low + states_other + states_high
		ys = [tuple(ts.data[time][0].get(state, 0) for state in states) for time in x] # numbers per state
		# We do not show jobstates atm. Too lazy.
		# ys are transposed – we need vectors by times, not by states.
		ys = list(zip(*ys))
		last = ts.data[max(ts.data.keys())][0]
		labels = [f'{state}: {last.get(state, 0)}' for state in states]
		plt.clf()
		plt.stackplot(x, ys, labels=labels)
		plt.legend(loc='upper left')
		print('showing plot')

		plt.pause(poll_rate)

main()