Add GalaxySnail's write-up

Author: GalaxySnail

players/GalaxySnail/README.md

@@ -0,0 +1,20 @@
+# pylint: disable=used-before-assignment
+import typing
+from typing import Literal, TypeVar, Protocol
+
+if typing.TYPE_CHECKING:
+    from _typeshed import SupportsWrite, ReadableBuffer
+
+
+T = TypeVar("T")
+T_co = TypeVar("T_co", covariant=True)
+T_contra = TypeVar("T_contra", contravariant=True)
+
+# because of `AnyStr`, `typing.IO` can't save us
+class File(SupportsWrite[T_contra], Protocol):
+    def close(self) -> None: ...
+    def flush(self) -> None: ...
+
+
+raise RuntimeError("Never import it at runtime. Use typing.TYPE_CHECKING.")
+

players/GalaxySnail/gist/ANSI.md

@@ -0,0 +1,98 @@
+"""a simple TerminatedFrameReceiver by @njsmith.
+https://github.com/python-trio/trio/issues/796#issuecomment-471428274
+"""
+
+import trio
+
+_RECEIVE_SIZE = 4096  # pretty arbitrary
+
+class TerminatedFrameReceiver:
+    """Parse frames out of a Trio stream, where each frame is terminated by a
+    fixed byte sequence.
+
+    For example, you can parse newline-terminated lines by setting the
+    terminator to b"\\n".
+
+    This uses some tricks to protect against denial of service attacks:
+
+    - It puts a limit on the maximum frame size, to avoid memory overflow; you
+    might want to adjust the limit for your situation.
+
+    - It uses some algorithmic trickiness to avoid "slow loris" attacks. All
+      algorithms are amortized O(n) in the length of the input.
+
+    """
+    def __init__(self, stream, terminator, max_frame_length=16384):
+        self.stream = stream
+        self.terminator = terminator
+        self.max_frame_length = max_frame_length
+        self._buf = bytearray()
+        self._next_find_idx = 0
+
+    async def receive(self):
+        while True:
+            terminator_idx = self._buf.find(
+                self.terminator, self._next_find_idx
+            )
+            if terminator_idx < 0:
+                # no terminator found
+                if len(self._buf) > self.max_frame_length:
+                    raise ValueError("frame too long")
+                # next time, start the search where this one left off
+                self._next_find_idx = max(0, len(self._buf) - len(self.terminator) + 1)
+                # add some more data, then loop around
+                more_data = await self.stream.receive_some(_RECEIVE_SIZE)
+                if more_data == b"":
+                    if self._buf:
+                        raise ValueError("incomplete frame")
+                    raise trio.EndOfChannel
+                self._buf += more_data
+            else:
+                # terminator found in buf, so extract the frame
+                frame = self._buf[:terminator_idx]
+                # Update the buffer in place, to take advantage of bytearray's
+                # optimized delete-from-beginning feature.
+                del self._buf[:terminator_idx+len(self.terminator)]
+                # next time, start the search from the beginning
+                self._next_find_idx = 0
+                return frame
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        try:
+            return await self.receive()
+        except trio.EndOfChannel:
+            raise StopAsyncIteration
+
+
+def example():
+    from trio.testing import memory_stream_pair
+    async def main():
+        sender_stream, receiver_stream = memory_stream_pair()
+
+        async def sender():
+            await sender_stream.send_all(b"hello\r\n\r\n")
+            await trio.sleep(1)
+            await sender_stream.send_all(b"split-up ")
+            await trio.sleep(1)
+            await sender_stream.send_all(b"message\r\n\r")
+            await trio.sleep(1)
+            await sender_stream.send_all(b"\n")
+            await trio.sleep(1)
+            await sender_stream.send_all(b"goodbye\r\n\r\n")
+            await trio.sleep(1)
+            await sender_stream.aclose()
+
+        async def receiver():
+            chan = TerminatedFrameReceiver(receiver_stream, b"\r\n\r\n")
+            async for message in chan:
+                print(f"Got message: {message!r}")
+
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(sender)
+            nursery.start_soon(receiver)
+
+    trio.run(main)
+

