### Start the sample server Source: https://github.com/chrysn/aiocoap/blob/master/doc/guidedtour.md This command starts the aiocoap sample server in a terminal. ```shell $ ./server.py ``` -------------------------------- ### Client GET Request Example Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md An example of a client making a GET request to a CoAP server. ```python 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.run(main()) ``` -------------------------------- ### Logging Setup and Server Initialization Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md This snippet demonstrates how to configure logging levels for aiocoap and set up a basic resource tree for a server. ```python import logging import asyncio import aiocoap from aiocoap import resource logging.basicConfig(level=logging.INFO) logging.getLogger("coap-server").setLevel(logging.DEBUG) async def main(): # Resource tree creation root = resource.Site() root.add_resource( [".well-known", "core"], resource.WKCResource(root.get_resources_as_linkheader) ) root.add_resource([], Welcome()) root.add_resource(["time"], TimeResource()) root.add_resource(["other", "block"], BlockResource()) root.add_resource(["other", "separate"], SeparateLargeResource()) root.add_resource(["whoami"], WhoAmI()) await aiocoap.Context.create_server_context(root) # Run forever await asyncio.get_running_loop().create_future() if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Install with specific extras Source: https://github.com/chrysn/aiocoap/blob/master/doc/installation.md Example of installing aiocoap by excluding problematic extras, like 'tinydtls', and including only desired ones. ```bash [oscore,prettyprint,docs] ``` -------------------------------- ### BlockResource Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md An example resource that supports GET and PUT methods and handles large responses triggering blockwise transfer. ```python 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, content_format=ContentFormat.TEXT) 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) ``` -------------------------------- ### TimeResource Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md An example resource that can be observed and notifies updates periodically. ```python class TimeResource(resource.ObservableResource): """Example resource that can be observed. The `notify` method keeps scheduling itself, and calls `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_running_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, content_format=ContentFormat.TEXT) ``` -------------------------------- ### Install and import aiocoap Source: https://github.com/chrysn/aiocoap/blob/master/contrib/aiocoap.ipynb Installs the aiocoap library with prettyprint support and imports it, then creates a client context. ```python import micropip await micropip.install("aiocoap[prettyprint]") import aiocoap ctx = await aiocoap.Context.create_client_context() ``` -------------------------------- ### Interactive asynchronous example Source: https://github.com/chrysn/aiocoap/blob/master/doc/guidedtour.md An example of creating a client context, sending a request, and printing the response in an interactive session. ```python >>> from aiocoap import * >>> protocol = await Context.create_client_context() >>> msg = Message(code=GET, uri="coap://localhost/other/separate") >>> response = await protocol.request(msg).response >>> print(response) , 1 option(s), 189 byte(s) payload, token 85e6, CON, MID 0x1234> ``` -------------------------------- ### Install from local checkout Source: https://github.com/chrysn/aiocoap/blob/master/doc/installation.md Installs aiocoap from a local Git checkout with all extras and documentation tools. ```bash $ pip3 install --upgrade ".[all,docs]" ``` -------------------------------- ### Making parallel requests Source: https://github.com/chrysn/aiocoap/blob/master/doc/guidedtour.md An example demonstrating how to make multiple requests concurrently using asyncio and process their responses as they complete. ```python >>> async def main(): ... responses = [ ... protocol.request(Message(code=GET, uri=u)).response ... for u ... in ("coap://localhost/time", "coap://vs0.inf.ethz.ch/obs", "coap://coap.me/test") ... ] ... for f in asyncio.as_completed(responses): ... response = await f ... print("Response from {}: {}".format(response.get_request_uri(), response.payload)) >>> run(main()) Response from coap://localhost/time: b'2016-12-07 18:16' Response from coap://vs0.inf.ethz.ch/obs: b'18:16:11' Response from coap://coap.me/test: b'welcome to the ETSI plugtest! last change: 2016-12-06 16:02:33 UTC' ``` -------------------------------- ### Client PUT Request Example with Large Payload Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md An example of a client making a PUT request with a payload larger than 1kB, demonstrating block transfer. ```python 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.run(main()) ``` -------------------------------- ### hostportjoin examples Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.util.md Examples of joining host and port into a host:port string, including IPv6 literals. ```python >>> hostportjoin('example.com') 'example.com' >>> hostportjoin('example.com', 1234) 'example.com:1234' >>> hostportjoin('127.0.0.1', 1234) '127.0.0.1:1234' ``` ```python >>> hostportjoin('2001:db8::1]') '[2001:db8::1]' >>> hostportjoin('2001:db8::1', 1234) '[2001:db8::1]:1234' >>> hostportjoin('[2001:db8::1]', 1234) '[2001:db8::1]:1234' ``` -------------------------------- ### SeparateLargeResource Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md An example resource that simulates a long-running operation using asyncio.sleep to force empty ACK first. ```python 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, content_format=ContentFormat.TEXT) ``` -------------------------------- ### Start Fileserver Source: https://github.com/chrysn/aiocoap/blob/master/doc/stateofedhoc.md Starts the aiocoap file server with the specified credentials file. ```shell-session $ ./aiocoap-fileserver --credentials fileserver.cred.diag ``` -------------------------------- ### Send a GET request and get the response Source: https://github.com/chrysn/aiocoap/blob/master/contrib/edhoc-demo-server.ipynb Sends a GET request to the /whoami endpoint and retrieves the response. ```python req = Message(code=GET, uri="coap://demo.coap.amsuess.com/whoami") res = await ctx.request(req).response res ``` -------------------------------- ### Install ipywidgets and create a slider Source: https://github.com/chrysn/aiocoap/blob/master/contrib/aiocoap-server.ipynb Installs the ipywidgets library and creates a FloatSlider widget. ```python await micropip.install("ipywidgets") import ipywidgets as widgets slider = widgets.FloatSlider() slider ``` -------------------------------- ### BlockOption reduced_to example Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.optiontypes.md Example demonstrating the usage of the reduced_to method for BlockOption. ```python >>> initial = BlockOption.BlockwiseTuple(10, 0, 5) >>> initial == initial.reduced_to(6) True >>> initial.reduced_to(3) BlockwiseTuple(block_number=40, more=0, size_exponent=3) ``` -------------------------------- ### Example configuration for UDP6 Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.config.md This example shows a configuration for UDP6 transport, including bind addresses. ```default [transport.udp6] bind = ["[::]:5683", "[::]:61616"] ``` -------------------------------- ### hostportsplit examples Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.util.md Examples of splitting a host:port string into host and port, returning port as an integer or None. ```python >>> hostportsplit('foo') ('foo', None) >>> hostportsplit('foo:5683') ('foo', 5683) >>> hostportsplit('[::1%eth0]:56830') ('::1%eth0', 56830) ``` -------------------------------- ### Example multicast client request Source: https://github.com/chrysn/aiocoap/blob/master/doc/faq.md An example of how to send a multicast request using the aiocoap-client, specifying an interface. ```default ./aiocoap-client coap://'[ff02::fd%eth0]'/.well-known/core --non ``` -------------------------------- ### Kivy examples were updated to current mainline Kivy. Source: https://github.com/chrysn/aiocoap/blob/master/doc/news.md This note indicates updates to Kivy examples. ```text Kivy ``` -------------------------------- ### Welcome Resource Source: https://github.com/chrysn/aiocoap/blob/master/doc/examples.md A basic resource that returns a welcome message in different formats. ```python import datetime import logging import asyncio import aiocoap.resource as resource from aiocoap.numbers.contentformat import ContentFormat import aiocoap class Welcome(resource.Resource): representations = { ContentFormat.TEXT: b"Welcome to the demo server", ContentFormat.LINKFORMAT: b",ct=40", # ad-hoc for application/xhtml+xml;charset=utf-8 ContentFormat(65000): b'' b"aiocoap demo" b"

