Added animation support and blobs

async-bot.py

@@ -1,12 +1,17 @@
 import asyncio
-from typing import Optional
+from typing import NewType, Optional
 from socketio import AsyncSimpleClient
 from aiohttp import ClientSession
 from aiohttp_socks import ProxyConnector
-from PIL import Image, ImageFont, ImageDraw, ImageFilter
+from PIL import Image, ImageFont, ImageDraw, ImageFilter, ImageSequence
 from base64 import b64decode
 from random import choice
 from json import load
+from time import time as time_now
+
+
+PixelMap = NewType("PixelMap", dict[int, bool])
+Animation = NewType("Animation", tuple[list[PixelMap], float])
 
 
 class AsyncBotManager:
@@ -16,11 +21,13 @@ class AsyncBotManager:
         self.canvas = Image.new("1", (1000, 1000))
         self.font = ImageFont.load_default(8)
 
-        self.difference: dict[int, bool] = {}
+        self.difference: PixelMap = PixelMap({})
         self._last_update = 0
         self._shutdown: bool = False
         self.avoid: set[int] = set()
 
+        self.animations: list[Animation] = []
+
     def put_text(self, x: int, y: int, text: str):
         with Image.new("LA", (int(self.font.getlength(text) + 12), 16)) as im:
             draw = ImageDraw.Draw(im)
@@ -43,18 +50,34 @@ class AsyncBotManager:
         self.put_index(x + y * 1000, val)
 
     def put_index(self, index: int, value: bool):
-        if not index in self.avoid:
+        if index not in self.avoid:
             self.difference[index] = value
 
-    def add_avoid_rect(self, sx: int, sy: int, w: int, h: int):
+    def add_animation(
+        self, ox: int, oy: int, image: Image.Image, spf: float = 1
+    ) -> None:
+        animation = Animation(([], spf))
+        for frame in ImageSequence.Iterator(image):
+            frame_la = frame.convert("LA")
+            pixelmap = PixelMap({})
+            for y in range(frame_la.height):
+                for x in range(frame_la.width):
+                    l, a = frame_la.getpixel((x, y))  # type: ignore
+                    index = x + ox + (y + oy) * 1000
+                    if a > 128 and index not in self.avoid:
+                        pixelmap[index] = l > 128
+            animation[0].append(pixelmap)
+        self.animations.append(animation)
+
+    def add_avoid_rect(self, sx: int, sy: int, w: int, h: int) -> None:
         for y in range(sy, sy + h):
             ox = y * 1000
             self.add_avoid_range(sx + ox, sx + w + ox)
 
-    def add_avoid_range(self, start: int, stop: int, step: int = 1):
+    def add_avoid_range(self, start: int, stop: int, step: int = 1) -> None:
         self.avoid |= set(range(start, stop, step))
 
-    def add_avoid_index(self, *indices: int):
+    def add_avoid_index(self, *indices: int) -> None:
         self.avoid |= set(indices)
 
     def get_difference_image(self) -> Image.Image:
@@ -71,13 +94,13 @@ class AsyncBotManager:
                 im.putpixel((x, y), (255, 0, 0))
             return im.copy()
 
-    async def listener(self):
+    async def listener(self) -> None:
         async with ClientSession() as http:
             async with http.get(f"{self.base}/api/initial-state") as req:
                 data = await req.json()
                 buffer = b64decode(data["full_state"].encode() + b"=")
                 self.canvas.paste(Image.frombytes("1", (1000, 1000), buffer))
-                self._last_update: int = data["timestamp"]
+                self._last_update = data["timestamp"]
             async with AsyncSimpleClient(http_session=http) as sio:
                 await sio.connect(f"{self.base}/socket.io")
                 while not self._shutdown:
@@ -87,12 +110,12 @@ class AsyncBotManager:
                         image = Image.frombytes("1", (1000, 1000), buffer)
                         self.canvas.paste(image)
                         image.close()
-                        self._last_update: int = data["timestamp"]
+                        self._last_update = data["timestamp"]
                     elif event == "batched_bit_toggles":
                         bits_on, bits_off, timestamp = data
                         if timestamp < self._last_update:
                             print("SKIPPING UPDATES: TOO OLD")
-                        self._last_update: int = timestamp
+                        self._last_update = timestamp
                         for ndx in bits_on:
                             y, x = divmod(ndx, 1000)
                             self.canvas.putpixel((x, y), 255)
@@ -101,6 +124,11 @@ class AsyncBotManager:
                             self.canvas.putpixel((x, y), 0)
                     else:
                         print("unknown event", event, data)
