Develop a CoAP Server and CoAP Client to exchange IoT sensor data, where the current sensor data is transferred from the server to a dashboard.


CoAP_IOT_Project/
│
├── server.py
├── client.py
├── dashboard.py
└── requirements.txt

pip install aiocoap streamlit streamlit-autorefresh requests

server.py

import asyncio
import json
import random

from aiocoap import *
from aiocoap.resource import Resource, Site

# Create sensor resource
class SensorResource(Resource):

    async def render_get(self, request):

        # Generate simulated sensor readings
        temperature = round(random.uniform(25, 35), 2)
        humidity = round(random.uniform(50, 80), 2)

        # Prepare sensor data
        data = {
            "temperature": temperature,
            "humidity": humidity
        }

        print("Sensor Data Generated")
        print(data)

        # Convert data into JSON
        payload = json.dumps(data).encode("utf-8")

        # Send response
        return Message(
            payload=payload,
            content_format=ContentFormat.JSON
        )


# Start CoAP server
async def main():

    site = Site()

    # Register sensor resource
    site.add_resource(
        ["sensors"],
        SensorResource()
    )

    # Start server
    await Context.create_server_context(
        site,
        bind=("127.0.0.1", 5683)
    )

    print("CoAP Server Started")
    print("Waiting for client requests...")

    # Keep server running
    await asyncio.get_running_loop().create_future()


if __name__ == "__main__":
    asyncio.run(main())

client.py

import asyncio
import json

from aiocoap import *

# CoAP client function
async def main():

    # Create CoAP client context
    context = await Context.create_client_context()

    # Server resource address
    request = Message(
        code=GET,
        uri="coap://127.0.0.1:5683/sensors"
    )

    print("Sending GET request to CoAP Server...")

    # Send request and receive response
    response = await context.request(request).response

    # Convert response into JSON
    data = json.loads(response.payload.decode("utf-8"))

    print("\nSensor Data Received")

    print("Temperature:", data["temperature"], "°C")
    print("Humidity:", data["humidity"], "%")

    # Close client context
    await context.shutdown()


if __name__ == "__main__":
    asyncio.run(main())


dashboard.py

import streamlit as st
import asyncio
import json

from aiocoap import *
from streamlit_autorefresh import st_autorefresh

# Dashboard title
st.title("CoAP IoT Sensor Dashboard")

st.write("Live Temperature and Humidity Monitoring")

# Automatic refresh every 5 seconds
st_autorefresh(
    interval=5000,
    key="refresh"
)

# Function to receive sensor data
async def get_sensor_data():

    # Create CoAP client
    context = await Context.create_client_context()

    try:

        # Send GET request
        request = Message(
            code=GET,
            uri="coap://127.0.0.1:5683/sensors"
        )

        response = await context.request(request).response

        # Convert response into JSON
        data = json.loads(
            response.payload.decode("utf-8")
        )

        return data

    finally:
        await context.shutdown()


# Display sensor readings
try:

    data = asyncio.run(get_sensor_data())

    st.subheader("Current Sensor Readings")

    col1, col2 = st.columns(2)

    col1.metric(
        "Temperature",
        str(data["temperature"]) + " °C"
    )

    col2.metric(
        "Humidity",
        str(data["humidity"]) + " %"
    )

    st.success("Connected to CoAP Server")

    st.subheader("Sensor Data")

    st.json(data)

except Exception as e:

    st.error("Unable to connect to CoAP Server")

    st.write(e)

