From d8853bf247c5b3628cf07a221ce6a7939019ac2f Mon Sep 17 00:00:00 2001 From: hkc Date: Tue, 9 Jul 2024 16:26:19 +0300 Subject: [PATCH] Added worker --- swarm/manager.py | 7 +++- swarm/worker.py | 100 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 2 deletions(-) create mode 100644 swarm/worker.py diff --git a/swarm/manager.py b/swarm/manager.py index 8306b05..6a89410 100644 --- a/swarm/manager.py +++ b/swarm/manager.py @@ -58,6 +58,7 @@ class Manager: assert self.shmem is not None print("Resetting shmem...") + self.shmem.buf[OFFSET_AVOID:] = bytes(500000 - OFFSET_AVOID) if fontconfig := settings.get("default_font"): self.default_font_size = int(fontconfig.get("size", 8)) @@ -178,6 +179,8 @@ class Manager: elem.get("size", self.default_font_size), ) + shrek_spf = elem.get("spf", 10) + with open(elem["path"], "r") as fp: lyrics = list(map(str.strip, fp)) @@ -187,7 +190,7 @@ class Manager: draw.rectangle((0, 0, 325, 10), fill=(0, 255)) now = datetime.datetime.now(datetime.timezone.utc) line = lyrics[ - int(now.timestamp() / elem["spf"]) % len(lyrics) + int(now.timestamp() / shrek_spf) % len(lyrics) ] draw.text( (2, -1), line, font=shrek_font, fill=(255, 255) @@ -205,7 +208,7 @@ class Manager: sio.on("batched_bit_toggles", self.on_batched_bit_toggles) sio.on("full_state", self.on_full_state) - await sio.connect(f"{self.base}") + await sio.connect(self.base.replace("http", "ws")) await sio.wait() def update_shmem(self, state: bytes): diff --git a/swarm/worker.py b/swarm/worker.py new file mode 100644 index 0000000..4791d0a --- /dev/null +++ b/swarm/worker.py @@ -0,0 +1,100 @@ +import functools +from multiprocessing.shared_memory import SharedMemory +import asyncio +import random +from typing import NamedTuple, Optional +from aiohttp_socks import ProxyConnector +import socketio +import aiohttp +import signal + +OFFSET_STATE = 0 +OFFSET_AVOID = 125000 +OFFSET_CANVAS = 250000 +OFFSET_MASK = 375000 + + +class PixelState(NamedTuple): + state: bool + avoid: bool + canvas: bool + mask: bool + + +class WorkerManager: + def __init__(self, shmem_name: str = "omcb-bot"): + self.shmem_name = shmem_name + self.base = "https://onemillioncheckboxes.com" + self.delay = 0.25 + self.queue: asyncio.Queue[int] = asyncio.Queue(128) + + async def queue_manager(self): + while True: + index = random.randint(0, 999999) + state = self.get_state(index) + if ( + not state.avoid + and state.mask + and (state.state != state.canvas) + ): + await self.queue.put(index) + + async def writer(self, bot_index: int, proxy: Optional[str] = None): + connector = ProxyConnector.from_url(proxy) if proxy else None + async with aiohttp.ClientSession(connector=connector) as http: + sio = socketio.AsyncClient(http_session=http) + + async def writer_itself(): + print("Writer running") + while not sio.connected: + await asyncio.sleep(0.1) + print("Connected and running") + while sio.connected: + index = await self.queue.get() + state = self.get_state(index) + if ( + not state.avoid + and state.mask + and (state.state != state.canvas) + ): + byte, bit = divmod(index, 8) + print("toggle", index) + self.shmem.buf[OFFSET_STATE + byte] ^= 0x80 >> bit + await sio.emit("toggle_bit", {"index": index}) + await asyncio.sleep(self.delay) + + await asyncio.sleep(0.1) + print("Writer closed") + + sio.on("connect", writer_itself) + + await sio.connect(self.base.replace("http", "ws")) + await sio.wait() + + def get_state(self, index: int) -> PixelState: + byte, bit = divmod(index, 8) + mask = 0x80 >> bit + return PixelState( + self.shmem.buf[OFFSET_STATE + byte] & mask != 0, + self.shmem.buf[OFFSET_AVOID + byte] & mask != 0, + self.shmem.buf[OFFSET_CANVAS + byte] & mask != 0, + self.shmem.buf[OFFSET_MASK + byte] & mask != 0, + ) + + async def __aenter__(self): + self.shmem = SharedMemory(self.shmem_name) + return self + + async def __aexit__(self, a, b, c): + self.shmem.close() + + +async def main(): + async with WorkerManager() as mgr: + await asyncio.gather( + mgr.queue_manager(), *[mgr.writer(i) for i in range(4)] + ) + + +if __name__ == "__main__": + asyncio.run(main())