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¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | import logging import asyncio from aiocoap import * logging.basicConfig(level=logging.INFO) async def main(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://localhost/time') try: response = await protocol.request(request).response except Exception as e: print('Failed to fetch resource:') print(e) else: print('Result: %s\n%r'%(response.code, response.payload)) if __name__ == "__main__": asyncio.get_event_loop().run_until_complete(main()) |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | import logging import asyncio from aiocoap import * logging.basicConfig(level=logging.INFO) async def main(): """Perform a single PUT request to localhost on the default port, URI "/other/block". The request is sent 2 seconds after initialization. The payload is bigger than 1kB, and thus sent as several blocks.""" context = await Context.create_client_context() await asyncio.sleep(2) payload = b"The quick brown fox jumps over the lazy dog.\n" * 30 request = Message(code=PUT, payload=payload, uri="coap://localhost/other/block") response = await context.request(request).response print('Result: %s\n%r'%(response.code, response.payload)) if __name__ == "__main__": asyncio.get_event_loop().run_until_complete(main()) |
Server¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | import datetime import logging import asyncio import aiocoap.resource as resource import aiocoap class BlockResource(resource.Resource): """Example resource which supports the GET and PUT methods. It sends large responses, which trigger blockwise transfer.""" def __init__(self): super().__init__() self.set_content(b"This is the resource's default content. It is padded "\ b"with numbers to be large enough to trigger blockwise "\ b"transfer.\n") def set_content(self, content): self.content = content while len(self.content) <= 1024: self.content = self.content + b"0123456789\n" async def render_get(self, request): return aiocoap.Message(payload=self.content) async def render_put(self, request): print('PUT payload: %s' % request.payload) self.set_content(request.payload) return aiocoap.Message(code=aiocoap.CHANGED, payload=self.content) class SeparateLargeResource(resource.Resource): """Example resource which supports the GET method. It uses asyncio.sleep to simulate a long-running operation, and thus forces the protocol to send empty ACK first. """ def get_link_description(self): # Publish additional data in .well-known/core return dict(**super().get_link_description(), title="A large resource") async def render_get(self, request): await asyncio.sleep(3) payload = "Three rings for the elven kings under the sky, seven rings "\ "for dwarven lords in their halls of stone, nine rings for "\ "mortal men doomed to die, one ring for the dark lord on his "\ "dark throne.".encode('ascii') return aiocoap.Message(payload=payload) class TimeResource(resource.ObservableResource): """Example resource that can be observed. The `notify` method keeps scheduling itself, and calles `update_state` to trigger sending notifications.""" def __init__(self): super().__init__() self.handle = None def notify(self): self.updated_state() self.reschedule() def reschedule(self): self.handle = asyncio.get_event_loop().call_later(5, self.notify) def update_observation_count(self, count): if count and self.handle is None: print("Starting the clock") self.reschedule() if count == 0 and self.handle: print("Stopping the clock") self.handle.cancel() self.handle = None async def render_get(self, request): payload = datetime.datetime.now().\ strftime("%Y-%m-%d %H:%M").encode('ascii') return aiocoap.Message(payload=payload) class WhoAmI(resource.Resource): async def render_get(self, request): text = ["Used protocol: %s." % request.remote.scheme] text.append("Request came from %s." % request.remote.hostinfo) text.append("The server address used %s." % request.remote.hostinfo_local) claims = list(request.remote.authenticated_claims) if claims: text.append("Authenticated claims of the client: %s." % ", ".join(repr(c) for c in claims)) else: text.append("No claims authenticated.") return aiocoap.Message(content_format=0, payload="\n".join(text).encode('utf8')) # logging setup logging.basicConfig(level=logging.INFO) logging.getLogger("coap-server").setLevel(logging.DEBUG) def main(): # Resource tree creation root = resource.Site() root.add_resource(['.well-known', 'core'], resource.WKCResource(root.get_resources_as_linkheader)) root.add_resource(['time'], TimeResource()) root.add_resource(['other', 'block'], BlockResource()) root.add_resource(['other', 'separate'], SeparateLargeResource()) root.add_resource(['whoami'], WhoAmI()) asyncio.Task(aiocoap.Context.create_server_context(root)) asyncio.get_event_loop().run_forever() if __name__ == "__main__": main() |