【嘉楠堪智K230开发板试用体验】高校竞赛-图传基于micropython

使用K230在micropython固件下图传至PC，采用的UDP通信。

概述

用手机热点跑的，个人测试大致情况如下，结果会根据个人网络水平，供电状况等因素浮动。

相关链接

01 studio社区-wifi图传大致讲解

01 studioa教程-socket基本概念及API

勘智开发者社区-UDP通信

代码

本人写的代码有很多冗余，建议根据自身情况删改。

图传的通信方案选择视情况而定，压缩解码方式同理。

因为竞赛一般用到图传的情况是运动控制类题目，尤其在高速运动情况下，所以我认为可以舍弃一些帧争取数据的高速传输，UDP更适合。但UDP和TCP抑或是其他通信方式，一般情况差距不大，选择因人而异。

与TCP相比，UDP单次发送数据规模较小，通常设置为1024字节（<=MTU），对于图像这类动辄上万字节的大数据，通常进行分包处理后发送（以下代码包含分包的发送和接收处理）。

但由于UDP是面向报文的快速的无连接的通信，分包发送下容易造成数据的乱序，建议可根据网络情况添加类似帧编号或时间戳的标志位减少数据乱序，以下代码仅包含帧头和帧大小校验。

# -*- coding: utf-8 -*-
"""
[url=home.php?mod=space&uid=2666770]@Brief[/url]      : K230发送端代码
@date       : 2025-08
[url=home.php?mod=space&uid=644434]@version[/url]    : v1.0
"""
import socket
import os,time
import network
from media.sensor import * 
from media.display import * 
from media.media import * 
​
def network_use_wlan():
    wlan = network.WLAN(network.STA_IF) 
    wlan.active(True)
    if not wlan.isconnected():
        for i in range(5):
            wlan.connect("Xiaomi13", "12341234")
            if wlan.isconnected():
                print("连接wifi成功")
                break
    if wlan.isconnected():
        HOST = wlan.ifconfig()[0]
        print(f'连接成功,IP:{HOST}')
​
def udpclient():
    network_use_wlan()
    
    ai = socket.getaddrinfo('192.168.149.119', 8080)  # 和接收端对应
    addr = ai[0][-1]
    print("Connect address:", addr)
    
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
#    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    s.connect(addr)
    
    sensor = Sensor(width=320, height=240)
    sensor.reset()
    sensor.set_framesize(width=320, height=240)
    sensor.set_pixformat(Sensor.RGB888)
    Display.init(Display.ST7701, to_ide=True)
    MediaManager.init()
    sensor.run()
    
    index = 0
    i=0
    time.sleep(0.2)
    while True:
#        i+=1
#        str="K230 udp client send test {0} \r\n".format(i)
#        print(s.sendto(str,addr))
#        time.sleep(0.2)
                
        img = sensor.snapshot()
        imge=img.compress(quality=25)
#        imge = img.to_jpeg()
        img_len = len(imge)
        
        # ===== 构造包头 (8字节) =====
        magic = 0xABCD1234
        header = magic.to_bytes(4, 'big') + img_len.to_bytes(4, 'big')
        header_len=len(header)
        # ===== 先发包头，再发数据 =====
        s.sendto(header, addr)
        print(f"已经发送帧头大小{header_len}")
        print(f"预备发送数据大小{img_len}")
        
        sent = 0
        while sent < img_len:
            chunk_size = min(1024, img_len - sent)
            chunk = bytes(imge[sent:sent + chunk_size])
            bytes_sent =s.sendto(chunk, addr)
            
            if bytes_sent <= 0:
                print("数据发送失败")
                break
            sent += bytes_sent
            print(f"发送了 {bytes_sent} 数据,还剩{img_len - sent}")
        index += 1
        if index > 1000:
           break
​
    s.close()
    print("end")
​
udpclient()
​

"""
@brief      : PC(windows)接收端端代码
@date       : 2025-08
@version    : v1.0
"""
import socket
import numpy as np
import cv2
import time  # 用于时间统计
​
​
UDP_IP = "192.168.149.119"
UDP_PORT = 8080
​
# 创建UDP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))
​
# 初始化接收缓冲区
pkt_header_size = 8  # 4字节魔数 + 4字节长度
expected_size = 0
received_data = bytearray()
​
print("开始等待第一帧数据...")
​
​
# FPS 相关
frame_count = 0
start_time = time.time()
fps = 0
​
while True:
    data, addr = sock.recvfrom(1024)
    # print(len(data))
    # 处理包头
    if len(data) != 8:
        continue
    elif len(data) == 8:
        # print("正在解析帧头")
        magic = int.from_bytes(data[:4], 'big')
        if magic == 0xABCD1234:
            # print("接收到帧头")
            expected_size = int.from_bytes(data[4:8], 'big')
            received_data = bytearray()
    # 收集数据包
    # print(f"预期收到数据大小{expected_size}")
    received_total=0
    while received_total < expected_size:
        recv_size=min(1024, expected_size - received_total+8)#自己完善吧
        data, addr = sock.recvfrom(recv_size)
        # if len(data)<=0:
        #     print("图像数据接收失败")
        # else:
        #     print(f"已经收到分包{len(data)}，还差{expected_size - received_total}")
        received_data.extend(data) 
        received_total += len(data)
​
    # print(expected_size)
    # 数据接收完成
    if received_total>= expected_size and expected_size > 0:
        # print("完美接收")
        # 解码图像
        # print("正在解码")
        img = cv2.imdecode(np.frombuffer(received_data[:expected_size], np.uint8), cv2.IMREAD_COLOR)
        if img is not None:
            cv2.imshow("Receiver Preview", img)
        
        # 统计 FPS
        frame_count += 1
        elapsed = time.time() - start_time
        if elapsed >= 1.0:  # 每秒更新一次 FPS
            fps = frame_count / elapsed
            print(f"FPS: {fps:.2f}")
            frame_count = 0
            start_time = time.time()
​
        # 重置状态
        expected_size = 0
        received_data = bytearray()
    
    if cv2.waitKey(1) == 27:
        break
​
cv2.destroyAllWindows()

以上就是简单的图传程序吧，如果需要更稳定的传输自行修改。