Hệ thống giám sát với các sự kiện trong Python

Question

Tôi đang tạo một dự án bằng Python và tôi muốn thêm một hệ thống giám sát sử dụng các sự kiện và trình xử lý sự kiện. Tôi muốn hệ thống này có sẵn trong suốt dự án. Tôi có các hoạt động sau đây trong đầu:

• Xác định sự kiện. Sự kiện có thể lấy một số dữ liệu làm tham số.
• Xác định màn hình. Một màn hình đăng ký cho một sự kiện nào đó. Nhiều màn hình có thể đăng ký cho cùng một sự kiện. Tôi muốn tạo các loại màn hình khác nhau, ví dụ: một để in dữ liệu, một để tạo lô với dữ liệu, vv Do đó một màn hình phải là một lớp, có khả năng giữ tất cả dữ liệu mà nó thu thập cho đến khi một số phương thức (ví dụ: in, tạo nhật ký, ...) được gọi.
• Xác định trình xử lý sự kiện cho cặp sự kiện theo dõi. Điều này xác định cách màn hình cho trước sẽ trả lời cho sự kiện đã cho. Hành động này chủ yếu sẽ là: thêm dữ liệu này vào danh sách dữ liệu của cá thể của một số lớp giám sát.
• Chức năng thông báo có thể thông báo khi có sự kiện xảy ra. Điều này sẽ kích hoạt trình xử lý sự kiện cho tất cả các màn hình được đăng ký cho sự kiện đó. Lý tưởng nhất, chức năng thông báo phải được gọi từ bất cứ nơi nào trong dự án.

Làm cách nào để tạo một hệ thống như vậy? Có thư viện nào có thể giúp tôi với điều này không? Tôi đặc biệt tự hỏi làm thế nào tôi có thể làm cho hệ thống này theo cách như vậy mà nó là minh bạch có sẵn trong suốt dự án.

Answer 1

Bạn có thể làm hầu hết những gì bạn muốn với bốn mươi dòng mã Python. Đây là thiết kế của riêng tôi mà tôi sử dụng mọi lúc. Các tên hàm được chọn để làm cho nó trở thành một sự thay thế cho các "tín hiệu" và "khe" của Qt.

Rất dễ sử dụng. Bạn tạo một PSignal. Bạn đăng ký trình xử lý bằng cách gọi phương thức connect. Một trình xử lý có thể được gọi bất kỳ. Khi một sự kiện xảy ra, bạn phát ra một tín hiệu (tức là, thông báo cho một sự kiện) bằng cách gọi hàm emit. Mỗi lần chạy được đăng ký có thể gọi tại thời điểm đó. Các đối tượng gọi emit không biết, hoặc quan tâm, cho dù bất cứ ai đang lắng nghe hoặc những gì sẽ xảy ra nếu họ đang có.

Bạn cũng có thể ngắt kết nối một trình xử lý.

Có rất nhiều mã gỡ lỗi vì tôi phát hiện ra rằng nếu không một số lỗi nhất định có thể khó theo dõi.

Trong câu hỏi của bạn, bạn muốn mỗi trình xử lý là một màn hình và trong các trình xử lý thiết kế của tôi chỉ hoạt động. Nhưng có vẻ như với tôi rằng khái niệm "màn hình" của bạn độc lập với cơ chế sự kiện/xử lý. Bạn sẽ phải viết các hàm để làm cho ứng dụng của bạn đi, và nó sẽ khá dễ dàng để làm cho các hàm đó gọi các màn hình của bạn.

Mã được thử nghiệm rộng rãi với Python 3.3.

#! python3 
import traceback 

class PSignal: 
    def __init__(self, debug=False): 
     self.debug = debug 
     self.__handlers = [] 

    def clear(self): 
     """Deletes all the handlers.""" 
     self.__handlers.clear() 

    def connect(self, f): 
     """f is a python function.""" 
     if not callable(f): 
      raise ValueError("Object {!r} is not callable".format(f)) 
     self.__handlers.append(f) 
     if self.debug: 
      print("PSIGNAL: Connecting", f, self.__handlers) 

    def disconnect(self, f): 
     for f1 in self.__handlers: 
      if f == f1: 
       self.__handlers.remove(f) 
       return 

    def emit(self, *x, **y): 
     self._emit(*x, **y) 

    def check_debug(self): 
     if self.debug and self.__handlers: 
      print("PSIGNAL: Signal emitted") 
      traceback.print_stack() 

    def _emit(self, *x, **y): 
     self.check_debug() 
     for f in self.__handlers: 
      try: 
       if self.debug: 
        print("PSIGNAL: emit", f, len(x), x, y) 
       f(*x, **y) 
      except Exception: 
       print("PSIGNAL: Error in signal", f) 
       traceback.print_exc() 

Answer 2

Kiểm tra phản ứng Python (RxPy)

ReactiveX

GitHub/python

Answer 3

Bạn có thể tạo hệ thống của riêng mình bằng hệ thống nhắn tin phân tán như zmq và mẫu "người đăng ký nhà xuất bản".

Tôi đã thực hiện một cái gì đó như thế xây dựng một động cơ công việc tùy biến (dòng chảy, https://github.com/mastro35/flows)

Bye D.