TCP socket with asyncio

[sishari]

Hello,
Can anybody help me?
I want to start a socket server on pico.
Then multiple clients can connect to server.
Each client can send data to the server.

On the pico is a subsystem connected over uart (the uart communication works)
The pico forward the data from the client to the subsystem.

When i get data from the subsystem, it should be send to all connected clients.

How can i solve this ??

[sishari]

I have treid following but i think it is not correct:

async def serve(reader, writer)

loop = asyncio.get_event_loop()
loop.create_task(asyncio.start_server(serve, "0.0.0.0", 10000))

[sosi-deadeye]

Example of main.py

import asyncio


async def alive_worker():
    """
    Async test function
    """
    while True:
        await asyncio.sleep(10)
        print("Still alive")


async def handler(sr: asyncio.stream.StreamReader, sw: asyncio.StreamWriter):
    """
    Async handler function to handle new connections.
    """
    # https://docs.micropython.org/en/latest/library/asyncio.html#asyncio.Stream
    ip, port = sr.get_extra_info('peername')
    print(f"Got a new connection from {ip}")
    data = await sr.read()
    
    print(f"Data from client {ip}: {data}")
    data = b"Re: " + data
    
    print("Sending data to client")
    # write is a normal function
    sw.write(data)

    # now sending the packet
    # StreamWriter.drain is a coroutine function
    await sw.drain()

    # closing the connection to the client
    # normal function
    sw.close()

    # now waiting for a closed connection
    await sw.wait_closed()
    
    print(f"Connection to client {ip} were closed.")
    print()


async def main():
    """
    Main entry point for asyncio.run
    """
    # https://docs.micropython.org/en/latest/library/asyncio.html#asyncio.start_server
    # start_server is a coroutine and requires await or async with can be used
    server = await asyncio.start_server(handler, "0.0.0.0", 10_000)
    # a new connection is handled by the coroutine handler

    # starting a task which is executed beside this coroutine
    # alive_worker is a coroutine function and
    # create_task expects a coroutine, so it must be called
    # and the result is the argument of create_task
    task = asyncio.create_task(alive_worker())

    while True:
        # sleeps 10 seconds in this coroutine
        # to preventing the return of the coroutine
        await asyncio.sleep(10)


# the boilerplate
if __name__ == "__main__":
    print()
    print("Starting Server")
    print("To send the server a command:")
    try:
        print(f'echo "Hello World" | nc -N {wlan.ifconfig()[0]} 10000')
    except NameError:
        print("No WIFI-Connection")
    print()
    
    
    # calling a coroutine function, returns a coroutine, which is
    # handled by the event loop
    asyncio.run(main())
    # this blocks until the main corotine returns
    # but this is prevented by the loop

My boot.py:

import time
import json
from network import WLAN, STA_IF


try:
    with open("wifi-creds.json") as fd:
        creds = json.load(fd)
except OSError:
    with open("wifi-creds.json", "w") as fd:
        json.dump({"ssid": "", "password": ""}, fd)
else:
    ssid, password = creds["ssid"], creds["password"]
    
    if ssid and password:
        wlan = WLAN(STA_IF)
        wlan.active(False)
        wlan.active(True)
        wlan.connect(ssid, password)
        
        while wlan.ifconfig()[0] == "0.0.0.0":
            time.sleep_ms(100)
            print(".", end="")
        
        print()
        print("Connected")
        print("IP:", wlan.ifconfig()[0])
    else:
        print("ssid and password is not set yet, skipping wifi connection")
        

The boot.py creates a JSON-file named wifi-creds.json. After the file is created, you must edit the file with your WiFi-credentials. This method prevents accidentally positing WIFI-Credentials to the public.

[sishari]

I already had this example.
Here it is received and sent on the same socket
I want to send the same data on all sockets that are connected

[sophiedegran]

Hi
But why not to have a list on your server with all your clients connected as [addr_ip, port_ip] .. ?
That's what I do with microdot to 'broadcast' message
And your refresh this list as you're in async with a simple 'ping' (if you want -> co-routine)
Hope that help

[sosi-deadeye]

I have written some test code to demonstrate it. I'll post it after work.
You should also read this here: https://github.com/peterhinch/micropython-async/tree/master/v3
I have used a Queue as a sender-queue.

[sishari]

Hello,

I wrote a solution myself.
I write each open socket in a list.
In a For loop I write the data to each socket.

But i still need a solution how I can check if a client is still connected to each open socket.
For example, if the network cable is unplugged, the sockets stay open.
Is it possibel to ping with micropythen every client to check if it is connected?

I used this with normal Python but this option doesn't exist in MicroPython with RP pico:
self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
self.s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10)
self.s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 3)
self.s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 4)

[sophiedegran]

