Translate a local CoAP reading into an HTTP record

Map a local CoAP reading to an HTTP record, reject malformed data, and replay an outage queue with a stable idempotency key.

Gateway Gus: I want you to preserve source meaning through mapping, failure, and replay., your practice guide

Gateway Gus: I want you to preserve source meaning through mapping, failure, and replay.
Predict the reading, then compare it with the measurement.

Python 3 + aiocoap

Python

Map a local CoAP reading to an HTTP record, reject malformed data, and replay an outage queue with a stable idempotency key.

Tier 2 · Python · install required · No account

Version tested: Python 3.9.21; aiocoap 0.4.7; local UDP and standard-library HTTP; Xfce Terminal 1.0.4 on Xvfb. Date: 2026-10-08.

Download the three files, create a Python virtual environment, install requirements.txt, and run main.py.

Open the Python run guide (new tab)

Get the files

Download all three prepared files into one folder. No account or paid service is required.

main.py

8,284 bytes · Python program

Download

"""Local aiocoap sensor and stdlib HTTP bridge with validation and replay."""
import argparse
import asyncio
import hashlib
import http.client
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import threading

import aiocoap
from aiocoap import resource

COAP_PORT = 17682
HTTP_PORT = 17683
QUEUE_LIMIT = 2

class Sensor(resource.Resource):
    payload = b'{"event_id":"reading-001","celsius":21.5,"timestamp":"2026-10-08T12:00:00Z"}'

async def render_get(self, request):
        reply = aiocoap.Message(payload=self.payload)
        reply.opt.content_format = 50
        return reply

class Receiver(BaseHTTPRequestHandler):
    receipts = {}
    wire = []

def do_POST(self):
        body = self.rfile.read(int(self.headers["Content-Length"]))
        key = self.headers.get("Idempotency-Key")
        self.__class__.wire.append((self.raw_requestline, body, key))
        if self.path != "/measurements" or not key:
            self.send_error(400)
            return
        if key in self.__class__.receipts:
            status = 200
        else:
            self.__class__.receipts[key] = body
            status = 201
        self.send_response(status)
        self.send_header("Content-Length", "0")
        self.end_headers()

def log_message(self, format, *args):
        pass

def start_http():
    server = HTTPServer(("127.0.0.1", HTTP_PORT), Receiver)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread

def stop_http(server, thread):
    server.shutdown()
    thread.join()
    server.server_close()

def map_payload(payload):
    record = json.loads(payload)
    if (set(record) != {"event_id", "celsius", "timestamp"}
            or not isinstance(record["event_id"], str)
            or not isinstance(record["celsius"], (int, float))
            or isinstance(record["celsius"], bool)
            or not -40 <= record["celsius"] <= 85
            or not isinstance(record["timestamp"], str)
            or not record["timestamp"].endswith("Z")):
        raise ValueError("sensor contract rejected: event_id, numeric celsius, UTC timestamp required")
    mapped = {"resource": "/temperature", "value": record["celsius"],
              "unit": "Cel", "timestamp": record["timestamp"]}
    return record["event_id"], json.dumps(mapped, separators=(",", ":")).encode()

def post(body, key):
    conn = http.client.HTTPConnection("127.0.0.1", HTTP_PORT, timeout=1)
    try:
        conn.request("POST", "/measurements", body=body, headers={
            "Content-Type": "application/json", "Idempotency-Key": key})
        response = conn.getresponse()
        response.read()
        return response.status
    finally:
        conn.close()

async def run(step):
    site = resource.Site()
    sensor = Sensor()
    site.add_resource(("temperature",), sensor)
    server = await aiocoap.Context.create_server_context(site, bind=("127.0.0.1", COAP_PORT))
    client = await aiocoap.Context.create_client_context()
    http_server, http_thread = start_http()
    outputs = {}
    try:
        async def read_sensor():
            request = aiocoap.Message(code=aiocoap.GET, uri=f"coap://127.0.0.1:{COAP_PORT}/temperature")
            return await asyncio.wait_for(client.request(request).response, 3)