players/GalaxySnail/pics/q06-green_hello.png

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import compiler
+from compiler import (
+    Point, CodeIndex,
+    NoArgCode, OneArgCode, TwoArgsCode, CommandCode
+)
+from utils import assert_never
+from code_gen import CodeGenerator
+
+
+import typing
+from typing import Tuple, Iterable
+from typing import BinaryIO, Literal, Final
+if typing.TYPE_CHECKING:
+    from _protos import File as FileProto
+
+
+# 跳转表的写法没法从 mypy 的穷尽性检查中受益，所以放弃
+# CodeHandler = Callable[[CodeGenerator, CommandCode], None]
+
+# mypy 的穷尽性检查甚至还不足以检查包含枚举的元组这样稍微复杂的情况，
+# 因此只能检查枚举是否穷尽，而无法检查枚举是否正确对应了参数 args 的类型
+
+
+@dataclass
+class Assembler:
+    code_gen: CodeGenerator
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.close()
+
+    def close(self):
+        self.code_gen.close()
+
+    @classmethod
+    def from_file(
+        cls,
+        file: FileProto,
+        use_text_io: bool,
+        *,
+        close_when_exit: bool = False
+    ) -> "Assembler":
+        code_gen = CodeGenerator(file, use_text_io,
+                                 close_when_exit=close_when_exit)
+        return cls(code_gen)
+
+    def _gen_code(self, code: CommandCode) -> None:
+        code_index: CodeIndex
+        args: tuple
+        code_index, args = code
+        g: CodeGenerator = self.code_gen
+
+        if code_index is CodeIndex.NOP:
+            pass
+        elif code_index is CodeIndex.UP:
+            g.up(*args)
+        elif code_index is CodeIndex.DOWN:
+            g.down(*args)
+        elif code_index is CodeIndex.LEFT:
+            g.left(*args)
+        elif code_index is CodeIndex.RIGHT:
+            g.right(*args)
+        elif code_index is CodeIndex.DROP:
+            g.drop(*args)
+        elif code_index is CodeIndex.GET:
+            g.get(*args)
+        elif code_index is CodeIndex.GOTO:
+            g.goto(*args)
+        elif code_index is CodeIndex.DROP_PLATE:
+            g.drop_plate()
+        elif code_index is CodeIndex.GET_PLATE:
+            g.get_plate()
+        elif code_index is CodeIndex.STAND_STILL:
+            g.stand_still()
+        elif code_index is CodeIndex.IF_GE_AND_GOTO:
+            g.if_ge_and_goto(*args)
+        else:
+            # 穷尽性检查
+            assert_never(code_index)
+
+    def __call__(self, codes: Iterable[CommandCode]) -> None:
+        for code in codes:
+            self._gen_code(code)
+

players/GalaxySnail/pics/q13_app-vm.png

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, InitVar
+
+from utils import require_positive as _require_positive
+from utils import UnreachableError
+
+
+import typing
+from typing import Generic, TypeVar, cast
+if typing.TYPE_CHECKING:
+    from _typeshed import ReadableBuffer
+    from _protos import File as FileProto
+
+
+__all__ = ["CodeGenerator"]
+
+T_AnyStr = TypeVar("T_AnyStr", str, "ReadableBuffer")
+
+
+@dataclass
+class CodeGenerator(Generic[T_AnyStr]):
+    file: FileProto[T_AnyStr]
+    use_text_io: InitVar[bool] = False
+    close_when_exit: bool = False  # kwonly
+
+    def __post_init__(self, use_text_io: bool) -> None:
+        # use `setattr` instead of assignment to make mypy happy, see:
+        # https://github.com/python/mypy/issues/2427#issuecomment-384229898
+        if use_text_io:
+            setattr(self, "_write", self._write_to_text_io)
+        else:
+            setattr(self, "_write", self._write_to_binary_io)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.close()
+
+    def close(self):
+        self.file.flush()
+        if self.close_when_exit:
+            self.file.close()
+
+    def _write(self, data: str) -> None:
+        """此方法在 __post_init__ 方法中初始化"""
+        raise UnreachableError(CodeGenerator._write.__doc__)
+
+    def _write_to_text_io(self, data: str):
+        # XXX mypy can't check this, we cast it here
+        file = cast("FileProto[str]", self.file)
+        length = len(data)
+        writed = 0
+        while length != writed:
+            writed += file.write(data[writed:])
+
+    def _write_to_binary_io(self, data: str):
+        # XXX mypy can't check this, we cast it here
+        bio = cast("FileProto[ReadableBuffer]", self.file)
+        b = memoryview(data.encode("utf-8"))
+        del data
+        length = len(b)
+        writed = 0
+        while length != writed:
+            writed += bio.write(b[writed:])
+
+    def _up(self, steps: int):
+        self._write(f"向上 {steps} 步\n")
+
+    def up(self, steps: int):
+        _require_positive(steps)
+        self._up(steps)
+
+    def stand_still(self):
+        self._up(0)
+
+    def down(self, steps: int):
+        _require_positive(steps)
+        self._write(f"向下 {steps} 步\n")
+
+    def left(self, steps: int):
+        _require_positive(steps)
+        self._write(f"向左 {steps} 步\n")
+
+    def right(self, steps: int):
+        _require_positive(steps)
+        self._write(f"向右 {steps} 步\n")
+
+    def drop(self, items: int):
+        _require_positive(items)
+        self._write(f"放下 {items} 个物品\n")
+
+    def get(self, items: int):
+        _require_positive(items)
+        self._write(f"拿起 {items} 个物品\n")
+
+    def drop_plate(self):
+        self._write("放下盘子\n")
+
+    def get_plate(self):
+        self._write("拿起盘子\n")
+
+    def _if_ge_and_goto(self, items: int, rel_lineno: int):
+        if rel_lineno == 0:
+            raise ValueError("`rel_lineno` requires a non-zero interger.")
+
+        self._write(f"如果手上的物品大于等于 {items} "
+                    f"向{'上' if rel_lineno < 0 else '下'}"
+                    f"跳转 {abs(rel_lineno)} 行\n")
+
+    def if_ge_and_goto(self, items: int, rel_lineno: int):
+        if items <= 0:
+            raise ValueError(
+                f"`items` requires a positive integer, but got {items}.")
+
+        self._if_ge_and_goto(items, rel_lineno)
+
+    def goto(self, rel_lineno: int):
+        self._if_ge_and_goto(0, rel_lineno)
+