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 sys

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.')
			sys.print_exc()

		# Save the pickle
		# Should probably save elsewhere and do atomic rename, but whatever.
		with open(pickle_file, 'wb') as f:
			pickle.dump(ts, f)
		print('pickle saved')

		# Show the plot
		x = sorted(ts.data.keys()) # times
		states = sorted(ts.states)
		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()