Python Protobuf图像数据处理与旋转：从Bytes到矩阵的转换

message Image {
    bool color = 1;
    bytes data = 2;
    int32 width = 3;
    int32 height = 4;
}

def bytes_to_matrix(image):
    """
    将protobuf Image消息中的bytes数据转换为图像矩阵。

    Args:
        image: protobuf Image消息。

    Returns:
        一个二维列表，表示图像矩阵。每个元素代表一个像素，彩色图像为RGB元组，灰度图像为单个字节值。
    """
    width = image.width
    height = image.height
    is_color = image.color
    data = image.data

    matrix = []
    current = 0
    for y in range(height):
        row = []
        for x in range(width):
            if is_color:
                # 彩色图像，每个像素3个字节 (RGB)
                pixel = (
                    data[current],
                    data[current + 1],
                    data[current + 2],
                )
                current += 3
            else:
                # 灰度图像，每个像素1个字节
                pixel = data[current]
                current += 1
            row.append(pixel)
        matrix.append(row)
    return matrix

def rotate_matrix_90(matrix):
    """顺时针旋转矩阵90度"""
    return list(zip(*matrix[::-1]))

def rotate_matrix_180(matrix):
    """旋转矩阵180度"""
    return rotate_matrix_90(rotate_matrix_90(matrix))

def rotate_matrix_270(matrix):
    """顺时针旋转矩阵270度"""
    return rotate_matrix_90(rotate_matrix_90(rotate_matrix_90(matrix)))

def matrix_to_bytes(matrix, is_color):
    """
    将图像矩阵转换为bytes数据。

    Args:
        matrix: 图像矩阵。
        is_color: 布尔值，指示图像是彩色还是灰度图像。

    Returns:
        bytes数据。
    """
    pixels = []
    for row in matrix:
        for pixel in row:
            if is_color:
                pixels.extend(pixel)  # pixel is a tuple (R, G, B)
            else:
                pixels.append(pixel)  # pixel is a single byte value
    return bytes(pixels)

import grpc
import image_pb2
import image_pb2_grpc
from concurrent import futures

class ImageService(image_pb2_grpc.ImageServiceServicer):
    def RotateImage(self, request, context):
        image = request.image
        rotation = request.rotation

        # 将bytes数据转换为图像矩阵
        matrix = bytes_to_matrix(image)

        # 根据请求的旋转角度进行旋转
        if rotation == image_pb2.ImageRotateRequest.NINETY_DEG:
            matrix = rotate_matrix_90(matrix)
            new_width = image.height
            new_height = image.width
        elif rotation == image_pb2.ImageRotateRequest.ONE_EIGHTY_DEG:
            matrix = rotate_matrix_180(matrix)
            new_width = image.width
            new_height = image.height
        elif rotation == image_pb2.ImageRotateRequest.TWO_SEVENTY_DEG:
            matrix = rotate_matrix_270(matrix)
            new_width = image.height
            new_height = image.width
        else:
            # 没有旋转，直接返回原始图像
            return image

        # 将旋转后的矩阵转换回bytes数据
        data = matrix_to_bytes(matrix, image.color)

        # 创建新的Image消息
        rotated_image = image_pb2.Image(
            color=image.color,
            data=data,
            width=new_width,
            height=new_height,
        )

        return rotated_image

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

if __name__ == '__main__':
    serve()