在实时绘制高速数据流时访问正在写入的文件的末尾

Question

我的问题是指以下问题的精彩答案：

随着这个答案的

gen.py

代码增长得很快，我在下面编写了自己的版本

gen_own.py

，这实际上在向文件写入新数据之前施加了1毫秒的延迟。我还调整了代码

plot.py

并编写了自己的

plot_own.py

，本质上添加了调试语句。尽管我尝试阅读有关

f.seek(0, io.SEEK_END)

行的几个组件的文档，但仍有几点我不明白。这是我所有的问题

我的问题是：我们如何调整

plot_own.py

来与

gen_own.py

（使用较慢的数据流）一起工作

这是代码

gen_own.py

：

#!/usr/bin/env python3

import time
import random

LIMIT_TIME = 100  # s
DATA_FILENAME = "data.txt"


def gen_data(filename, limit_time):
    start_time = time.time()
    elapsed_time = time.time() - start_time
    old_time = time.time()
    with open(filename, "w") as f:
        while elapsed_time < limit_time:
            new_time = time.time()
            if new_time > old_time + 0.001:
                f.write(f"{time.time():30.12f} {random.random():30.12f}\n")  # produces 64 bytes
                f.flush()
                old_time = time.time()
                elapsed = old_time - start_time
            

gen_data(DATA_FILENAME, LIMIT_TIME)

对于能力，这里是

gen.py

的代码（从原始问题复制）

#!/usr/bin/env python3

import time
import random

LIMIT_TIME = 100  # s
DATA_FILENAME = "data.txt"


def gen_data(filename, limit_time):
    start_time = time.time()
    elapsed_time = time.time() - start_time
    with open(filename, "w") as f:
        while elapsed_time < limit_time:
            f.write(f"{time.time():30.12f} {random.random():30.12f}\n")  # produces 64 bytes
            f.flush()
            elapsed = time.time() - start_time
            

gen_data(DATA_FILENAME, LIMIT_TIME)

这是代码

plot_own.py

：

#!/usr/bin/env python3


import io
import time
import matplotlib.pyplot as plt
import matplotlib as mpl
import matplotlib.animation


BUFFER_LEN = 64
DATA_FILENAME = "data.txt"
PLOT_LIMIT = 20
ANIM_FILENAME = "video.gif"


fig, ax = plt.subplots(1, 1, figsize=(10,8))
ax.set_title("Plot of random numbers from `gen.py`")
ax.set_xlabel("time / s")
ax.set_ylabel("random number / #")
ax.set_ylim([0, 1])


def get_data(filename, buffer_len, delay=0.0):
    with open(filename, "r") as f:
        print("f.seek(0, io.SEEK_END): " + str(f.seek(0, io.SEEK_END)))
        data = f.read(buffer_len)
        print("f.tell(): " + str(f.tell()))
        print("f.readline(): " + f.readline())
        print("data: " + data)
        if delay:
            time.sleep(delay)
    return data


def animate(i, xs, ys, limit=PLOT_LIMIT, verbose=False):
    # grab the data
    try:
        data = get_data(DATA_FILENAME, BUFFER_LEN)
        if verbose:
            print(data)
        x, y = map(float, data.split())
        if x > xs[-1]:
            # Add x and y to lists
            xs.append(x)
            ys.append(y)
            # Limit x and y lists to 10 items
            xs = xs[-limit:]
            ys = ys[-limit:]
        else:
            print(f"W: {time.time()} :: STALE!")
    except ValueError:
        print(f"W: {time.time()} :: EXCEPTION!")
    else:
        # Draw x and y lists
        ax.clear()
        ax.set_ylim([0, 1])
        ax.plot(xs, ys)


# save video (only to attach here) 
#anim = mpl.animation.FuncAnimation(fig, animate, fargs=([time.time()], [None]), interval=1, frames=3 * PLOT_LIMIT, repeat=False)
#anim.save(ANIM_FILENAME, writer='imagemagick', fps=10)
#print(f"I: Saved to `{ANIM_FILENAME}`")

# show interactively
anim = mpl.animation.FuncAnimation(fig, animate, fargs=([time.time()], [None]), interval=1)
plt.show()
plt.close()

这是与

plot_own.py

 同时运行时

gen.py

的输出

f.seek(0, io.SEEK_END): 36998872
f.tell(): 36998936
f.readline():      1731141285.629011392593                 0.423847536979

data:        1731141285.629006385803                 0.946414017554

f.seek(0, io.SEEK_END): 37495182
f.tell(): 37495246
f.readline():      1731141285.670451402664                 0.405303398216

data:        1731141285.670446395874                 0.103460518242

f.seek(0, io.SEEK_END): 38084306
f.tell(): 38084370
f.readline():      1731141285.719735860825                 0.360983611461

data:        1731141285.719730854034                 0.318057761442

这是与

plot_own.py

 同时运行时

gen_own.py

的输出

