Usage Examples

These files can serve as reference implementations for a simplistic server and client. In order to test them, run ./server.py in one terminal, and use ./clientGET.py and ./clientPUT.py to interact with it.

The programs’ source code should give you a good starting point to get familiar with the library if you prefer reading code to reading tutorials. Otherwise, you might want to have a look at the Guided Tour through aiocoap, where the relevant concepts are introduced and explained step by step.

Note

These example programs are not shipped in library version of aiocoap. They are present if you followed the Development version section of the installation instructions; otherwise, you can download them from the project website.

Client

 1import logging
 2import asyncio
 3
 4from aiocoap import *
 5
 6logging.basicConfig(level=logging.INFO)
 7
 8
 9async def main():
10    protocol = await Context.create_client_context()
11
12    request = Message(code=GET, uri="coap://localhost/time")
13
14    try:
15        response = await protocol.request(request).response
16    except Exception as e:
17        print("Failed to fetch resource:")
18        print(e)
19    else:
20        print("Result: %s\n%r" % (response.code, response.payload))
21
22
23if __name__ == "__main__":
24    asyncio.run(main())

 1import logging
 2import asyncio
 3
 4from aiocoap import *
 5
 6logging.basicConfig(level=logging.INFO)
 7
 8
 9async def main():
10    """Perform a single PUT request to localhost on the default port, URI
11    "/other/block". The request is sent 2 seconds after initialization.
12
13    The payload is bigger than 1kB, and thus sent as several blocks."""
14
15    context = await Context.create_client_context()
16
17    await asyncio.sleep(2)
18
19    payload = b"The quick brown fox jumps over the lazy dog.\n" * 30
20    request = Message(code=PUT, payload=payload, uri="coap://localhost/other/block")
21
22    response = await context.request(request).response
23
24    print("Result: %s\n%r" % (response.code, response.payload))
25
26
27if __name__ == "__main__":
28    asyncio.run(main())

Server

  1import datetime
  2import logging
  3
  4import asyncio
  5
  6import aiocoap.resource as resource
  7from aiocoap.numbers.contentformat import ContentFormat
  8import aiocoap
  9
 10
 11class Welcome(resource.Resource):
 12    representations = {
 13        ContentFormat.TEXT: b"Welcome to the demo server",
 14        ContentFormat.LINKFORMAT: b"</.well-known/core>,ct=40",
 15        # ad-hoc for application/xhtml+xml;charset=utf-8
 16        ContentFormat(65000): b'<html xmlns="http://www.w3.org/1999/xhtml">'
 17        b"<head><title>aiocoap demo</title></head>"
 18        b"<body><h1>Welcome to the aiocoap demo server!</h1>"
 19        b'<ul><li><a href="time">Current time</a></li>'
 20        b'<li><a href="whoami">Report my network address</a></li>'
 21        b"</ul></body></html>",
 22    }
 23
 24    default_representation = ContentFormat.TEXT
 25
 26    async def render_get(self, request):
 27        cf = (
 28            self.default_representation
 29            if request.opt.accept is None
 30            else request.opt.accept
 31        )
 32        try:
 33            return aiocoap.Message(payload=self.representations[cf], content_format=cf)
 34        except KeyError:
 35            raise aiocoap.error.UnsupportedContentFormat
 36
 37
 38class BlockResource(resource.Resource):
 39    """Example resource which supports the GET and PUT methods. It sends large
 40    responses, which trigger blockwise transfer."""
 41
 42    def __init__(self):
 43        super().__init__()
 44        self.set_content(
 45            b"This is the resource's default content. It is padded "
 46            b"with numbers to be large enough to trigger blockwise "
 47            b"transfer.\n"
 48        )
 49
 50    def set_content(self, content):
 51        self.content = content
 52        while len(self.content) <= 1024:
 53            self.content = self.content + b"0123456789\n"
 54
 55    async def render_get(self, request):
 56        return aiocoap.Message(payload=self.content)
 57
 58    async def render_put(self, request):
 59        print("PUT payload: %s" % request.payload)
 60        self.set_content(request.payload)
 61        return aiocoap.Message(code=aiocoap.CHANGED, payload=self.content)
 62
 63
 64class SeparateLargeResource(resource.Resource):
 65    """Example resource which supports the GET method. It uses asyncio.sleep to
 66    simulate a long-running operation, and thus forces the protocol to send
 67    empty ACK first."""
 68
 69    def get_link_description(self):
 70        # Publish additional data in .well-known/core
 71        return dict(**super().get_link_description(), title="A large resource")
 72
 73    async def render_get(self, request):
 74        await asyncio.sleep(3)
 75
 76        payload = (
 77            "Three rings for the elven kings under the sky, seven rings "
 78            "for dwarven lords in their halls of stone, nine rings for "
 79            "mortal men doomed to die, one ring for the dark lord on his "
 80            "dark throne.".encode("ascii")
 81        )
 82        return aiocoap.Message(payload=payload)
 83
 84
 85class TimeResource(resource.ObservableResource):
 86    """Example resource that can be observed. The `notify` method keeps
 87    scheduling itself, and calles `update_state` to trigger sending
 88    notifications."""
 89
 90    def __init__(self):
 91        super().__init__()
 92
 93        self.handle = None
 94
 95    def notify(self):
 96        self.updated_state()
 97        self.reschedule()
 98
 99    def reschedule(self):
100        self.handle = asyncio.get_event_loop().call_later(5, self.notify)
101
102    def update_observation_count(self, count):
103        if count and self.handle is None:
104            print("Starting the clock")
105            self.reschedule()
106        if count == 0 and self.handle:
107            print("Stopping the clock")
108            self.handle.cancel()
109            self.handle = None
110
111    async def render_get(self, request):
112        payload = datetime.datetime.now().strftime("%Y-%m-%d %H:%M").encode("ascii")
113        return aiocoap.Message(payload=payload)
114
115
116class WhoAmI(resource.Resource):
117    async def render_get(self, request):
118        text = ["Used protocol: %s." % request.remote.scheme]
119
120        text.append("Request came from %s." % request.remote.hostinfo)
121        text.append("The server address used %s." % request.remote.hostinfo_local)
122
123        claims = list(request.remote.authenticated_claims)
124        if claims:
125            text.append(
126                "Authenticated claims of the client: %s."
127                % ", ".join(repr(c) for c in claims)
128            )
129        else:
130            text.append("No claims authenticated.")
131
132        return aiocoap.Message(content_format=0, payload="\n".join(text).encode("utf8"))
133
134
135# logging setup
136
137logging.basicConfig(level=logging.INFO)
138logging.getLogger("coap-server").setLevel(logging.DEBUG)
139
140
141async def main():
142    # Resource tree creation
143    root = resource.Site()
144
145    root.add_resource(
146        [".well-known", "core"], resource.WKCResource(root.get_resources_as_linkheader)
147    )
148    root.add_resource([], Welcome())
149    root.add_resource(["time"], TimeResource())
150    root.add_resource(["other", "block"], BlockResource())
151    root.add_resource(["other", "separate"], SeparateLargeResource())
152    root.add_resource(["whoami"], WhoAmI())
153
154    await aiocoap.Context.create_server_context(root)
155
156    # Run forever
157    await asyncio.get_running_loop().create_future()
158
159
160if __name__ == "__main__":
161    asyncio.run(main())