Nice!
You can have a look to this nice example from Peter
github.com/peterhinch/micropython-async/blob/master/v3/as_drivers/client_server/userver.py (and uclient.py)
After about your list..
You have to do a simple function to verify if [addr_ip, port_ip]
for x in range (len (_socket_list)): ..
-> verif ip
-> verif port
-> if not add : _socket_list.append([addr_ip, port_ip])
...
Hope that help

[sosi-deadeye]

Another example with normal tcp sockets.

main.py

try:
    from primitives.queue import Queue, QueueEmpty
except ImportError:
    import mip

    mip.install("github:peterhinch/micropython-async/v3/primitives")
    mip.install("github:peterhinch/micropython-async/v3/threadsafe")

import asyncio


class async_closing:
    def __init__(self, stream: asyncio.StreamWriter):
        self.stream = asyncio.StreamWriter

    async def __aenter__(self):
        return self.stream

    async def __aexit__(self, *args):
        self.stream.close()
        await self.stream.closed()


class ChatServer:
    def __init__(self, ip: str, port: int):
        self.ip = ip
        self.port = port
        self.clients = []
        self.server = None
        self._last_id = 0

    @property
    def next_client_id(self):
        self._last_id += 1
        return self._last_id

    async def start(self):
        await self._start_server()

    async def _reader_queue(
        self,
        sr: asyncio.stream.StreamReader,
        sw: asyncio.stream.StreamWriter,
        client_sender_queue: Queue,
        client_addr: list[str, int],
        client_id: int,
    ):
        while True:
            try:
                data = await sr.read()
                if not data:
                    return
                data = b"".join((f"{client_id}: ".encode(), data))
            except OSError:
                return

            for addr, sender_queue in self.clients.copy():
                if client_addr != addr:
                    await sender_queue.put(data)
            await asyncio.sleep_ms(100)

    async def _sender_queue(
        self,
        sr: asyncio.stream.StreamReader,
        sw: asyncio.stream.StreamWriter,
        client_sender_queue: Queue,
        client_addr: list[str, int],
        client_id: int,
    ):
        active = True
        while active:
            for addr, sender_queue in self.clients:
                if client_addr == addr:
                    try:
                        data = sender_queue.get_nowait()
                        sw.write(data)
                        await sw.drain()
                        print(f"_sender_queue: {addr} | data: {data}")
                    except QueueEmpty:
                        pass
                    except OSError:
                        active = False
                        break
            await asyncio.sleep_ms(100)

        self.clients.remove((client_addr, client_sender_queue))
        leave_message = f"Client {client_addr} left the chat.\n".encode()
        for addr, sender_queue in self.clients:
            await sender_queue.put(leave_message)

    async def _start_server(self):
        if self.server is None:
            self.server = await asyncio.start_server(
                self._new_connection, self.ip, self.port
            )

    async def _new_connection(
        self, sr: asyncio.stream.StreamReader, sw: asyncio.StreamWriter
    ):
        """
        Async handler function to handle new connections.
        """
        client_addr = sr.get_extra_info("peername")
        sender_queue = Queue()
        self.clients.append((client_addr, sender_queue))

        sw.write(f"Welcome {client_addr}\n".encode())

        client_id = self.next_client_id

        sw.write(f"Your client id is: {client_id}\n")
        await sw.drain()

        for addr, queue in self.clients:
            if client_addr != addr:
                await queue.put(f"Client {client_addr} entered the chat\n".encode())

        asyncio.create_task(
            self._reader_queue(sr, sw, sender_queue, client_addr, client_id)
        )
        asyncio.create_task(
            self._sender_queue(sr, sw, sender_queue, client_addr, client_id)
        )


async def main():
    """
    Main entry point for asyncio.run
    """
    server = ChatServer("0.0.0.0", 10_000)
    await server.start()

    while True:
        await asyncio.sleep(10)


if __name__ == "__main__":
    print()
    print("Starting Server")
    print("To send the server a command:")
    try:
        print(f'echo "Hello World" | nc -N {wlan.ifconfig()[0]} 10000')
    except NameError:
        print("No WIFI-Connection")
    else:
        asyncio.run(main())
    finally:
        print()

boot.py

import time
import json
from network import WLAN, STA_IF


try:
    with open("wifi-creds.json") as fd:
        creds = json.load(fd)
except OSError:
    with open("wifi-creds.json", "w") as fd:
        json.dump({"ssid": "", "password": ""}, fd)
else:
    ssid, password = creds["ssid"], creds["password"]
    
    if ssid and password:
        wlan = WLAN(STA_IF)
        wlan.active(False)
        wlan.active(True)
        wlan.connect(ssid, password)
        
        while wlan.ifconfig()[0] == "0.0.0.0":
            time.sleep_ms(100)
            print(".", end="")
        
        print()
        print("Connected")
        print("IP:", wlan.ifconfig()[0])
    else:
        print("ssid and password is not set yet, skipping wifi connection")
        

As a side note. To write indented code, do the following:

#```python
#code
#```

Without the #