Mercurial > code > home > repos > front-door-display

comparison web_to_mqtt.py @ 4:e273cc60b389

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
draft of web-to-lcd and simulator
author drewp@bigasterisk.com
date Tue, 05 Mar 2024 18:12:15 -0800
parents
children d97a5930db7e
comparison
equal deleted inserted replaced
3:045013c772ed 4:e273cc60b389
1 import asyncio
2 import base64
3 import io
4 import subprocess
5 import tempfile
6
7 import aiomqtt
8 from PIL import Image, ImageChops
9
10
11 class WebRenderer:
12 def __init__(self):
13 self.chrome_proc = subprocess.Popen(["google-chrome"])
14 print("Chrome subprocess started.")
15
16 def capture_screenshot(self, url, output_path):
17 out = tempfile.NamedTemporaryFile(suffix=".png")
18 screenshot_command = [
19 "google-chrome",
20 "--headless",
21 f"--screenshot={out.name}",
22 url,
23 ]
24 subprocess.run(screenshot_command)
25 return Image.open(out.name)
26
27
28
29 async def render_webpage_to_png():
30 # Code to render the webpage to a PNG
31 # Replace this with your actual code to render the webpage to a PNG
32 # For example, you can use libraries like Selenium or requests-html to render the webpage and capture a screenshot
33 pass
34
35
36 async def check_for_changes(renderer, last_image):
37
38 renderer.capture_screenshot("https://en.wikipedia.org", "/tmp/output.png")
39
40 current_image = await render_webpage_to_png()
41 if last_image is not None:
42 diff_image = ImageChops.difference(last_image, current_image)
43 # Iterate over 64x64 pixel squares and check for changes
44 for y in range(0, diff_image.height, 64):
45 for x in range(0, diff_image.width, 64):
46 box = (x, y, x + 64, y + 64)
47 region = diff_image.crop(box)
48 if (region.getbbox()
49 ): # Check if region is not empty (i.e., contains changes)
50 # Send changed square as MQTT message
51 queue these
52 await send_mqtt_message(region)
53 return current_image
54
55
56 async def send_mqtt_message(region):
57 # Convert changed region to base64 encoded string
58 buffer = io.BytesIO()
59 region.save(buffer, format="PNG")
60 base64_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
61 mqtt_client = aiomqtt.Client("mqtt_client")
62 await mqtt_client.connect("mqtt://broker.example.com")
63 await mqtt_client.publish("changed_squares", base64_image, qos=1)
64 await mqtt_client.disconnect()
65
66
67 async def main():
68 # also listen for dirty msgs from the web page; see ts
69 renderer = WebRenderer()
70 async with aiomqtt.Client("mqtt_client") as client:
71 last_image = None
72 while True:
73 last_image = await check_for_changes(renderer,last_image)
74 await asyncio.sleep(2) # Adjust the interval as needed
75
76
77 if __name__ == "__main__":
78 asyncio.run(main())