response = await read_sensor()
        event_id, body = map_payload(response.payload)
        key = hashlib.sha256(event_id.encode()).hexdigest()[:16]
        outputs[1] = [
            "STEP 1: GET /temperature from the local aiocoap sensor",
            f"CoAP response={response.code} content-format={response.opt.content_format}",
            f"CoAP payload bytes hex={response.payload.hex()}",
            f"CoAP payload UTF-8={response.payload.decode()}",
            "These payload bytes came from the local CoAP server in this run.",
        ]
        status = await asyncio.to_thread(post, body, key)
        line, received_body, received_key = Receiver.wire[-1]
        outputs[2] = [
            "STEP 2: map and POST the sensor record",
            f"HTTP request line bytes hex={line.hex()}",
            f"HTTP request line={line.decode().strip()}",
            f"HTTP Idempotency-Key={received_key}",
            f"HTTP body bytes hex={received_body.hex()}",
            f"HTTP body UTF-8={received_body.decode()}",
            f"HTTP status={status}; stored receipts={len(Receiver.receipts)}",
            "HTTP bytes above were received by the local stdlib server.",
        ]
        sensor.payload = b'{"event_id":"reading-002","celsius":"bad","timestamp":"2026-10-08T12:01:00Z"}'
        bad = await read_sensor()
        try:
            map_payload(bad.payload)
            raise AssertionError("malformed value passed validation")
        except ValueError as error:
            reason = str(error)
        outputs[3] = [
            "STEP 3: reject a malformed CoAP reading",
            f"CoAP response={bad.code} payload bytes hex={bad.payload.hex()}",
            f"decoded sensor reading={bad.payload.decode()}",
            f"validation={reason}",
            f"HTTP receipts remain={len(Receiver.receipts)}",
            "Malformed input was sent by the real local CoAP server, then rejected before POST.",
        ]
        sensor.payload = b'{"event_id":"reading-003","celsius":22.0,"timestamp":"2026-10-08T12:02:00Z"}'
        outage_response = await read_sensor()
        outage_id, outage_body = map_payload(outage_response.payload)
        outage_key = hashlib.sha256(outage_id.encode()).hexdigest()[:16]
        stop_http(http_server, http_thread)
        http_server = None
        queue = []
        try:
            await asyncio.to_thread(post, outage_body, outage_key)
            raise AssertionError("outage request unexpectedly succeeded")
        except OSError as error:
            if len(queue) >= QUEUE_LIMIT:
                raise RuntimeError("retry queue full")
            queue.append((outage_body, outage_key))
            failure = error.__class__.__name__
        outputs[4] = [
            "STEP 4: HTTP endpoint outage after a valid CoAP GET",
            f"CoAP payload UTF-8={outage_response.payload.decode()}",
            f"HTTP POST result={failure}",
            f"queue depth={len(queue)}/{QUEUE_LIMIT} key={outage_key}",
            f"queued HTTP body={outage_body.decode()}",
            "The HTTP listener was closed before POST; this is a local failure, not a cloud outage.",
        ]
        http_server, http_thread = start_http()
        queued_body, queued_key = queue.pop(0)
        retry_status = await asyncio.to_thread(post, queued_body, queued_key)
        line, received_body, received_key = Receiver.wire[-1]
        outputs[5] = [
            "STEP 5: replay the queued HTTP POST",
            f"HTTP request line bytes hex={line.hex()}",
            f"HTTP Idempotency-Key={received_key}",
            f"HTTP body UTF-8={received_body.decode()}",
            f"HTTP retry status={retry_status}; queue depth={len(queue)}",
            f"stored receipts={len(Receiver.receipts)}",
            "The request line and body were received by the restarted local HTTP server.",
        ]
        duplicate_status = await asyncio.to_thread(post, queued_body, queued_key)
        outputs[6] = [
            "STEP 6: submit the same idempotency key again",
            f"HTTP request line bytes hex={Receiver.wire[-1][0].hex()}",
            f"HTTP Idempotency-Key={Receiver.wire[-1][2]}",
            f"HTTP duplicate status={duplicate_status}",
            f"stored receipts={len(Receiver.receipts)}",
            f"same key stored once={sum(1 for item in Receiver.receipts if item == queued_key) == 1}",
            "The second POST reached the real local HTTP server; its dedupe table did not grow.",
        ]
        for number in ([step] if step else sorted(outputs)):
            print("\n".join(outputs[number]), flush=True)
    finally:
        if http_server:
            stop_http(http_server, http_thread)
        await client.shutdown()
        await server.shutdown()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--step", type=int, choices=range(1, 7))
    args = parser.parse_args()
    asyncio.run(run(args.step))

