はじめに

第三章です
ついに QRBとSTM32側で双方向な通信をさせました
前の章もお読み下さいませ
第一章
 第二章

コード


# arduino-routerを止める
sudo systemctl stop arduino-router


# python で使うモジュールを持ってくる
sudo apt-get install scrot python3-pil python3-numpy python3-serial


#!/usr/bin/env python3
import serial
import time
import numpy as np
import subprocess
from PIL import Image
import os
import argparse

class TFT_UART:
    CMD_FILL_SCREEN = 0x02
    CMD_DRAW_PARTIAL = 0x05
    
    def __init__(self, port='/dev/ttyHS1', baudrate=921600):
        self.ser = serial.Serial(port, baudrate, timeout=2)
        time.sleep(2)
        self.ser.reset_input_buffer()
        print(f"Serial port {port} opened at {baudrate} bps")
    
    def wait_ack(self, timeout=1.0):
        start = time.time()
        while (time.time() - start) < timeout:
            if self.ser.in_waiting > 0:
                ack = self.ser.read(1)
                if ack[0] == 0xAA:
                    return True
            time.sleep(0.001)
        return False
    
    def fill_screen(self, color):
        payload = bytes([self.CMD_FILL_SCREEN]) + color.to_bytes(2, 'big')
        self.ser.write(payload)
        return self.wait_ack()
    
    def draw_partial(self, x, y, w, h, fb_tile):
        header = bytes([self.CMD_DRAW_PARTIAL]) + \
                 x.to_bytes(2, 'big') + \
                 y.to_bytes(2, 'big') + \
                 w.to_bytes(2, 'big') + \
                 h.to_bytes(2, 'big')
        
        fb_bytes = fb_tile.astype('>u2').tobytes()
        
        self.ser.write(header + fb_bytes)
        
        return self.wait_ack()

class DifferentialUpdater:
    def __init__(self, width=320, height=240, tile_size=40):
        self.width = width
        self.height = height
        self.tile_size = tile_size
        self.prev_frame = None
        self.tiles_x = (width + tile_size - 1) // tile_size
        self.tiles_y = (height + tile_size - 1) // tile_size
    
    def find_changed_tiles(self, current_frame):
        if self.prev_frame is None:
            self.prev_frame = current_frame.copy()
            return [(0, 0, self.width, self.height)]
        
        changed_regions = []
        for ty in range(self.tiles_y):
            for tx in range(self.tiles_x):
                x, y = tx * self.tile_size, ty * self.tile_size
                w = min(self.tile_size, self.width - x)
                h = min(self.tile_size, self.height - y)
                
                if not np.array_equal(current_frame[y:y+h, x:x+w], self.prev_frame[y:y+h, x:x+w]):
                    changed_regions.append((x, y, w, h))
        
        self.prev_frame = current_frame.copy()
        return changed_regions

class X11ScreenCapture:
    def __init__(self, display=None, target_width=320, target_height=240):
        self.display = display or os.environ.get('DISPLAY', ':0')
        self.target_width = target_width
        self.target_height = target_height
        self.temp_file = '/tmp/x11_frame.png'
    
    def capture_to_rgb565(self):
        try:
            subprocess.run(['scrot', '-o', self.temp_file], 
                           env={**os.environ, 'DISPLAY': self.display},
                           check=True, capture_output=True)
            
            with Image.open(self.temp_file) as img:
                img = img.resize((self.target_width, self.target_height), Image.Resampling.NEAREST)
                pixels = np.array(img.convert('RGB'))
            
            r = pixels[:, :, 0].astype(np.uint16)
            g = pixels[:, :, 1].astype(np.uint16)
            b = pixels[:, :, 2].astype(np.uint16)
            
            rgb565 = ((r & 0xF8) << 8) | ((g & 0xFC) << 3) | (b >> 3)
            return rgb565
        except Exception as e:
            print(f"Capture error: {e}")
            return None

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--port', default='/dev/ttyHS1')
    parser.add_argument('--baud', type=int, default=921600)
    parser.add_argument('--fps', type=int, default=10)
    args = parser.parse_args()

    tft = TFT_UART(args.port, args.baud)
    capture = X11ScreenCapture(target_width=320, target_height=240) # ROT90用
    differ = DifferentialUpdater(width=320, height=240, tile_size=40)

    print("Clearing Screen...")
    tft.fill_screen(0x0000)

    frame_interval = 1.0 / args.fps
    
    try:
        while True:
            start_time = time.time()
            
            fb = capture.capture_to_rgb565()
            if fb is not None:
                changes = differ.find_changed_tiles(fb)
                for rect in changes:
                    x, y, w, h = rect
                    tile_data = fb[y:y+h, x:x+w]
                    if not tft.draw_partial(x, y, w, h, tile_data):
                        print("Timeout/Error in transmission")
            
            elapsed = time.time() - start_time
            time.sleep(max(0, frame_interval - elapsed))
            
    except KeyboardInterrupt:
        print("\nExit.")

if __name__ == '__main__':
    main()

まとめ

ここにPOSTが表示されます

コメント無し