Protobuf bytes图像转矩阵及旋转实现疑问求助

问题分析与修正实现 #

你的核心问题是未按照Protobuf定义的行优先存储规则将bytes数据映射为正确的宽×高像素矩阵，当前代码仅将所有像素（或RGB三元组）平铺为一维列表，未还原成二维行结构，导致旋转逻辑完全错误。

修正步骤：

1. 正确解析Bytes为二维像素矩阵
   • 灰度图：每一行包含width个单字节像素，共height行
   • 彩色图：每一行包含width个RGB三元组（每个三元组3字节），共height行
2. 实现对应旋转角度的矩阵变换
   • 针对90/180/270度的逆时针旋转，分别处理二维矩阵的转置与反转逻辑
3. 旋转后重新编码为Bytes
   • 按行优先规则将旋转后的矩阵重新展平为一维Bytes数组

完整修正代码：

from image_pb2 import Image, ImageRotateRequest, ImageService
from concurrent import futures
import grpc

def rotate_image(image, rotation_enum):
    # 提取图像基础信息
    is_color = image.color
    original_width = image.width
    original_height = image.height
    channel_count = 3 if is_color else 1

    # 将Bytes解析为二维像素矩阵（行优先）
    pixel_data = list(image.data)
    image_matrix = []
    
    for row_idx in range(original_height):
        # 计算当前行的起始/结束索引
        start_idx = row_idx * original_width * channel_count
        end_idx = start_idx + original_width * channel_count
        row_pixels = pixel_data[start_idx:end_idx]
        
        if is_color:
            # 彩色图：拆分当前行为RGB三元组列表
            rgb_row = [row_pixels[i:i+3] for i in range(0, len(row_pixels), 3)]
            image_matrix.append(rgb_row)
        else:
            # 灰度图：直接作为单通道行
            image_matrix.append(row_pixels)

    # 根据旋转角度处理矩阵
    rotated_matrix = []
    new_width, new_height = original_width, original_height
    
    if rotation_enum == ImageRotateRequest.NINETY_DEG:
        # 逆时针旋转90度：反转行顺序后转置
        rotated_matrix = list(zip(*image_matrix[::-1]))
        new_width, new_height = original_height, original_width
    elif rotation_enum == ImageRotateRequest.ONE_EIGHTY_DEG:
        # 逆时针旋转180度：反转行顺序+反转每行元素
        rotated_matrix = [row[::-1] for row in image_matrix[::-1]]
    elif rotation_enum == ImageRotateRequest.TWO_SEVENTY_DEG:
        # 逆时针旋转270度：转置后反转行顺序
        rotated_matrix = list(zip(*image_matrix))[::-1]
        new_width, new_height = original_height, original_width
    else:
        # 默认不旋转
        rotated_matrix = image_matrix

    # 将旋转后的矩阵重新编码为Bytes
    flattened_data = []
    for row in rotated_matrix:
        if is_color:
            # 彩色行：展平每个RGB三元组
            flattened_data.extend([pixel for rgb in row for pixel in rgb])
        else:
            # 灰度行：直接展平
            flattened_data.extend(row)

    # 构造返回的Image对象
    return Image(
        color=is_color,
        data=bytes(flattened_data),
        width=new_width,
        height=new_height
    )

class ImageServiceImpl(ImageService):
    def RotateImage(self, request, context):
        return rotate_image(request.image, request.rotation)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    ImageService.add_RotateImageServicer_to_server(ImageServiceImpl(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

关键细节说明：

• 行优先解析：通过计算每行的起始/结束索引，从Bytes中准确截取每一行的像素数据，完全匹配Protobuf定义的存储规则
• 旋转逻辑：
  • 90度逆时针：zip(*matrix[::-1]) 先反转行顺序，再转置矩阵
  • 180度：先反转行顺序，再反转每行内的像素顺序
  • 270度逆时针：先转置矩阵，再反转行顺序
• 编码回Bytes：严格按照行优先规则，将二维矩阵重新展平为一维列表，再转换为Bytes类型

内容的提问来源于stack exchange，提问作者Sahil__