From 5d17b37bea843529fe07dbeed800b10df72c1794 Mon Sep 17 00:00:00 2001 From: hkc Date: Tue, 9 Jul 2024 14:49:48 +0300 Subject: [PATCH] Formatting --- swarm/manager.py | 77 +++++++++++++++++++++++++++++------------------- 1 file changed, 47 insertions(+), 30 deletions(-) diff --git a/swarm/manager.py b/swarm/manager.py index 06783aa..8306b05 100644 --- a/swarm/manager.py +++ b/swarm/manager.py @@ -1,9 +1,8 @@ from multiprocessing.shared_memory import SharedMemory -from typing import Callable, NamedTuple, NewType, Optional +from typing import Callable, NamedTuple, Optional import asyncio import socketio import aiohttp -from aiohttp_socks import ProxyConnector from PIL import ( Image, ImageFont, @@ -32,6 +31,7 @@ OFFSET_AVOID = 125000 OFFSET_CANVAS = 250000 OFFSET_MASK = 375000 + class Manager: def __init__(self, settings_path: str): self.shmem: Optional[SharedMemory] = None @@ -62,8 +62,7 @@ class Manager: if fontconfig := settings.get("default_font"): self.default_font_size = int(fontconfig.get("size", 8)) self.fonts["default", self.default_font_size] = self.get_font( - fontconfig["path"], - self.default_font_size + fontconfig["path"], self.default_font_size ) for avoid in settings.get("avoid", []): @@ -118,7 +117,10 @@ class Manager: self.make_text_image( line["text"], line.get("font", elem.get("font", "default")), - line.get("size", elem.get("size", self.default_font_size)), + line.get( + "size", + elem.get("size", self.default_font_size), + ), line.get( "negative", elem.get("negative", False) ), @@ -132,10 +134,15 @@ class Manager: ) elif elem["type"] in ("anim", "animation"): with Image.open(elem["path"]) as anim: - self.put_animation(elem["x"], elem["y"], [ - frame.convert("LA") - for frame in ImageSequence.Iterator(anim) - ], elem.get("spf", 10)) + self.put_animation( + elem["x"], + elem["y"], + [ + frame.convert("LA") + for frame in ImageSequence.Iterator(anim) + ], + elem.get("spf", 10), + ) elif elem["type"] == "image": with Image.open(elem["path"]).convert("LA") as im: self.put_image(elem["x"], elem["y"], im) @@ -150,19 +157,26 @@ class Manager: def update_time(): now = datetime.datetime.now(datetime.timezone.utc) - self.put_image(pos_x, pos_y, self.make_text_image( - now.strftime(time_format), - time_font, - time_font_size, - time_negative, - time_outline - )) + self.put_image( + pos_x, + pos_y, + self.make_text_image( + now.strftime(time_format), + time_font, + time_font_size, + time_negative, + time_outline, + ), + ) self.animation_functions.append(update_time) elif elem["type"] == "shrek": shrek_x, shrek_y = elem["x"], elem["y"] - shrek_font = self.get_font(elem.get("font", "default"), elem.get("size", self.default_font_size)) + shrek_font = self.get_font( + elem.get("font", "default"), + elem.get("size", self.default_font_size), + ) with open(elem["path"], "r") as fp: lyrics = list(map(str.strip, fp)) @@ -175,7 +189,9 @@ class Manager: line = lyrics[ int(now.timestamp() / elem["spf"]) % len(lyrics) ] - draw.text((2, -1), line, font=shrek_font, fill=(255, 255)) + draw.text( + (2, -1), line, font=shrek_font, fill=(255, 255) + ) self.put_image(shrek_x, shrek_y, im) self.animation_functions.append(update_shrek) @@ -194,7 +210,7 @@ class Manager: def update_shmem(self, state: bytes): assert self.shmem is not None - self.shmem.buf[OFFSET_STATE:OFFSET_STATE + 125000] = state + self.shmem.buf[OFFSET_STATE : OFFSET_STATE + 125000] = state async def on_connect(self): async with aiohttp.ClientSession() as http: @@ -223,7 +239,7 @@ class Manager: for ndx in bits_on: byte, bit = divmod(ndx, 8) - self.shmem.buf[OFFSET_STATE + byte] |= (0x80 >> bit) + self.shmem.buf[OFFSET_STATE + byte] |= 0x80 >> bit for ndx in bits_off: byte, bit = divmod(ndx, 8) self.shmem.buf[OFFSET_STATE + byte] &= 0xFF ^ (0x80 >> bit) @@ -252,7 +268,11 @@ class Manager: (x, y), ( 255 if (buf[OFFSET_MASK + byte] << bit) & 0x80 else 0, - 255 if (buf[OFFSET_CANVAS + byte] << bit) & 0x80 else 0, + ( + 255 + if (buf[OFFSET_CANVAS + byte] << bit) & 0x80 + else 0 + ), 255 if (buf[OFFSET_STATE + byte] << bit) & 0x80 else 0, ), ) @@ -268,7 +288,6 @@ class Manager: im.save("avoid.png") print("Dump done") - def on_sigusr2(self, signum, frame): print("Reloading config") self.reload_config() @@ -290,7 +309,7 @@ class Manager: assert self.shmem is not None for ndx in rng: byte, bit = divmod(ndx, 8) - self.shmem.buf[OFFSET_AVOID + byte] |= (0x80 >> bit) + self.shmem.buf[OFFSET_AVOID + byte] |= 0x80 >> bit def add_avoid_rect(self, sx: int, sy: int, w: int, h: int): for y in range(sy, sy + h): @@ -301,7 +320,7 @@ class Manager: assert self.shmem is not None assert 0 <= index < 1000000 byte, bit = divmod(index, 8) - self.shmem.buf[OFFSET_AVOID + byte] |= (0x80 >> bit) + self.shmem.buf[OFFSET_AVOID + byte] |= 0x80 >> bit def add_avoid_image(self, im: Image.Image): assert im.width == 1000 @@ -317,9 +336,9 @@ class Manager: assert 0 <= index <= 1000000 assert self.shmem is not None byte, bit = divmod(index, 8) - self.shmem.buf[OFFSET_MASK + byte] |= (0x80 >> bit) + self.shmem.buf[OFFSET_MASK + byte] |= 0x80 >> bit if value: - self.shmem.buf[OFFSET_CANVAS + byte] |= (0x80 >> bit) + self.shmem.buf[OFFSET_CANVAS + byte] |= 0x80 >> bit else: self.shmem.buf[OFFSET_CANVAS + byte] &= 0xFF ^ (0x80 >> bit) @@ -430,12 +449,10 @@ async def main(cfg_path: str = "./settings.json", *_): signal.signal(signal.SIGUSR2, mgr.on_sigusr2) print("Listening...") - await asyncio.gather( - mgr.listener(), - mgr.animator() - ) + await asyncio.gather(mgr.listener(), mgr.animator()) if __name__ == "__main__": from sys import argv + asyncio.run(main(*argv[1:]))