Cách đặt khoảng thời gian trong python

Tất cả chúng ta đều viết mã JavaScript, tôi chắc chắn, tôi làm tốt và tôi thích có thể chạy một chức năng trên một Khoảng thời gian trong JavaScript;

Ảnh của Veri Ivanova trên Bapt

ý tưởng thật đơn giản. Bạn muốn chạy một chức năng trong khoảng thời gian hẹn giờ với Python nhưng điều tốt nhất bạn có thể làm là vòng lặp while trong đó mã của bạn ngủ một lúc. Điều này hoạt động nhưng nó không phải là thanh lịch

Bây giờ chúng ta có thể sử dụng Khoảng thời gian Pythonic thực

Đây là lớp học làm nên điều kỳ diệu này

from threading import Timer

class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.interval   = interval
        self.function   = function
        self.args       = args
        self.kwargs     = kwargs
        self.is_running = False
        self.start()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

Và đây là cách mã này được sử dụng

from time import sleep

    d = {'__count__': 0, '_message__': None}
    def hello(dd):
        dd['__count__'] = dd.get('__count__', 0) + 1
        print("Hello %s! (%d)" % (dd.get('__message__'), dd.get('__count__', 0)))

    print("starting...")
    d['__message__'] = 'World!'
    rt = RepeatedTimer(1, hello, d) # it auto-starts, no need of rt.start()
    try:
        sleep(30) # your long-running job goes here...
    finally:
        rt.stop() # better in a try/finally block to make sure the program ends!

Mã này đưa ra một thông báo kèm theo bộ đếm để cho biết rằng nó đang hoạt động. Lưu ý cách biến toàn cục được bao bọc trong một lệnh chính tả vì đây là cách bạn có thể cập nhật toàn cầu theo cách an toàn cho luồng;

Bây giờ tôi có thể đặt mã này vào thư viện của riêng mình và sau đó cập nhật Tweepy Twitter Bot của mình để nó có thể trở nên thanh lịch hơn một chút bằng cách thực hiện một số lịch trình thời gian thực thay vì lặp lại các khoảng thời gian ngủ. Có thể mất một lúc trước khi tôi bắt đầu cập nhật Bot Twitter đó bằng mã này, vì vậy hãy chú ý theo dõi

Điều thú vị về giải pháp này là trong lĩnh vực Lập trình hàm bởi vì nếu bạn có thể đặt một hàm trong Khoảng thời gian thì bạn cũng có thể lên lịch cho các tác vụ chạy một lần mỗi khi chúng được lên lịch thay vì sử dụng Khoảng thời gian.

Hai phương pháp để chạy một lần

Chạy một lần

from threading import Timer

class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.interval   = interval
        self.function   = function
        self.args       = args
        self.kwargs     = kwargs
        self.is_running = False
        self.start()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(self, *self.args, **self.kwargs)
        if (self.kwargs.get('run_once', False)):
            self.stop()

    def start(self):
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

Ở đây, chúng tôi chuyển “run_once=” kwarg cho tác vụ và khung xử lý việc chấm dứt khoảng thời gian

Ngoài ra, hãy chú ý đến "bản thân" mà chúng ta chuyển sang nhiệm vụ để nó có thể xử lý số phận của chính mình như sau

from time import sleep

    def hello(rt, dd, **kwargs):
        dd['__count__'] = dd.get('__count__', 0) + 1
        print("Hello %s! (%d)" % (dd.get('__message__'), dd.get('__count__', 0)))
        rt.stop()


    d = {'__count__': 0, '_message__': None}
    for i in range(0, 10):
        d['__message__'] = 'World!'
        rt = RepeatedTimer(i*2, hello, d) # it auto-starts, no need of rt.start()

    try:
        sleep(30) # your long-running job goes here...
    finally:
        print('Shutting down.')

Bây giờ tác vụ có thể tự kết thúc để nó có thể trở thành tác vụ chạy một lần

Điều này hữu ích cho các tình huống mà bạn muốn lên lịch cho một nhiệm vụ xảy ra vào một thời điểm nào đó trong tương lai nhưng bạn cũng muốn nhiệm vụ lên lịch cho một nhiệm vụ khác dựa trên một số tiêu chí, nghĩa là đây không còn là một khoảng thời gian nữa.

Vì vậy, bây giờ bạn có thể sử dụng tham số “run_once=” có thể là Đúng hoặc Sai hoặc bất kỳ thứ gì mà bộ thực thi Python có thể lấy hoặc bạn có thể sử dụng thể hiện đối tượng để thực hiện chức năng tương tự