python: Windows 相当于 SIGALRM

Question

我有这个装饰器：

def timed_out(timeout):
    def decorate(f):
        if not hasattr(signal, "SIGALRM"):
            return f

        def handler(signum, frame):
            raise TimedOutExc()

        @functools.wraps(f)
        def new_f(*args, **kwargs):
            old = signal.signal(signal.SIGALRM, handler)
            signal.alarm(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

该代码只在 Linux 上执行任何操作，但与 Windows 上一样，没有 SIGALRM。让这段代码在 Windows 中也能工作的最简单方法是什么？

Answer 1

它不是很漂亮，但我必须以跨平台的方式做类似的事情，并且我想出了使用单独的线程。基于信号的系统并不能在所有平台上可靠地工作。

此类的使用可以包装在装饰器中，或者制作成 with 上下文处理程序。

YMMV。

#!/usr/bin/env python2.7
import time, threading

class Ticker(threading.Thread):
  """A very simple thread that merely blocks for :attr:`interval` and sets a
  :class:`threading.Event` when the :attr:`interval` has elapsed. It then waits
  for the caller to unset this event before looping again.

  Example use::

    t = Ticker(1.0) # make a ticker
    t.start() # start the ticker in a new thread
    try:
      while t.evt.wait(): # hang out til the time has elapsed
        t.evt.clear() # tell the ticker to loop again
        print time.time(), "FIRING!"
    except:
      t.stop() # tell the thread to stop
      t.join() # wait til the thread actually dies

  """
  # SIGALRM based timing proved to be unreliable on various python installs,
  # so we use a simple thread that blocks on sleep and sets a threading.Event
  # when the timer expires, it does this forever.
  def __init__(self, interval):
    super(Ticker, self).__init__()
    self.interval = interval
    self.evt = threading.Event()
    self.evt.clear()
    self.should_run = threading.Event()
    self.should_run.set()

  def stop(self):
    """Stop the this thread. You probably want to call :meth:`join` immediately
    afterwards
    """
    self.should_run.clear()

  def consume(self):
    was_set = self.evt.is_set()
    if was_set:
      self.evt.clear()
    return was_set

  def run(self):
    """The internal main method of this thread. Block for :attr:`interval`
    seconds before setting :attr:`Ticker.evt`

    .. warning::
      Do not call this directly!  Instead call :meth:`start`.
    """
    while self.should_run.is_set():
      time.sleep(self.interval)
      self.evt.set()

Answer 2

这个答案仅适用于 Cygwin..

...并且它可能不再适用于最新版本的 Cygwin...

我发现这个超时装饰器代码也非常方便。 （我最初在这个问题的答案中找到它：如何限制函数调用的执行时间？）

为了使其在 Windows 上运行，我使用随 Cygwin 一起安装的 Python。

我运行 setup-x86_64.exe，然后从 Python 文件夹中选择 python3 包。 （或者，如果您更喜欢 Python 2，则使用 python 包。）

定义了别名

alias python=python3

为了将 python3 重命名为 python2，我从 Cygwin 命令提示符 。由于我不经常使用此功能，因此我可能不会将其放入 .bashrc 或其他任何内容中。

相关问题：
Python 信号甚至在 Cygwin 上也不起作用？

Answer 3

解决原始问题：python：Windows 相当于 SIGALRM

有一种方法可以在 Windows 上创建 SIGALRM 类型事件。它利用多媒体定时器来触发事件。这是一个通用计时器，可以触发单个事件 (periodic=False)、多次重复 (n_reps >= 1) 或通过外部停止连续触发事件（.stop() 函数调用）

可以指定此计时器来生成间隔非常精确的事件，这些事件不依赖于 Python 运行时。此处，时间间隔以秒为单位指定（间隔和分辨率）

import ctypes as ct
from ctypes.wintypes import UINT, DWORD
from typing import Callable, Optional

class mmPeriodicTimer:
    """Periodic timer based upon the Windows Multimedia library
    This produces a timed callback with high precision ( ~ < 1ms difference)
    executed outside of a threaded structure (based on Windows runtime ).
    The advantage of this is that the Python GIL limitation is avoided,
    and an aligned "lag" time between all Python processes is not generated;

    The disadvantage is that since this is thread independent, the callback
    procedure must complete its tasks before the next callback is executed 
    otherwise collisions may occur

    This is based on the example:
    https://stackoverflow.com/questions/10717589/how-to-implement-high-speed-consistent-sampling
    """

    timeproc = ct.WINFUNCTYPE(None, ct.c_uint, ct.c_uint, DWORD, DWORD, DWORD)
    timeSetEvent = ct.windll.winmm.timeSetEvent
    timeKillEvent = ct.windll.winmm.timeKillEvent

    def Tick(self):
        self.i_reps += 1
        self.tickFunc()

        if not self.periodic:
            self.stop()

        if self.n_reps is not None and self.i_reps >= self.n_reps:
            self.stop()

    def CallBack(self, uID, uMsg, dwUser, dw1, dw2):
        if self.running:
            self.Tick()

    def __init__(
        self,
        interval: int,
        tickfunc: Callable,
        resolution: Optional[int] = 0,
        stopFunc: Optional[Callable] = None,
        periodic: Optional[bool] = True,
        n_reps: Optional[int] = None,
    ):
        self.interval = UINT(int(interval * 1000))
        self.resolution = UINT(int(resolution * 1000))
        self.tickFunc = tickfunc
        self.stopFunc = stopFunc
        self.periodic = periodic
        self.n_reps = n_reps
        if not self.periodic and self.n_reps is not None:
            raise ValueError("n_reps must be None if periodic is False")

        self.i_reps = 0
        self.id = None
        self.running = False
        self.calbckfn = self.timeproc(self.CallBack)

    def start(self, instant: bool = False):
        if not self.running:
            self.running = True
            if instant:
                self.Tick()

            self.id = self.timeSetEvent(
                self.interval,
                self.resolution,
                self.calbckfn,
                ct.c_ulong(0),
                ct.c_uint(self.periodic),
            )

    def stop(self):
        if self.running:
            self.timeKillEvent(self.id)
            self.running = False

            if self.stopFunc:
                self.stopFunc()

---

单个事件示例：

import time
from datetime import datetime

def my_func():
    print("current time=", datetime.now())

t = mmPeriodicTimer(interval=1, tickfunc=my_func, periodic=False)
t.start()

输出：
当前时间= 2024-03-24 22:42:16.475416

---

多个事件示例（然后停止）：

t = mmPeriodicTimer(interval=1, tickfunc=my_func, n_reps=5)
t.start()

输出：

current time= 2024-03-24 22:42:11.474459
current time= 2024-03-24 22:42:12.475201
current time= 2024-03-24 22:42:13.474462
current time= 2024-03-24 22:42:14.475113
current time= 2024-03-24 22:42:15.474802

---

给定时间后手动终止示例：

t = mmPeriodicTimer(interval=1, tickfunc=my_func, periodic=True)
t.start()
time.sleep(5)
t.stop()

输出：

current time= 2024-03-24 22:46:32.864423
current time= 2024-03-24 22:46:33.865319
current time= 2024-03-24 22:46:34.864554
current time= 2024-03-24 22:46:35.864762
current time= 2024-03-24 22:46:36.864778

---

原始信用归于：
如何实现高速、一致的采样？