PPMファイルを監視してリアルタイムに画像を見るやつ

README.md

ppmwatcher.py: PPMファイルを監視してリアルタイムに画像を見るやつ

CPU実験でレイトレの出力を見たいので作ったツールです

つかいかた

uv run ppmwatch.py out.ppm (必要package (pygame) が入ってれば他の実行形式でもいいとおもいます)

キー操作:

• 数字キー: 倍率変更
• q/Esc: 終了

めも

• P3(ascii RGB), P6(binary RGB)形式に対応
• キャッシュを効かせて毎回パースしないようにしてる(つもり)

LICENSE

NYSLのもとで公開します。自由に使ってください

ppmwatch.py

# /// script
# dependencies = [ "pygame" ]
# ///

from typing import BinaryIO
import pygame
import sys
# import time

def main():
    if len(sys.argv) < 2:
        print("Usage: ppmwatch <ppm file>")
        sys.exit(1)

    filename = sys.argv[1]
    pygame.init()
    size = 256
    screen = pygame.display.set_mode((size, size))
    pygame.display.set_caption('PPM Watcher')
    clock = pygame.time.Clock()
    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            if event.type == pygame.KEYDOWN:
                if event.key in (pygame.K_ESCAPE, pygame.K_q):
                    running = False
                elif pygame.K_1 <= event.key <= pygame.K_9:
                    size = 128 * (event.key - pygame.K_0)
                    screen = pygame.display.set_mode((size, size))

        try:
            # t = time.time()
            (w, h), data = parse_ppm(filename)
            # t2 = time.time()
            # print(f"Read PPM {filename}: {w}x{h} in {(t2 - t)*1000:.1f} ms")
            image_surface = pygame.image.frombuffer(data, (w, h), 'RGB')
            pygame.display.set_caption(f'PPM Watcher - {filename} ({w}x{h}, {size*100//w}%)')

            scaled = pygame.transform.scale(image_surface, (size, size))
            screen.blit(scaled, (0, 0))
            pygame.display.flip()
        except Exception as e:
            print(f"Error reading PPM file: {e}")

        clock.tick(5) # fps
    pygame.quit()

def read_int(f: BinaryIO):
    v = -1
    while True:
        c = f.read(1)
        if not c:
            return None
        if c in b' \n\r\t':
            if v != -1:
                return v
        elif c == b'#':
            while c not in b'\n\r':
                c = f.read(1)
        else:
            if v == -1:
                v = 0
            v = v * 10 + (ord(c) - ord(b'0'))

def parse_ppm_stream(f: BinaryIO, ofs) -> tuple[tuple[int, int], bytes, tuple[int, int]]:
    f.seek(0)
    ret = b''
    header = f.read(3)
    if header[:2] == b'P6':
        # binary mode
        if ofs == 0:
            w = read_int(f) or 0
            h = read_int(f) or 0
            read_int(f)  # max color value
        else:
            w = h = 0
            f.seek(ofs)
        last_pos = (0, 0)
        while True:
            last_pos = (f.tell(), len(ret))
            c = f.read(1)
            if not c:
                break
            ret += c
        return (w, h), ret, last_pos
    elif header[:2] == b'P3':
        # ascii mode
        if ofs == 0:
            w = read_int(f) or 0
            h = read_int(f) or 0
            read_int(f)  # max color value
        else:
            w = h = 0
            f.seek(ofs)
        last_pos = (0, 0)
        while True:
            last_pos = (f.tell(), len(ret))
            c = read_int(f)
            if c is None:
                break
            ret += bytes([c])
        return (w, h), ret, last_pos
    elif not header:
        # empty file
        return (1, 1), b'\x80\x80\x80', (0, 0)
    else:
        raise ValueError("not supported PPM format")

last: tuple[tuple[int, int], bytes, bytes, tuple[int, int]] | None = None
def parse_ppm(filename):
    global last

    with open(filename, 'rb') as f:
        d = f.read()
        if last is not None:
            dim, last_result, last_data, last_pos = last
            if d[:len(last_data)] == last_data:
                if len(d) == len(last_data):
                    # identical
                    return dim, pad_ppm_bytes(last_result, dim[0], dim[1])
                # common prefix; delta decoding
                _, d2, p = parse_ppm_stream(f, last_pos[0])
                last_result = last_result[:last_pos[1]]
                print(f"Applying delta of {len(d2)} bytes")
                last_result += d2
                # last[2] = d
                p = (p[0], p[1] + last_pos[1])
                last = (dim, last_result, d, p)
                data = pad_ppm_bytes(last_result, dim[0], dim[1])
                return dim, data

        (w, h), ppm, pos = parse_ppm_stream(f, 0)
        data = bytes(i for i in ppm)

        if len(d) > 20:
            last = ((w, h), data, d, pos)
        else:
            last = None

        data = pad_ppm_bytes(data, w, h)

        return (w, h), data

def pad_ppm_bytes(data: bytes, width: int, height: int) -> bytes:
    # make a copy
    data = bytes(data)
    targsize = width * height * 3
    if len(data) < targsize:
        # data += bytes(targsize - len(data))
        data += b'\x80' * (targsize - len(data))
    elif len(data) > targsize:
        print(f"warning: ppm data size {len(data)} larger than expected {targsize}")
        data = data[:targsize]
    return data

if __name__ == '__main__':
    main()

server.py.diff

--- server.py	2022-03-03 17:04:00.000000000 +0900
+++ server2.py	2026-03-10 14:47:30.442338475 +0900
@@ -35,7 +35,7 @@


 class UART:
-    def __init__(self, port: str, one_byte_each: bool, max_n_bytes_per_recv: int, baudrate: int, parity, stopbits, no_progress: bool):
+    def __init__(self, port: str, one_byte_each: bool, max_n_bytes_per_recv: int, baudrate: int, parity, stopbits, no_progress: bool, ppm_fp):
         """
         prepare and initialize UART port

@@ -88,6 +88,8 @@
         self.startTime = None
         self.endTime = None

+        self.ppm_fp = ppm_fp
+
     def _recv_bytes(self, n_bytes: int = 1):
         """
         receive `n_bytes` byte(s) from COM port
@@ -118,6 +120,7 @@
         # write the received byte to raw_bin when it is not None
         if self.raw_bin is not None:
             self.raw_bin += b
+            self.ppm_fp.write(b)

         return b

@@ -521,7 +524,8 @@
     program_bytes = bin_reader(args.program, args.endian, length=True)
     data_bytes = data_reader(args.data, args.endian)

-    uart = UART(args.port, args.one_byte_each, args.max_n_bytes, args.baudrate, args.parity, args.stopbits, args.no_progress)
+    ppm_fp = open(output_name + ".ppm", 'wb')
+    uart = UART(args.port, args.one_byte_each, args.max_n_bytes, args.baudrate, args.parity, args.stopbits, args.no_progress, ppm_fp)

     if program_bytes is not None:
         # wait for a 0x99 byte
@@ -550,9 +554,6 @@
                 txt_fp.write("%3d %3d %3d%s\n" % (r, g, b, comment))
                 comment = ""

-        # save ppm image
-        with open(output_name + ".ppm", 'wb') as ppm_fp:
-            ppm_fp.write(uart.raw_bin)

         # save png image
         np_img = np.array(uart.ppm_img, dtype=np.uint8).reshape(uart.ppm_info["height"], uart.ppm_info["width"], 3)
@@ -563,10 +564,10 @@
         # receive args.raw_output bytes of raw data
         uart.recv_raw_data(args.raw_output)

-        print("saving raw data:")
-        with open(output_name, 'wb') as bin_fp:
-            bin_fp.write(uart.raw_bin)
-        print("\tsaved in \"%s\"." % output_name)

     uart.close()