Передать переменную из одного класса в другой
пытаюсь реализовать обмен данными между сервером и клиентом, хочу уведомлять клиента всякий раз, когда происходят изменения в файлах .txt, используя модуль watchdog.
Подскажите, как передать переменную my_event_handler из класса EventHandler классу Server в метод send_data()? После выполнения кода получаю ошибку TypeError: a bytes-like object is required, not 'socket'
from Socket import Socket
import threading
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler, PatternMatchingEventHandler, FileSystemEventHandler
global my_event_handler
class EventHandler(FileSystemEventHandler): # класс обработчика событий
def __init__(self):
patterns = ["*.txt"]
ignore_patterns = None
ignore_directories = True
case_sensitive = True
self.my_event_handler = PatternMatchingEventHandler(
patterns,
ignore_patterns,
ignore_directories,
case_sensitive
)
def on_modified(self, event):
self.my_event_handler = f"Your file has {event.src_path} been modified"
return self.my_event_handler
class Server(Socket):
def __init__(self):
super(Server, self).__init__()
print("...........Server is starting...........")
self.users = []
def set_up(self):
self.bind(("127.0.0.1", 6006))
self.listen(5)
self.accept_sockets()
def send_data(self, my_event_handler):
for user in self.users:
while True:
user.send(my_event_handler).encode("utf-8")
def accept_sockets(self):
while True:
user_socket, address = self.accept()
print(f"User <{address[0]}> connected!")
self.users.append(user_socket)
sent_accepted_user = threading.Thread(
target=self.send_data,
args=(user_socket,)
)
sent_accepted_user.start()
sent_accepted_user.join()
if __name__ == "__main__":
server = Server()
server.set_up()
event_handler = EventHandler()
observer = Observer()
observer.schedule(event_handler, path='.', recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
Код клиента выглядит следующим образом
from Socket import Socket
from threading import Thread
class Client(Socket):
def __init__(self):
super(Client, self).__init__()
print("Connecting.............")
def set_up(self):
self.connect(("127.0.0.1", 6006))
listen_thread = Thread(target=self.listen_socket)
listen_thread.start()
def listen_socket(self, listened_socket=None):
print("...........Listening server...........")
while True:
data = self.recv(4096)
print(data.decode("utf-8"))
if __name__ == '__main__':
client = Client()
client.set_up()
Отдельный файл Socket с настройками
import socket
class Socket(socket.socket):
def __init__(self):
super(Socket, self).__init__(
socket.AF_INET,
socket.SOCK_STREAM,
)
def send_data(self, my_event_handler):
raise NotImplementedError()
def listen_socket(self, listened_socket=None):
raise NotImplementedError()
def set_up(self):
raise NotImplementedError()