W: 1731141977.7246473 :: EXCEPTION!
f.seek(0, io.SEEK_END): 156426
f.tell(): 156426
f.readline():
data:
W: 1731141977.7611823 :: EXCEPTION!
f.seek(0, io.SEEK_END): 158472
f.tell(): 158472
f.readline():
data:
W: 1731141977.79479 :: EXCEPTION!
f.seek(0, io.SEEK_END): 160518
f.tell(): 160518
f.readline():        1731141977.828338146210                 0.165056626254

data:
W: 1731141977.8283837 :: EXCEPTION!
f.seek(0, io.SEEK_END): 162626
f.tell(): 162626
f.readline():
data:
W: 1731141977.8621912 :: EXCEPTION!
f.seek(0, io.SEEK_END): 164734
f.tell(): 164734
f.readline():
data:

Answer 1

即使没有延迟，你也必须注意，只有 2000 行中的 1 行被读取、打印和显示，延迟 1ms 就是 20 行中的 1 行，但是其中在寻尾和读取时存在一些问题，导致数据空几次，

您可以从这个可书签答案

因此你的plot_own.py 变成：

#!/usr/bin/env python3


import io
import os
import subprocess
import time
import matplotlib.pyplot as plt
import matplotlib as mpl
import matplotlib.animation

def tail(f, lines=1, _buffer=4098):
    """Tail a file and get X lines from the end"""
    # place holder for the lines found
    lines_found = []

    # block counter will be multiplied by buffer
    # to get the block size from the end
    block_counter = -1

    # loop until we find X lines
    while len(lines_found) < lines:
        try:
            f.seek(block_counter * _buffer, os.SEEK_END)
        except IOError:  # either file is too small, or too many lines requested
            f.seek(0)
            lines_found = f.readlines()
            break

        lines_found = f.readlines()

        # we found enough lines, get out
        # Removed this line because it was redundant the while will catch
        # it, I left it for history
        # if len(lines_found) > lines:
        #    break

        # decrement the block counter to get the
        # next X bytes
        block_counter -= 1

    return lines_found[-lines:]

BUFFER_LEN = 64
DATA_FILENAME = "data.txt"
PLOT_LIMIT = 20
ANIM_FILENAME = "video.gif"


fig, ax = plt.subplots(1, 1, figsize=(10,8))
ax.set_title("Plot of random numbers from `gen.py`")
ax.set_xlabel("time / s")
ax.set_ylabel("random number / #")
ax.set_ylim([0, 1])


def get_data(filename, buffer_len, delay=0.0):
    with open(filename, "r") as f:
        data=tail(f, 1, 4098)[0]
        print(data)

        if delay:
            time.sleep(delay)
    return data


def animate(i, xs, ys, limit=PLOT_LIMIT, verbose=False):
    # grab the data
    try:
        data = get_data(DATA_FILENAME, BUFFER_LEN)
        if verbose:
            print(data)
        x, y = map(float, data.split())
        if x > xs[-1]:
            # Add x and y to lists
            xs.append(x)
            ys.append(y)
            # Limit x and y lists to 10 items
            xs = xs[-limit:]
            ys = ys[-limit:]
        else:
            print(f"W: {time.time()} :: STALE!")
    except ValueError:
        print(f"W: {time.time()} :: EXCEPTION!")
    else:
        # Draw x and y lists
        ax.clear()
        ax.set_ylim([0, 1])
        ax.plot(xs, ys)


# save video (only to attach here) 
#anim = mpl.animation.FuncAnimation(fig, animate, fargs=([time.time()], [None]), interval=1, frames=3 * PLOT_LIMIT, repeat=False)
#anim.save(ANIM_FILENAME, writer='imagemagick', fps=10)
#print(f"I: Saved to `{ANIM_FILENAME}`")

# show interactively
anim = mpl.animation.FuncAnimation(fig, animate, fargs=([time.time()], [None]), interval=1)
plt.show()
plt.close()

或

至于你的错误您只需在绘图之前确保数据不为空，这样您的plot_own.py中就不会引发异常：

def animate(i, xs, ys, limit=PLOT_LIMIT, verbose=False):
    # grab the data
    try:
        data = get_data(DATA_FILENAME, BUFFER_LEN)
        if data:
            if verbose:
                print(data)
            x, y = map(float, data.split())
            if x > xs[-1]:
                # Add x and y to lists
                xs.append(x)
                ys.append(y)
                # Limit x and y lists to 10 items
                xs = xs[-limit:]
                ys = ys[-limit:]
            else:
                print(f"W: {time.time()} :: STALE!")
    except ValueError:
        print(f"W: {time.time()} :: EXCEPTION!")
    else:
        # Draw x and y lists
        ax.clear()
        ax.set_ylim([0, 1])
        ax.plot(xs, ys)

是的，您仍然会丢失数据，但是第二个代码是最好的，即只需在使用

if data:

进行绘图之前验证代码中的数据即可

在实时绘制高速数据流时访问正在写入的文件的末尾

问题描述投票：0回答：1

1个回答

最新问题

在实时绘制高速数据流时访问正在写入的文件的末尾

问题描述 投票：0回答：1

1个回答

最新问题

问题描述投票：0回答：1