Skip to content

Instantly share code, notes, and snippets.

Last active Jul 14, 2020
What would you like to do?
Experiments with live animation using asyncio and writing to a file. Works with two different processes, but only data generator works in asyncio, not the animation itself.
import json
import random
import asyncio
import argparse
import matplotlib.pyplot as plt
import matplotlib.animation as animation
from functools import partial
def load_data(path, lines=False):
Can load x,y data in two formats: plain json or json lines. The lines argument is
used for the continuous_updater mode, where data is written to an open file.
if lines:
data = {"x": [], "y": []}
with open(path, 'r') as f:
for line in f:
row = json.loads(line.strip())
return data
with open(path, 'r') as f:
return json.load(f)
def plot_file(i, path, ax, title=None, lines=False):
Animates each frame i by fully reading the data from the file and plotting a new
axis, clearing what was there before.
data = load_data(path, lines)
ax.plot(data["x"], data["y"])
if title is not None:
def animate(path, title=None, lines=False):
Creates the animation figure, runs the function animation, and shows it.
fig, ax = plt.subplots(figsize=(9, 6))
plotter = partial(plot_file, path=path, ax=ax, title=title, lines=lines)
# Why do we need the variable for this to work?
_ = animation.FuncAnimation(fig, plotter, interval=1000)
async def animator(path, title=None):
Attempt at coroutine based animator ... doesn't work so far.
animate(path, title)
async def updater(path, delay=0.75):
A continuous data updater coroutine that writes a whole json object, closing the
file after each write.
for _ in range(1000):
await asyncio.sleep(delay)
except KeyboardInterrupt:
async def continuous_updater(path, delay=0.75):
A continuous data updater coroutine that keeps the file open and writes json lines.
# Does not close the file between each write
y = 50
with open(path, 'w') as f:
for x in range(1000):
await asyncio.sleep(delay)
y += random.randrange(-5, 5)
data = json.dumps({"x": x, "y": y})
# If we don't flush, the animator won't see any data
except KeyboardInterrupt:
def update_data(path, x=None, y=None):
Modifies the underlying data with a random walk - writes a whole json file.
with open(path, 'r') as f:
data = json.load(f)
except Exception:
data = {"x": [0.0], "y": [50.0]}
if x is None:
x = data["x"][-1] + 1
if y is None:
y = data["y"][-1] + random.randrange(-5, 5)
with open(path, 'w') as o:
json.dump(data, o)
async def concurrent_animation(path, delay=0.75, title=None):
Attempt at running the updater and the animation in the same coroutine, didn't work.
utask = asyncio.create_task(updater(path, delay))
atask = asyncio.create_task(animator(path, title))
await asyncio.gather(utask, atask)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("-f", "--file", default="test.json")
parser.add_argument("-l", "--lines", action="store_true")
parser.add_argument("-u", "--updater", action="store_true")
parser.add_argument("-d", "--delay", type=float, default=0.75)
parser.add_argument("-a", "--animate", action="store_true")
parser.add_argument("-t", "--title", default="Super Awesome Animation")
parser.add_argument("-x", type=float, default=None)
parser.add_argument("-y", type=float, default=None)
args = parser.parse_args()
if args.updater and args.animate:, args.delay, args.title))
elif args.updater:
if args.lines:, args.delay))
else:, args.delay))
elif args.animate:
animate(args.file, args.title, args.lines)
update_data(args.file, args.x, args.y)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment