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.

ani.py

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())
                data["x"].append(row["x"])
                data["y"].append(row["y"])
        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.clear()
    ax.plot(data["x"], data["y"])
    ax.set_xlabel("x")
    ax.set_ylabel("y")

    if title is not None:
        ax.set_title(title)


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)
    plt.show()


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):
        try:
            await asyncio.sleep(delay)
            update_data(path)
        except KeyboardInterrupt:
            return


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):
            try:
                await asyncio.sleep(delay)
                y += random.randrange(-5, 5)
                data = json.dumps({"x": x, "y": y})
                f.write(data+"\n")

                # If we don't flush, the animator won't see any data
                f.flush()
            except KeyboardInterrupt:
                return


def update_data(path, x=None, y=None):
    """
    Modifies the underlying data with a random walk - writes a whole json file.
    """
    try:
        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)

    data["x"].append(x)
    data["y"].append(y)

    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:
        asyncio.run(concurrent_animation(args.file, args.delay, args.title))
    elif args.updater:
        if args.lines:
            asyncio.run(continuous_updater(args.file, args.delay))
        else:
            asyncio.run(updater(args.file, args.delay))
    elif args.animate:
        animate(args.file, args.title, args.lines)
    else:
        update_data(args.file, args.x, args.y)