requirements.txt

15 bytes · Pinned dependencies

Download

README.md

1,322 bytes · Run guide

Download

# CoAP to HTTP gateway translation

Use Python 3.9 or newer. In this folder:

```bash
python3 -m venv .venv
. .venv/bin/activate
python -m pip install -r requirements.txt
python main.py
python main.py --step 4
```

Each `--step` reruns the complete short scenario from a fresh local state and prints only the selected step. Steps 1–2 GET a fictional reading over CoAP and POST a mapped record to the standard-library HTTP server. Step 3 sends a malformed CoAP value and rejects it. Step 4 closes the HTTP listener and queues one validated event in a two-entry queue; a full queue raises an error rather than accepting more work. Steps 5–6 restart the listener and replay the event twice with one idempotency key. The programs use 127.0.0.1 CoAP port 17682 and HTTP port 17683; stop another service using these ports. No radio, cloud endpoint, authentication, or production durability is claimed. The in-memory queue and receipt table disappear when the program exits.

After the full run, choose a failure question to investigate: compare the malformed reading with the validation rule in step 3, or compare the outage queue and duplicate replay in steps 4–6. aiocoap supplies the CoAP endpoint, while Python's standard-library HTTP server supplies an independent destination and records the received request bytes.

Put main.py, requirements.txt, and README.md in the same folder.
Create and activate a Python virtual environment, then install requirements.txt.
Run python3 main.py and compare its output with each observed result below.

Steps

Screens captured against Python 3 + aiocoap Python 3.9.21; aiocoap 0.4.7; local UDP and standard-library HTTP; Xfce Terminal 1.0.4 on Xvfb on 2026-10-08; the tool may have moved on — the text steps are the contract.

Step 1
Do
In the terminal window panel, run `python3 main.py --step 1` to GET the local CoAP temperature reading.
You will see
The CoAP response is `2.05 Content`, content format JSON, and the actual payload bytes and UTF-8 record are printed.
Why it matters
A bridge needs a source payload and resource contract before mapping to HTTP.
Step 2
Do
In the terminal window panel, run `python3 main.py --step 2` to map the reading and POST it to the local HTTP receiver.
You will see
The HTTP receiver sees `POST /measurements HTTP/1.1`, a stable `Idempotency-Key`, mapped JSON body, and status `201` with one receipt.
Why it matters
The request line and body are bytes received by the server, proving the translated delivery and its unit and timestamp.
Step 3
Do
Run `python3 main.py --step 3` after the sensor sends a string in its numeric `celsius` field.
You will see
Validation reports `sensor contract rejected: event_id, numeric celsius, UTC timestamp required`; HTTP receipts remain `1`.
Why it matters
Rejecting malformed source meaning before POST prevents a false measurement from entering the destination.
Step 4
Do
In the terminal window panel, run `python3 main.py --step 4` while the script closes the HTTP listener before POST.
You will see
The POST gets `ConnectionRefusedError`; queue depth becomes `1` with the mapped record and idempotency key.
Why it matters
The outage is visible and a valid record is retained for bounded local replay.
Step 5
Do
In the terminal window panel, run `python3 main.py --step 5` to restart the HTTP receiver and replay the queued record.
You will see
The retry returns HTTP `201`, queue depth `0`, and stored receipts `2`.
Why it matters
The replay moves one retained record across the recovered boundary.
Step 6
Do
In the terminal window panel, run `python3 main.py --step 6` to resend the same record and key once more.
You will see
The duplicate returns HTTP `200`; stored receipts remain `2` and `same key stored once=True`.
Why it matters
The receiver uses the idempotency key to avoid a second stored side effect on duplicate delivery.

Chapter checks

These questions refer to the chapter’s examples. Use the return links to review their answers.

Why is a protocol bridge more than converting bytes from one protocol to another?
Return to the chapter’s knowledge check
A gateway polls a legacy meter every 10 seconds and publishes results to an event stream. The meter stops responding for one minute. What should the gateway do?
Return to the chapter’s knowledge check

Return to Protocol Bridging Examples · Browse Labs

Python 3 + aiocoap

Get the files

main.py

requirements.txt

README.md

Steps

1 Step 1

2 Step 2

3 Step 3

4 Step 4

5 Step 5

6 Step 6

Chapter checks

Step 1

Step 2

Step 3

Step 4

Step 5

Step 6