Created
July 12, 2021 18:33
-
-
Save wybiral/159a2798f253f71c49728e4ca52e0fb0 to your computer and use it in GitHub Desktop.
MicroPython Server-Sent Events
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import network | |
| import web | |
| import uasyncio as asyncio | |
| CLIENTS = set() | |
| # access point credentials | |
| AP_SSID = 'SSE AP' | |
| AP_PASSWORD = 'donthackmebro' | |
| AP_AUTHMODE = network.AUTH_WPA_WPA2_PSK | |
| app = web.App(host='0.0.0.0', port=80) | |
| # root route handler | |
| @app.route('/') | |
| async def index_handler(r, w): | |
| w.write(b'HTTP/1.0 200 OK\r\n') | |
| w.write(b'Content-Type: text/html; charset=utf-8\r\n') | |
| w.write(b'\r\n') | |
| w.write(b'''<html><head><script> | |
| window.onload = () => { | |
| const out = document.querySelector('#out'); | |
| const sse = new EventSource('/events'); | |
| sse.onmessage = evt => { | |
| const el = document.createElement('div'); | |
| el.innerText = evt.data; | |
| out.appendChild(el); | |
| }; | |
| }; | |
| </script></head> | |
| <body><div id="out"></div></body></html>''') | |
| await w.drain() | |
| # /events EventSource handler | |
| @app.route('/events') | |
| async def events_handler(r, w): | |
| global CLIENTS | |
| sse = await web.EventSource.upgrade(r, w) | |
| CLIENTS.add(sse) | |
| while True: | |
| try: | |
| await sse.send('', event='ping') | |
| except: | |
| break | |
| await asyncio.sleep(60) | |
| CLIENTS.discard(sse) | |
| async def background_task(): | |
| global CLIENTS | |
| while True: | |
| await asyncio.sleep(1) | |
| for client in CLIENTS: | |
| try: | |
| await client.send('Sent from background_task') | |
| except: | |
| continue | |
| # Create WiFi access point | |
| wifi = network.WLAN(network.AP_IF) | |
| wifi.active(True) | |
| wifi.config(essid=AP_SSID, password=AP_PASSWORD, authmode=AP_AUTHMODE) | |
| while wifi.active() == False: | |
| pass | |
| print(wifi.ifconfig()) | |
| # Start event loop and create server task | |
| loop = asyncio.get_event_loop() | |
| loop.create_task(app.serve()) | |
| loop.create_task(background_task()) | |
| loop.run_forever() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment