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

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)
	# TODO: error handling?
	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())
	
	while True:
		time, states, jobstates = get_data()
		assert time not in ts.data
		ts.data[time] = (states, jobstates)
		ts.states |= states.keys()
		ts.jobstates |= jobstates.keys()

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

		# Show the plot
		x = sorted(ts.data.keys()) # times
		ys = [tuple(ts.data[time][0].get(state, 0) for state in ts.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))
		plt.stackplot(x, *ys)
		plt.show(block=False)

		plt.pause(poll_rate)
		plt.close()
		plt.clf()


main()