Async server that broadcast's client's location

server.py

import asyncio
import logging
import os
from pathlib import Path


class LocationShareServer:
    def __init__(self, host: str, port: int):
        """
        Initializes a new instance of the LocationShareServer class.

        Parameters:
            host (str): The hostname or IP address to bind the server to.
            port (int): The port number to bind the server to.
        """
        self.__host: str = host
        self.__port: int = port
        self.__logger: logging.Logger = self.initialize_logger()
        self._clients: dict = {}

    @property
    def host(self) -> str:
        """
        Returns the hostname or IP address the server is bound to.
        """
        return self.__host

    @property
    def port(self) -> int:
        """
        Returns the port number the server is bound to.
        """
        return self.__port

    @property
    def logger(self) -> logging.Logger:
        """
        Returns the logger for the server.
        """
        return self.__logger

    def initialize_logger(self) -> logging.Logger:
        """
        Initializes the logger for the server.

        Returns:
            logging.Logger: The initialized logger.
        """
        path: Path = Path(os.path.join(os.getcwd(), "logs"))
        path.mkdir(parents=True, exist_ok=True)

        logger = logging.getLogger("Location Share Server")
        logger.setLevel(logging.DEBUG)

        file_handler = logging.FileHandler("./logs/location_share_server.log")

        file_handler.setLevel(logging.DEBUG)

        formatter = logging.Formatter("[%(asctime)s] - [%(levelname)s] - %(message)s",
                                      "%Y-%m-%d %H:%M:%S")

        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

        return logger

    async def start(self):
        """
        Starts the server.
        """
        try:
            server = await asyncio.start_server(self.handle_client, self.host, self.port)
            async with server:
                self.logger.info("Connection: Launching Server")
                await server.serve_forever()
        except Exception as e:
            self.logger.error(e)

    async def handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        """
        Handles a new client connection.

        Parameters:
            reader (asyncio.StreamReader): The reader for the client connection.
            writer (asyncio.StreamWriter): The writer for the client connection.
        """
        address, port = writer.get_extra_info("peername")
        identifier: str = f"{address}:{port}"
        self._clients[identifier] = (reader, writer)

        self.logger.info(f"Connection: {identifier} connected to the server")
        self.logger.info(
            f"Connection: {list(self._clients.keys())} are the list of clients connected to the server")
        await self.recieve_location(reader, identifier)

    async def recieve_location(self, reader: asyncio.StreamReader, identifier: str):
        """
        Receives location data from the client.

        Parameters:
            reader (asyncio.StreamReader): The reader for the client connection.
            identifier (str): The identifier for the client connection.
        """
        try:
            while True:
                location_byte: bytes = await reader.read(256)
                location: str = location_byte.decode().strip()
                if location:
                    self.logger.info(
                        f"Recieved: {identifier} -- {location} recieved by the server")

                    await self.broadcast_location(identifier, location)
                else:
                    break
        except Exception as e:
            self.logger.error(e)
        finally:
            self.logger.info(
                f"Connection: {identifier} is disconnected from the server")
            del self._clients[identifier]

    async def broadcast_location(self, sender: str, location: str):
        """
        Broadcasts the location data to all connected clients.

        Parameters:
            sender (str): The identifier for the client sending the location data.
            location (str): The location data to be broadcasted.
        """
        for address, (_, writer) in self._clients.items():
            if sender != address:
                try:
                    data: str = f"{sender} -- {location}"
                    self.logger.info(
                        f"Broadcasted: {data} is broacasted to all clients")
                    writer.write(data.encode())
                    await writer.drain()
                except Exception as e:
                    self.logger.error(e)


if __name__ == "__main__":
    # firewall-cmd --zone=public --permanent --add-port=5555/tcp
    # firewall-cmd --reload
    # sudo timedatectl set-timezone Asia/Kolkata
    
    location_share_server = LocationShareServer("", 5555)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(location_share_server.start())