Welcome to the aiocoap demo server!

" b'", } default_representation = ContentFormat.TEXT async def render_get(self, request): cf = ( self.default_representation if request.opt.accept is None else request.opt.accept ) try: return aiocoap.Message(payload=self.representations[cf], content_format=cf) except KeyError: raise aiocoap.error.UnsupportedContentFormat ``` -------------------------------- ### Install aiocoap with OSCORE and prettyprint Source: https://github.com/chrysn/aiocoap/blob/master/contrib/edhoc-demo-server.ipynb Installs the necessary aiocoap library with OSCORE and prettyprint support using micropip. ```python import micropip, pyodide # The custom index has both the latest aiocoap and a current and compatible lakers-python. await micropip.install( "aiocoap[oscore,prettyprint]", index_urls=["https://aiocoap.codeberg.page/aiocoap", "PYPI"], ) ``` -------------------------------- ### Send a GET request to the server for /.well-known/core Source: https://github.com/chrysn/aiocoap/blob/master/doc/guidedtour.md This command uses the aiocoap-client tool to send a GET request to the server for the /.well-known/core resource. ```shell $ ./aiocoap-client coap://localhost/.well-known/core … application/link-format content was re-formatted ; ct="40", ; obs, , ; title="A large resource", , ; rel="impl-info" ``` -------------------------------- ### Awaiting the response Source: https://github.com/chrysn/aiocoap/blob/master/doc/guidedtour.md Shows how to await the completion of the Future to get the actual response. ```python >>> await protocol.request(msg).response , 186 byte(s) payload> ``` -------------------------------- ### Install aiocoap and create a proxy client context Source: https://github.com/chrysn/aiocoap/blob/master/contrib/aiocoap-proxy.ipynb This snippet shows how to install the aiocoap library using micropip and then create a client context that forwards requests through a proxy. ```python import micropip await micropip.install("aiocoap") import aiocoap, aiocoap.proxy.client ctx = await aiocoap.Context.create_client_context() ctx = aiocoap.proxy.client.ProxyForwarder("coaps+ws://proxy.coap.amsuess.com", ctx) ``` -------------------------------- ### Link.get_target() example Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.util.vendored.link_header.md Demonstrates how to get the absolute URI of the target of a link by joining the retrieval address with the link-value. ```python >>> Link('../', rel='index').get_target('http://www.example.com/book1/chapter1/') 'http://www.example.com/book1/' >>> Link('', rel='next', anchor='../').get_target('http://www.example.com/book1/chapter1/') 'http://www.example.com/book1/chapter1/' ``` -------------------------------- ### Link.get_context() example Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.util.vendored.link_header.md Illustrates how to get the absolute URI of the context of a link, considering the base address and anchor parameter. ```python >>> Link('../', rel='index').get_context('http://www.example.com/book1/chapter1/') 'http://www.example.com/book1/chapter1/' >>> Link('', rel='next', anchor='../').get_context('http://www.example.com/book1/chapter1/') 'http://www.example.com/book1/chapter1/ தாவர' ``` -------------------------------- ### Create a client context Source: https://github.com/chrysn/aiocoap/blob/master/contrib/edhoc-demo-server.ipynb Initializes a new aiocoap client context. ```python import aiocoap, aiocoap.proxy.client from aiocoap import * ctx = await Context.create_client_context() ``` -------------------------------- ### Asynchronous code in a .py file Source: https://github.com/chrysn/aiocoap/blob/master/doc/guidedtour.md Example of how to structure asynchronous code in a Python file using asyncio.run. ```python import asyncio from aiocoap import * async def main(): protocol = await Context.create_client_context() msg = Message(code=GET, uri="coap://localhost/other/separate") response = await protocol.request(msg).response print(response) asyncio.run(main()) ``` -------------------------------- ### Nesting Sites and adding resources Source: https://github.com/chrysn/aiocoap/blob/master/doc/module/aiocoap.resource.md Example showing how to nest Site instances and add resources to them. ```python >>> batch = Site() >>> batch.add_resource(["light1"], Resource()) >>> batch.add_resource(["light2"], Resource()) >>> batch.add_resource([], Resource()) >>> s = Site() >>> s.add_resource(["batch"], batch) ``` -------------------------------- ### Install latest released version Source: https://github.com/chrysn/aiocoap/blob/master/doc/installation.md Installs the latest released version of aiocoap with all extras. ```bash $ pip3 install --upgrade "aiocoap[all]" ``` -------------------------------- ### Configure a proxy forwarder Source: https://github.com/chrysn/aiocoap/blob/master/contrib/edhoc-demo-server.ipynb Configures a proxy forwarder to run the demo even from a restricted environment. ```python # Configure a proxy so that the demo can be run even from a ctx = aiocoap.proxy.client.ProxyForwarder("coaps+ws://proxy.coap.amsuess.com", ctx) ``` -------------------------------- ### Install on pyodide using micropip Source: https://github.com/chrysn/aiocoap/blob/master/doc/installation.md Installs aiocoap with all extras in a pyodide environment using micropip. ```python >>> import micropip >>> await micropip.install("aiocoap[all]") ``` -------------------------------- ### Configure client credentials for the demo server Source: https://github.com/chrysn/aiocoap/blob/master/contrib/edhoc-demo-server.ipynb Loads client credentials from a dictionary for the demo server. ```python for_demo = { "coap://demo.coap.amsuess.com/*": { "edhoc-oscore": { "suite": 2, "method": 3, "own_cred": {"unauthenticated": True}, "peer_cred": { 14: { 2: "demo.coap.amsuess.com", 8: { 1: { 1: 2, 2: b"\0", -1: 1, -2: bytes.fromhex( "b9cc746df6641d55044478b29df019ef22b4d2e96ffcf8de85434e5d0f27c33c" ), -3: bytes.fromhex( "e14e87330d093b469b121c3d0e4d9452cb90036a6e209f21f37d35d2a05c426c" ), } }, } }, } } } cxt.client_credentials.load_from_dict(for_demo) ``` -------------------------------- ### Install latest development version from web Source: https://github.com/chrysn/aiocoap/blob/master/doc/installation.md Installs the latest development version of aiocoap directly from its GitHub repository. ```bash $ pip3 install --upgrade "git+https://github.com/chrysn/aiocoap#egg=aiocoap[all]" ``` -------------------------------- ### Running the HTTP server Source: https://github.com/chrysn/aiocoap/blob/master/contrib/html-viewer/README.rst Command to start a simple HTTP server to serve the viewer files. ```bash python3 -m http.server ```