如何每10秒将数据写入文件

Question

我是一名 JS 开发人员，在处理从蓝牙温度传感器读取数据的 Raspberry-Pi3 项目时试图学习一些 Python。

我需要每 10 秒将数据写入我的 file.txt，请问我该怎么做？我在这里找到了类似的主题 (Run certain code every n seconds)，但我不知道如何让它在我当前的场景中工作。

#!/usr/bin/env python3

import argparse
import re
import logging
import sys
import time

from btlewrap import available_backends, BluepyBackend, GatttoolBackend, PygattBackend
from mitemp_bt.mitemp_bt_poller import MiTempBtPoller, \
    MI_TEMPERATURE, MI_HUMIDITY, MI_BATTERY

def valid_mitemp_mac(mac, pat=re.compile(r"4C:65:A8:[0-9A-F]{2}:[0-9A-F]{2}:[0-9A-F]{2}")):
    """Check for valid mac adresses."""
    if not pat.match(mac.upper()):
        raise argparse.ArgumentTypeError('The MAC address "{}" seems to be in the wrong format'.format(mac))
    return mac

def poll(args):
    """Poll data from the sensor."""
    backend = _get_backend(args)
    poller = MiTempBtPoller(args.mac, backend)
    line1 = "Temperature: {}".format(poller.parameter_value(MI_TEMPERATURE))
    line2 = "Humidity: {}".format(poller.parameter_value(MI_HUMIDITY))
    print("Getting data from Mi Temperature and Humidity Sensor")
    print("FW: {}".format(poller.firmware_version()))
    print("Name: {}".format(poller.name()))
    print("Battery: {}".format(poller.parameter_value(MI_BATTERY)))
    print(line1)
    print(line2)
    f = open('file.txt', 'w')
    f.write("%s \n %s \n" % (line1, line2))
    f.close()

def _get_backend(args):
    """Extract the backend class from the command line arguments."""
    if args.backend == 'gatttool':
        backend = GatttoolBackend
    elif args.backend == 'bluepy':
        backend = BluepyBackend
    elif args.backend == 'pygatt':
        backend = PygattBackend
    else:
        raise Exception('unknown backend: {}'.format(args.backend))
    return backend


def list_backends(_):
    """List all available backends."""
    backends = [b.__name__ for b in available_backends()]
    print('\n'.join(backends))

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--backend', choices=['gatttool', 'bluepy', 'pygatt'], default='gatttool')
    parser.add_argument('-v', '--verbose', action='store_const', const=True)
    subparsers = parser.add_subparsers(help='sub-command help', )

    parser_poll = subparsers.add_parser('poll', help='poll data from a sensor')
    parser_poll.add_argument('mac', type=valid_mitemp_mac)
    parser_poll.set_defaults(func=poll)

    parser_scan = subparsers.add_parser('backends', help='list the available backends')
    parser_scan.set_defaults(func=list_backends)

    args = parser.parse_args()
    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)

    if not hasattr(args, "func"):
        parser.print_help()
        sys.exit(0)

    args.func(args)


if __name__ == '__main__':
    main()

Answer 1

您可以使用 time 模块在每次迭代时暂停程序 10 秒；

from time import sleep

def func(n):
    print(n+1)

for i in range(5):
    func(i)
    sleep(10)

>1
>2
>3
>4
>5

# (every 10 seconds)

然而，这将阻塞程序的其余部分运行ning，尽管调用写入函数的简单多线程脚本就足够了。

关于您正在使用的代码，在 poll 函数中插入 sleep 调用并包装您的代码。如果你想循环程序 10 次那么;

def poll(args):
    """Poll data from the sensor."""
    for _ in range(10):
        # code things
        f = open('file.txt', 'a') # << Use append here or you will keep overwriting file contents
        f.write('hello')
        f.close()
        sleep(10)

或者如果您希望它永远运行直到您 KeyboardInterrupt 或以某种方式退出：

def poll(args):
    """Poll data from the sensor."""
    while True:
        # code things
        f = open('file.txt', 'a') # << Use append here or you will keep overwriting file contents
        f.write('hello')
        f.close()
        sleep(10)

Answer 2

您需要某种循环来轮询您的传感器 - 我没有看到任何人浏览您的代码。您在 JS 中也有 while 和 for 循环 - 如果您不确定语法，请在 http://docs.python.org/3/tutorial 中查找它们。
存储你写入变量的时间，睡一会儿轮询下一个值，检查是否超过 10 秒，如果有则写入，否则不写入。（或者如果您不想打印中间值，则只需在轮询之间休眠 10 秒

阅读有关循环的信息：

import time

def poll():
    return time.time(), 42

last_write = None    # when did we record to file last?

# loop for as long as you want - while True would loop endlessly
for _ in range(7):
    t,c = poll()     # call poll() to get time and temperature from mocked data reader

    # check if enough time has passed
    if last_write is None or (t-last_write) > 2:  # check if new reading needed
        with open("t.txt","a") as f:
            f.write(f"{t}    {c}\n")
        last_write=t
        print("in file ", t,c)            
    else:
        print("just output ", t,c)
    time.sleep(0.7)                               # sleep some

输出：

in file  1552978725.5224085 42                # ...25.5
just output  1552978726.2232893 42            # ...26.2  - not 2s passed 
just output  1552978726.9241226 42            # ...26.9  - not 2s passed
in file  1552978727.6249442 42                # ...27.6  - 2.1s passed 
just output  1552978728.3259027 42            # ...28.3  - not 2s passed
just output  1552978729.0267787 42            # ...29.0  - not 2s passed 
in file  1552978729.7275977 42                # ...29.7  - 2.1s passed

更多备注：

使用 with open(filename,mode) as f: 并确定其下方的文件操作范围 - 它会在确定范围后自动关闭您的文件，并通过关闭文件句柄来处理异常。
使用模式 "w" 将在写入文件之前截断文件 - 您可能想改用追加：reading and writing files

如何每10秒将数据写入文件

How to write data to a file every 10 seconds

python

raspberry-pi3