#!/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()) plt.ion() plt.clf() plt.show(block=False) while True: 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() # 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()