1

我有一个从 USB 气象站获取数据的 python 脚本，现在只要从气象站接收到数据，它就会将数据放入 MySQL。

我有一个带有插入函数的 MySQL 类，我想要该函数检查它是否在过去 5 分钟内运行过，如果有，退出。

在互联网上找不到任何执行此操作的代码。

也许我需要一个子流程，但我一点也不熟悉。

有人有我可以使用的例子吗？

4

5 回答 5

2

使用这个超时装饰器。

import signal

class TimeoutError(Exception):
    def __init__(self, value = "Timed Out"):
        self.value = value
    def __str__(self):
        return repr(self.value)

def timeout(seconds_before_timeout):
    def decorate(f):
        def handler(signum, frame):
            raise TimeoutError()
        def new_f(*args, **kwargs):
            old = signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds_before_timeout)
            try:
                result = f(*args, **kwargs)
            finally:
                signal.signal(signal.SIGALRM, old)
            signal.alarm(0)
            return result
        new_f.func_name = f.func_name
        return new_f
    return decorate

用法：

import time

@timeout(5)
def mytest():
    print "Start"
    for i in range(1,10):
        time.sleep(1)
        print "%d seconds have passed" % i

if __name__ == '__main__':
    mytest()
于 2012-04-28T18:44:05.877 回答
1 
import time

def timeout(f, k, n):
    last_time = [time.time()]
    count = [0]
    def inner(*args, **kwargs):
        distance = time.time() - last_time[0]
        if distance > k:
            last_time[0] = time.time()
            count[0] = 0
            return f(*args, **kwargs)
        elif distance < k and (count[0]+1) == n:
            return False
        else:
            count[0] += 1
            return f(*args, **kwargs)
    return inner

timed = timeout(lambda x, y : x + y, 300, 1)

print timed(2, 4)

第一个参数是你要运行的函数，第二个是时间间隔，第三个是允许在那个时间间隔内运行的次数。

于 2012-04-28T19:01:21.913 回答
1

可能是最直接的方法（如果您愿意，可以将其放入装饰器中，但我认为这只是化妆品）：

import time
import datetime

class MySQLWrapper:
  def __init__(self, min_period_seconds):
    self.min_period = datetime.timedelta(seconds=min_period_seconds)
    self.last_calltime = datetime.datetime.now() - self.min_period
  def insert(self, item):
    now = datetime.datetime.now()
    if now-self.last_calltime < self.min_period:
      print "not insert"
    else:
     self.last_calltime = now
     print "insert", item

m = MySQLWrapper(5)
m.insert(1) # insert 1
m.insert(2) # not insert
time.sleep(5)
m.insert(3) # insert 3

附带说明：您在网络搜索相关内容时是否注意到RRDTool ？它显然可以实现您想要实现的目标，即

  • 存储任意分辨率/更新频率的最新值的数据库。
  • 如果更新过于频繁或缺失，则外推/内插值。
  • 从数据中生成图表。

一种方法可能是将所有可以获取的数据存储到 MySQL 数据库中，并将子集转发到此类 RRDTool 数据库，以生成漂亮的时间序列可视化。取决于你可能需要什么。

于 2012-04-28T19:27:02.693 回答
0

每次运行该函数时，都会使用当前时间保存一个文件。再次运行该函数时，检查文件中存储的时间并确保它足够旧。

于 2012-04-28T18:41:21.493 回答
0

只需派生到一个新类并覆盖插入函数。在覆盖函数中，检查最后插入时间，如果超过五分钟就调用父亲的插入方法，当然更新最近的插入时间。

于 2012-04-28T18:43:04.043 回答