+                    for pixmaps, spf in self.animations:
+                        frame_index = int(time_now() / spf)
+                        self.difference.update(
+                            pixmaps[frame_index % len(pixmaps)]
+                        )
 
     async def writer(
         self,
@@ -133,13 +161,8 @@ class AsyncBotManager:
         self._shutdown = True
         await self._listener_task
 
-    def copy(self):
-        copy = self.__class__(self.base)
-        copy.difference = dict.copy(self.difference)
-        return copy
 
-
-async def amain():
+async def amain() -> None:
     with open("settings.json", "r") as fp:
         settings = load(fp)
 
@@ -168,6 +191,9 @@ async def amain():
             elif elem["type"] == "image":
                 with Image.open(elem["path"]).convert("LA") as im:
                     mgr.put_image(elem["x"], elem["y"], im)
+            elif elem["type"] == "animation":
+                with Image.open(elem["path"]) as anim:
+                    mgr.add_animation(elem["x"], elem["y"], anim, elem["spf"])
             elif elem["type"] == "rgb111":
                 ox, oy = elem["x"], elem["y"]
                 with Image.open(elem["path"]).convert("RGBA") as im:
@@ -189,8 +215,8 @@ async def amain():
                             r, g, b, a = im.getpixel((x, y))  # type: ignore
                             if a < 128:
                                 continue
-                            ndx_start: int = (x + ox + (y + oy) * 250) * 16
-                            color = (r >> 3) << 13
+                            ndx_start = (x + ox + (y + oy) * 250) * 16
+                            color: int = (r >> 3) << 13
                             color |= (g >> 2) << 5
                             color |= b >> 3
 
@@ -202,6 +228,21 @@ async def amain():
                 for y in range(elem["y"], elem["y"] + elem["h"]):
                     for x in range(elem["x"], elem["x"] + elem["w"]):
                         mgr.put_pixel(x, y, elem["fill"])
+            elif elem["type"] == "blob":
+                with open(elem["path"], "rb") as fp:
+                    offset = elem["offset"]
+                    length = elem.get("length", 1000000)
+                    written = 0
+                    while (char := fp.read(1)):
+                        byte = char[0]
+                        if written > length:
+                            break
+                        for i in range(8):
+                            if written > length:
+                                break
+                            mgr.put_index(offset, bool((byte >> (7 - i)) & 1))
+                            written += 1
+                            offset += 1
 
         mgr.get_difference_image().save("result.png")
         mgr.get_avoid_image().save("avoid.png")

pictures/bit.txt

@@ -0,0 +1 @@
+The bit is the most basic unit of information in computing and digital communication. The name is a portmanteau of binary digit. The bit represents a logical state with one of two possible values. These values are most commonly represented as either "1" or "0", but other representations such as true/false, yes/no, on/off, or +/− are also widely used. The relation between these values and the physical states of the underlying storage or device is a matter of convention, and different assignments may be used even within the same device or program. It may be physically implemented with a two-state device. A contiguous group of binary digits is commonly called a bit string, a bit vector, or a single-dimensional (or multi-dimensional) bit array. A group of eight bits is called one byte, but historically the size of the byte is not strictly defined. Frequently, half, full, double and quadruple words consist of a number of bytes which is a low power of two. A string of four bits is usually a nibble. In information theory, one bit is the information entropy of a random binary variable that is 0 or 1 with equal probability, or the information that is gained when the value of such a variable becomes known. As a unit of information, the bit is also known as a shannon, named after Claude E. Shannon. The symbol for the binary digit is either "bit", per the IEC 80000-13:2008 standard, or the lowercase character "b", per the IEEE 1541-2002 standard. Use of the latter may create confusion with the capital "B" which is the international standard symbol for the byte. 

settings.json

@@ -50,6 +50,12 @@
     }
   ],
   "elements": [
+    {
+      "type": "blob",
+      "path": "./pictures/bit.txt",
+      "offset": 16384,
+      "length": 12632
+    },
     {
       "type": "image",
       "path": "./pictures/kangel.png",
@@ -69,16 +75,24 @@
       "y": 333
     },
     {
-      "type": "image",
-      "path": "./pictures/neko.png",
+      "type": "animation",
+      "path": "./pictures/neko.gif",
       "x": 263,
-      "y": 229
+      "y": 225,
+      "spf": 5
     },
     {
       "type": "image",
       "path": "./pictures/casey.png",
       "x": 256,
       "y": 256
+    },
+    {
+      "type": "animation",
+      "path": "./pictures/loading.gif",
+      "x": 436,
+      "y": 436,
+      "spf": 10
     }
   ]
 }