Python Bytes Materials

python-bytes/README.md

@@ -0,0 +1,17 @@
+# Bytes Objects: Handling Binary Data in Python
+
+This folder contains code associated with the Real Python tutorial [Bytes Objects: Handling Binary Data in Python](https://realpython.com/python-bytes/).
+
+## Prerequisites
+
+Install the requirements into your virtual environment:
+
+```shell
+(venv) $ python -m pip install -r requirements.txt
+```
+
+Start the Redis server in a Docker container:
+
+```sh
+$ docker run --rm -d -p 6379:6379 redis
+```

python-bytes/checksum.py

@@ -0,0 +1,11 @@
+import hashlib
+import sys
+from pathlib import Path
+
+python_path = Path(sys.executable)
+checksum_path = python_path.with_suffix(".md5")
+
+machine_code = python_path.read_bytes()
+checksum_path.write_bytes(hashlib.md5(machine_code).digest())
+
+print("Saved", checksum_path)

python-bytes/checksum_incremental.py

@@ -0,0 +1,15 @@
+import hashlib
+import sys
+from pathlib import Path
+
+
+def calculate_checksum(path: Path, chunk_size: int = 4096) -> bytes:
+    checksum = hashlib.md5()
+    with path.open(mode="rb") as file:
+        while chunk := file.read(chunk_size):
+            checksum.update(chunk)
+    return checksum.digest()
+
+
+if __name__ == "__main__":
+    print(calculate_checksum(Path(sys.executable)))

python-bytes/embed.py

@@ -0,0 +1,15 @@
+from base64 import b64encode
+from pathlib import Path
+
+
+def embed_image(path: Path, label: str = "") -> str:
+    ascii_string = b64encode(path.read_bytes()).decode("ascii")
+    print(len(path.read_bytes()))
+    print(len(ascii_string))
+    return f"![{label}](data:image/jpeg;base64,{ascii_string})"
+
+
+if __name__ == "__main__":
+    output_path = Path("picture.md")
+    output_path.write_text(embed_image(Path("picture.jpg")))
+    print("Saved", output_path)

python-bytes/find_timestamps.py

@@ -0,0 +1,8 @@
+import re
+from pathlib import Path
+
+binary_data = Path("picture.jpg").read_bytes()
+pattern = rb"\d{4}:\d{2}:\d{2} \d{2}:\d{2}:\d{2}"
+
+for match in re.finditer(pattern, binary_data):
+    print(match)

python-bytes/generate_image.py

@@ -0,0 +1,6 @@
+import numpy as np
+from PIL import Image
+
+pixels = np.random.default_rng().random((1080, 1920, 3)) * 255
+image = Image.fromarray(pixels.astype("uint8"))
+image.save("image.png")

python-bytes/osm_mmap.py

@@ -0,0 +1,28 @@
+from mmap import ACCESS_READ, mmap
+from pathlib import Path
+from typing import Iterator, Self
+
+
+class OpenStreetMap:
+    def __init__(self, path: Path) -> None:
+        self.file = path.open(mode="rb")
+        self.stream = mmap(self.file.fileno(), 0, access=ACCESS_READ)
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.stream.close()
+        self.file.close()
+
+    def __iter__(self) -> Iterator[bytes]:
+        end = 0
+        while (begin := self.stream.find(b"<way", end)) != -1:
+            end = self.stream.find(b"</way>", begin)
+            yield self.stream[begin : end + len(b"</way>")]
+
+
+if __name__ == "__main__":
+    with OpenStreetMap(Path("map.osm")) as osm:
+        for way_tag in osm:
+            print(way_tag)

python-bytes/read_chunks.py

@@ -0,0 +1,15 @@
+from typing import Iterator
+
+
+def read_chunks(filename: str, max_bytes: int = 1024) -> Iterator[bytes]:
+    with open(filename, mode="rb") as file:
+        while True:
+            chunk = file.read(max_bytes)
+            if chunk == b"":
+                break
+            yield chunk
+
+
+if __name__ == "__main__":
+    for chunk in read_chunks("picture.jpg"):
+        print(len(chunk))

python-bytes/redis_client.py

@@ -0,0 +1,144 @@
+import socket
+from functools import partial
+from typing import Callable, Self
+
+
+class RedisClient:
+    def __init__(self, address: str = "localhost", port: int = 6379) -> None:
+        self._socket = socket.create_connection((address, port))
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self._socket.close()
+
+    def flush(self) -> None:
+        self._socket.sendall(flush_command())
+
+    def ping(self) -> None:
+        self._socket.sendall(ping_command())
+
+    def echo(self, message: str) -> None:
+        self._socket.sendall(echo_command(message))
+
+    def last_save(self) -> None:
+        self._socket.sendall(last_save_command())
+
+    def keys(self, pattern: str = "*") -> None:
+        self._socket.sendall(keys_command(pattern))
+
+    def get(self, key: str) -> None:
+        self._socket.sendall(get_command(key))
+
+    def set(self, key: str, value: str, ex: int | None = None) -> None:
+        self._socket.sendall(set_command(key, value, ex))
+
+    def delete(self, *keys: str) -> None:
+        self._socket.sendall(del_command(*keys))
+
+    def get_response(self) -> str | int | None | list:
+        line = bytearray()
+        while not line.endswith(b"\r\n"):
+            line.extend(self._socket.recv(1))
+        match prefix := line[:1]:
+            case b"+" | b"-":
+                return line[1:-2].decode("ascii")
+            case b":":
+                return int(line[1:-2])
+            case b"$":
+                if (length := int(line[1:])) == -1:
+                    return None
+                else:
+                    data = self._socket.recv(length + 2)
+                    return data[:-2].decode("utf-8")
+            case b"*":
+                return [self.get_response() for _ in range(int(line[1:]))]
+            case _:
+                raise ValueError(f"Unsupported type: {prefix}")
+
+
+def flush_command() -> bytes:
+    return array(bulk_string("FLUSHDB"))
+
+
+def ping_command() -> bytes:
+    return array(bulk_string("PING"))
+
+
+def echo_command(message: str) -> bytes:
+    return array(bulk_string("ECHO"), bulk_string(message))
+
+
+def keys_command(pattern: str) -> bytes:
+    return array(bulk_string("KEYS"), bulk_string(pattern))
+
+
+def set_command(key: str, value: str, ex: int | None) -> bytes:
+    items = [
+        bulk_string("SET"),
+        bulk_string(key),
+        bulk_string(value),
+    ]
+    if ex is not None:
+        items.append(bulk_string("EX"))
+        items.append(bulk_string(str(ex)))
+    return array(*items)
+
+
+def get_command(key: str) -> bytes:
+    return array(bulk_string("GET"), bulk_string(key))
+
+
+def del_command(*keys: str) -> bytes:
+    return array(bulk_string("DEL"), *map(bulk_string, keys))
+
+
+def last_save_command() -> bytes:
+    return array(bulk_string("LASTSAVE"))
+
+
+def array(*items: bytes) -> bytes:
+    binary = bytearray(f"*{len(items)}\r\n".encode("ascii"))
+    for item in items:
+        binary.extend(item)
+    return bytes(binary)
+
+
+def bulk_string(value: str) -> bytes:
+    binary = value.encode("utf-8")
+    return f"${len(binary)}\r\n".encode("utf-8") + binary + b"\r\n"
+
+
+def simple_string(value: str) -> bytes:
+    return f"+{value}\r\n".encode("ascii", errors="strict")
+
+
+def integer(value: int) -> bytes:
+    return f":{value}\r\n".encode("ascii")
+
+
+if __name__ == "__main__":
+    with RedisClient() as redis:
+        commands: list[Callable] = [
+            redis.ping,
+            partial(redis.echo, "Café"),
+            redis.last_save,
+            redis.flush,
+            partial(redis.get, "key"),
+            partial(redis.set, "key1", "value1"),
+            partial(redis.set, "key2", "value2"),
+            partial(redis.get, "key1"),
+            redis.keys,
+            partial(redis.delete, "key1", "key2"),
+            partial(redis.set, "key3", "value3", 5),
+        ]
+        for command in commands:
+            if isinstance(command, partial):
+                args = " ".join([str(x) for x in command.args])
+                print(f"{command.func.__name__.upper()} {args}")
+            else:
+                print(f"{command.__name__.upper()}")
+            command()
+            print(repr(redis.get_response()))
+            print()

python-bytes/requirements.txt

@@ -0,0 +1,2 @@
+numpy==2.2.2
+pillow==11.1.0

python-bytes/user_status.py

@@ -0,0 +1,17 @@
+class UserStatus:
+    def __init__(self, user_id: int, message: str) -> None:
+        self.user_id = user_id
+        self.message = message
+
+    def __bytes__(self) -> bytes:
+        return b"".join(
+            [
+                self.user_id.to_bytes(4, "little", signed=False),
+                self.message.encode("utf-8"),
+            ]
+        )
+
+
+if __name__ == "__main__":
+    user_status = UserStatus(42, "Away from keyboard \N{PALM TREE}")
+    print(bytes(user_status))