import sys import time import socket import random import logging import threading import selectors import collections import configparser import os, inspect # Add parent dir to PATH to import messaging_lib current_dir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe()))) parent_dir = os.path.dirname(current_dir) sys.path.insert(0, parent_dir) import messaging_lib as messaging random.seed() logging.basicConfig( # TODO all prints as logs level=logging.DEBUG, format="%(asctime)s [%(name)-7.7s] [%(threadName)-19.19s] [%(levelname)-7.7s] %(message)s", handlers=[ logging.FileHandler("server_logs.log"), logging.StreamHandler() ]) ConfigOption = collections.namedtuple("ConfigOption", ["section", "option", "value"]) class Server: BUFFER_SIZE = 1024 def __init__(self, server_id=None, config_path="server_config.ini", on_stop=None): self.id = server_id if server_id else str(random.randint(0, 9999)).zfill(4) self.time_started = 0 self.on_stop = on_stop # Init socket self.sel = selectors.DefaultSelector() self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.host = socket.gethostname() self.ip = Server.get_ip_address() # Init configs self.config_path = config_path self.config = configparser.ConfigParser() self.load_config() # Init threads self.autoconnect_thread = threading.Thread(target=self._client_processor, daemon=True, name='Client processor') self.client_processor_thread_running = threading.Event() # Can be used for manual thread killing self.broadcast_thread = threading.Thread(target=self._ip_broadcast, daemon=True, name='IP broadcast sender') self.broadcast_thread_running = threading.Event() self.listener_thread = threading.Thread(target=self._broadcast_listen, daemon=True, name='IP broadcast listener') self.listener_thread_running = threading.Event() def load_config(self): self.config.read(self.config_path) self.port = int(self.config['SERVER']['port']) # TODO try, init def self.broadcast_port = int(self.config['SERVER']['broadcast_port']) self.BROADCAST_DELAY = int(self.config['SERVER']['broadcast_delay']) Server.BUFFER_SIZE = int(self.config['SERVER']['buffer_size']) self.USE_NTP = self.config.getboolean('NTP', 'use_ntp') self.NTP_HOST = self.config['NTP']['host'] self.NTP_PORT = int(self.config['NTP']['port']) def start(self): # do_auto_connect=True, do_ip_broadcast=True, do_listen_broadcast=False self.time_started = time.time() logging.info("Starting server with id: {} on {}:{} !".format(self.id, self.ip, self.port)) logging.info("Starting server socket!") self.server_socket.bind((self.ip, self.port)) logging.info("Starting client processor thread!") self.client_processor_thread_running.set() self.autoconnect_thread.start() logging.info("Starting broadcast sender thread!") self.broadcast_thread_running.set() self.broadcast_thread.start() logging.info("Starting broadcast listener thread!") self.listener_thread_running.set() self.listener_thread.start() def stop(self): logging.info("Stopping server") self.client_processor_thread_running.clear() self.broadcast_thread_running.clear() self.listener_thread_running.clear() self.server_socket.close() self.sel.close() logging.info("Server stopped") if self.on_stop is not None: self.on_stop() sys.exit("Stopped") @staticmethod def get_ip_address(): try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as ip_socket: ip_socket.connect(("8.8.8.8", 80)) return ip_socket.getsockname()[0] except OSError: logging.warning("No network connection detected, starting on localhost") return "localhost" @staticmethod def get_ntp_time(ntp_host, ntp_port): NTP_DELTA = 2208988800 # 1970-01-01 00:00:00 NTP_QUERY = b'\x1b' + bytes(47) with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as ntp_socket: ntp_socket.sendto(NTP_QUERY, (ntp_host, ntp_port)) msg, _ = ntp_socket.recvfrom(1024) return int.from_bytes(msg[-8:], 'big') / 2 ** 32 - NTP_DELTA def time_now(self): if self.USE_NTP: timenow = self.get_ntp_time(self.NTP_HOST, self.NTP_PORT) else: timenow = time.time() return timenow # noinspection PyArgumentList def _client_processor(self): logging.info("Client processor (selector) thread started!") self.server_socket.listen() self.server_socket.setblocking(False) self.sel.register(self.server_socket, selectors.EVENT_READ | selectors.EVENT_WRITE, data=None) while self.client_processor_thread_running.is_set(): events = self.sel.select(timeout=0) for key, mask in events: if key.data is None: self._connect_client(key.fileobj) else: client = key.data try: client.process_events(mask) except Exception as error: logging.error("Exception {} occurred for {}! Resetting connection!".format(error, client.addr)) client.close() logging.info("Client autoconnect thread stopped!") def _connect_client(self, sock): conn, addr = sock.accept() logging.info("Got connection from: {}".format(str(addr))) conn.setblocking(False) if not any(client_addr == addr[0] for client_addr in Client.clients.keys()): client = Client(addr[0]) logging.info("New client") else: client = Client.clients[addr[0]] logging.info("Reconnected client") self.sel.register(conn, selectors.EVENT_READ, data=client) client.connect(self.sel, conn, addr) def _ip_broadcast(self): logging.info("Broadcast sender thread started!") msg = messaging.MessageManager.create_simple_message( "server_ip", {"host": self.ip, "port": str(self.port), "id": self.id, "start_time": str(self.time_started)}) broadcast_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) broadcast_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) broadcast_sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) logging.info("Formed broadcast message: {}".format(msg)) time.sleep(self.BROADCAST_DELAY) try: while self.broadcast_thread_running.is_set(): broadcast_sock.sendto(msg, ('255.255.255.255', self.broadcast_port)) logging.debug("Broadcast sent") time.sleep(self.BROADCAST_DELAY) finally: broadcast_sock.close() logging.info("Broadcast sender thread stopped, socked closed!") def _broadcast_listen(self): logging.info("Broadcast listener thread started!") broadcast_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) broadcast_client.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) try: broadcast_client.bind(("", self.broadcast_port)) except OSError: logging.critical("Another server is running on this computer, shutting down!") # TODO popup and as function self.stop() try: while self.listener_thread_running.is_set(): data, addr = broadcast_client.recvfrom(1024) # TODO nonblock message = messaging.MessageManager() message.income_raw = data message.process_message() if message.content: if message.content["command"] == "server_ip": if message.content["args"]["id"] != str(self.id) \ and float(message.content["args"]["start_time"]) <= self.time_started: # younger server should shut down logging.critical("Another server detected over the network, shutting down!") # TODO popup self.stop() else: logging.warning("Got wrong broadcast message from {}".format(addr)) finally: broadcast_client.close() logging.info("Broadcast listener thread stopped, socked closed!") def send_starttime(self, copter, dt=0): timenow = self.time_now() print('Now:', time.ctime(timenow), "+ dt =", dt) copter.send_message("start", {"time": str(timenow + dt)}) def requires_connect(f): def wrapper(*args, **kwargs): if args[0].connected: return f(*args, **kwargs) else: logging.warning("Function requires client to be connected!") return wrapper def requires_any_connected(f): def wrapper(*args, **kwargs): if Client.clients: return f(*args, **kwargs) else: logging.warning("No clients were connected!") return wrapper class Client(messaging.ConnectionManager): clients = {} on_connect = None # Use as callback functions on_first_connect = None on_disconnect = None def __init__(self, ip): super(Client, self).__init__() self.copter_id = None self.connected = False self.clients[ip] = self @staticmethod def get_by_id(copter_id): for client in Client.clients.values(): if client.copter_id == copter_id: return client def connect(self, client_selector, client_socket, client_addr): logging.info("Client connected") if not self.resume_queue: self._send_queue = collections.deque() super(Client, self).connect(client_selector, client_socket, client_addr) self.connected = True if self.copter_id is None: self.get_response("id", self._got_id) if self.on_connect: self.on_connect(self) def _got_id(self, value): logging.info("Got copter id: {} for client {}".format(value, self.addr)) self.copter_id = value if Client.on_first_connect: Client.on_first_connect(self) def close(self): self.connected = False if Client.on_disconnect: Client.on_disconnect(self) super(Client, self).close() @requires_connect def _send(self, data): super(Client, self)._send(data) logging.debug("Queued data to send: {}".format(data)) def send_config_options(self, *options: ConfigOption, reload_config=True): logging.info("Sending config options: {} to {}".format(options, self.addr)) sending_options = [{'section': option.section, 'option': option.option, 'value': option.value} for option in options] print(sending_options) self.send_message( 'config_write', {"options": sending_options, "reload": reload_config} ) @staticmethod @requires_any_connected def broadcast(message, force_all=False): for client in Client.clients.values(): if client.connected or force_all: client._send(message) @classmethod @requires_any_connected def broadcast_message(cls, command, args=None, force_all=False): cls.broadcast(messaging.MessageManager.create_simple_message(command, args), force_all) if __name__ == '__main__': server = Server() server.start() while True: pass