【得捷电子Follow me第4期】成果展示汇总帖

shazhongjin

【得捷电子Follow me第4期】成果展示汇总帖 [复制链接]

**首先感谢 DigiKey得捷 | 电子工程世界EEWorld 举办的Follow me活动，与得捷电子一起解锁开发板的超能力！**

# 前言

<font color = red >**龙年新春到，福气满天飘。愿所有看到此贴的人事业如龙腾飞跃，财源似龙脉绵长。家庭和睦如龙凤呈祥，身体健康赛龙马精神。祝你在新的一年里，龙运亨通，万事如意！**</font>:Onion-54: :Onion-54: :Onion-54:
# 项目介绍

本项目包含以下内容：

入门任务：开发环境搭建，BLINK，驱动液晶显示器进行显示（没有则串口HelloWorld）
搭配器件： [W5500-EVB-Pico](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824)、 [Adafruit Sharp Memory Display Breakout](https://www.digikey.cn/zh/products/detail/adafruit-industries-llc/3502/7386264)

基础任务一：完成主控板W5500初始化(静态IP配置)，并能使用局域网电脑ping通，同时W5500可以ping通互联网站点；通过抓包软件(Wireshark、Sniffer等)抓取本地PC的ping报文，展示并分析。
搭配器件： [W5500-EVB-Pico](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824)、 [Adafruit Sharp Memory Display Breakout](https://www.digikey.cn/zh/products/detail/adafruit-industries-llc/3502/7386264)

基础任务二：主控板建立TCPIP或UDP服务器，局域网PC使用TCPIP或UDP客户端进行连接并发送数据，主控板接收到数据后，送液晶屏显示(没有则通过串口打印显示)；通过抓包软件抓取交互报文，展示并分析。(TCP和UDP二选一，或者全都操作)
搭配器件： [W5500-EVB-Pico](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824)、 [Adafruit Sharp Memory Display Breakout](https://www.digikey.cn/zh/products/detail/adafruit-industries-llc/3502/7386264)

进阶任务：从NTP服务器(注意数据交互格式的解析)同步时间，获取时间送显示屏(串口)显示。
搭配器件： [W5500-EVB-Pico](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824)、 [Adafruit Sharp Memory Display Breakout](https://www.digikey.cn/zh/products/detail/adafruit-industries-llc/3502/7386264)

终极任务（二选一）
■  **终极任务一：**访问 [https://www.digikey.cn/zh/resources/api-solutions](https://www.digikey.cn/zh/resources/api-solutions)，以了解得捷(DigiKey)数字化解决方案及其API操作，设计一款DigiKey电子元器件价格及库存监视器，能实时同步并显示指定，电子元器件的价格、库存等信息。

■  **终极任务二：**使用外部存储器，组建简易FTP文件服务器，并能正常上传下载文件。
搭配器件： [W5500-EVB-Pico](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824)、 [Adafruit Sharp Memory Display Breakout](https://www.digikey.cn/zh/products/detail/adafruit-industries-llc/3502/7386264)

本项目视频及源码：

- 视频 [【得捷Follow me第4期】+ 任务成果视频展示](https://training.eeworld.com.cn/video/39268)
- 源码 [【得捷Follow me第4期】+ 任务成果代码展示](https://download.eeworld.com.cn/detail/shazhongjin/631134)

# 开箱展示

购买的配件有：
[W5500-EVB-Pico](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824) 
[MICROSD CARD BFF QT PY/XIAO|](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103100142/13688265) 
[Pico LCD 1.14](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103030400/14317043) 
[Pico RTC DS3231](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103030398/14317036)

开箱全家福：

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786273" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786273&handlekey=savephoto_786273">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786274" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786274&handlekey=savephoto_786274">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

# 任务成果展示及说明

## 入门任务

开发环境搭建，BLINK，驱动液晶显示器进行显示（没有则串口HelloWorld）

### 一、硬件介绍

| 硬件           | 链接、资料                                                                                                                                                                                                                                           | 图片                                      |
| -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------- |
| W5500-EVB-Pico | [W5500-EVB-PICO WIZnet 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824) </br> [W5500-EVB-Pico WIZnet Document System](https://docs.wiznet.io/Product/iEthernet/W5500/w5500-evb-pico) | ![](https://docs.wiznet.io/assets/images/w5500_evb_pico_side-da676c5d9c41adedc0469b9f1810b81b.png) |
| Pico LCD 1.14  | [103030400 Seeed Technology Co., Ltd 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103030400/14317043) </br> [Pico LCD 1.14 - Waveshare Wiki](https://www.waveshare.com/wiki/Pico-LCD-1.14)  | ![](https://www.waveshare.com/media/catalog/product/cache/1/image/800x800/9df78eab33525d08d6e5fb8d27136e95/p/i/pico-lcd-1.14-2_4.jpg) |

### 二、搭建开发环境
本项目使用MicroPython进行开发。
1. 下载 Wiznet W5500-EVB-Pico MicroPython固件 [MicroPython - Python for microcontrollers](https://micropython.org/download/W5500_EVB_PICO/)

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786277_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>Pasted image 20240120141518.png</strong> <em class="xg1">(564.23 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786277" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786277&handlekey=savephoto_786277">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

2. 按住核心板的boot键并连接数据线到电脑，或者已连接电脑状态下按住boot键再按下Run（reset）键。然后将下载的固件应复制到出现的 USB 大容量存储设备中，新固件将自动写入并复位。
3. 将Wiznet W5500-EVB-Pico焊接排针，Pico LCD 1.14 显示模块与Wiznet W5500-EVB-Pico进行连接。需要注意的是核心板焊接排针的方向及与显示模块的针脚的对应。

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786276_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>116091bf84fabce735bb2e902197e62.jpg</strong> <em class="xg1">(6.05 MB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786276" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786276&handlekey=savephoto_786276">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

### 三、代码展示

/beginner_task_code.py

``` python
import LCD_1inch14,time
from machine import Pin,SPI

led = Pin(25, Pin.OUT)

if __name__=='__main__':
    
    lcd = LCD_1inch14.LCD()
    
    lcd.WriteLine('Follow Me 4')
    lcd.WriteLine('by shazhongjin')
    lcd.WriteLine('Hello World!')

while True:
        led.value(1)
        lcd.WriteLine("LED ON")
        time.sleep(1)
        led.value(0)
        lcd.WriteLine("LED OFF")
        time.sleep(1)    
```

本项目使用到的LCD_1inch14库为根据微雪例程 [Demo codes](https://files.waveshare.com/upload/2/28/Pico_code.7z) , 屏幕显示参考Follow me 2中Adafruit ESP32-S3 TFT Feather 开发板 刷入CircuitPython的屏幕显示效果来对此库进行调整。设计思路如下：

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786278" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786278&handlekey=savephoto_786278">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

项目中所有用到屏幕的任务均使用该库，后续不在提及。

/lib/LCD_1inch14.py

``` python
from machine import Pin,SPI,PWM
import framebuf,time

BL = 13
DC = 8
RST = 12
MOSI = 11
SCK = 10
CS = 9
Row = 1

class LCD(framebuf.FrameBuffer):
    
    def __init__(self):
        self.width = 240
        self.height = 135
        
        self.cs = Pin(CS,Pin.OUT)
        self.rst = Pin(RST,Pin.OUT)
        
        self.cs(1)
        self.spi = SPI(1)
        self.spi = SPI(1,1000_000)
        self.spi = SPI(1,10000_000,polarity=0, phase=0,sck=Pin(SCK),mosi=Pin(MOSI),miso=None)
        self.dc = Pin(DC,Pin.OUT)
        self.dc(1)
        self.buffer = bytearray(self.height * self.width * 2)
        super().__init__(self.buffer, self.width, self.height, framebuf.RGB565)
        self.init_display()
        
        self.red   =   0x07E0
        self.green =   0x001f
        self.blue  =   0xf800
        self.white =   0xffff
        self.black =   0x0000
        
    def write_cmd(self, cmd):
        self.cs(1)
        self.dc(0)
        self.cs(0)
        self.spi.write(bytearray([cmd]))
        self.cs(1)

def write_data(self, buf):
        self.cs(1)
        self.dc(1)
        self.cs(0)
        self.spi.write(bytearray([buf]))
        self.cs(1)

def init_display(self):
        """Initialize dispaly"""  
        self.rst(1)
        self.rst(0)
        self.rst(1)
        
        self.write_cmd(0x36)
        self.write_data(0x70)

self.write_cmd(0x3A) 
        self.write_data(0x05)

self.write_cmd(0xB2)
        self.write_data(0x0C)
        self.write_data(0x0C)
        self.write_data(0x00)
        self.write_data(0x33)
        self.write_data(0x33)

self.write_cmd(0xB7)
        self.write_data(0x35)

self.write_cmd(0xBB)
        self.write_data(0x19)

self.write_cmd(0xC0)
        self.write_data(0x2C)

self.write_cmd(0xC2)
        self.write_data(0x01)

self.write_cmd(0xC3)
        self.write_data(0x12)

self.write_cmd(0xC4)
        self.write_data(0x20)

self.write_cmd(0xC6)
        self.write_data(0x0F)

self.write_cmd(0xD0)
        self.write_data(0xA4)
        self.write_data(0xA1)

self.write_cmd(0xE0)
        self.write_data(0xD0)
        self.write_data(0x04)
        self.write_data(0x0D)
        self.write_data(0x11)
        self.write_data(0x13)
        self.write_data(0x2B)
        self.write_data(0x3F)
        self.write_data(0x54)
        self.write_data(0x4C)
        self.write_data(0x18)
        self.write_data(0x0D)
        self.write_data(0x0B)
        self.write_data(0x1F)
        self.write_data(0x23)

self.write_cmd(0xE1)
        self.write_data(0xD0)
        self.write_data(0x04)
        self.write_data(0x0C)
        self.write_data(0x11)
        self.write_data(0x13)
        self.write_data(0x2C)
        self.write_data(0x3F)
        self.write_data(0x44)
        self.write_data(0x51)
        self.write_data(0x2F)
        self.write_data(0x1F)
        self.write_data(0x1F)
        self.write_data(0x20)
        self.write_data(0x23)
        
        self.write_cmd(0x21)

self.write_cmd(0x11)

self.write_cmd(0x29)
        
        # 初始化内容
        Row = 1
        
        self.fill(0x0000)
        self.text("MicroPython",20,4,0xffff)
        self.fill_rect(1,1,3,13,0xffff)
        self.fill_rect(4,1,1,2,0xffff)
        self.fill_rect(5,1,2,13,0xffff)
        self.fill_rect(7,11,2,3,0xffff)
        self.fill_rect(8,1,2,13,0xffff)
        self.fill_rect(10,1,1,2,0xffff)
        self.fill_rect(11,1,3,13,0xffff)
        self.fill_rect(12,10,1,3,0x0000)
        self.show()

def show(self):
        self.write_cmd(0x2A)
        self.write_data(0x00)
        self.write_data(0x28)
        self.write_data(0x01)
        self.write_data(0x17)
        
        self.write_cmd(0x2B)
        self.write_data(0x00)
        self.write_data(0x35)
        self.write_data(0x00)
        self.write_data(0xBB)
        
        self.write_cmd(0x2C)
        
        self.cs(1)
        self.dc(1)
        self.cs(0)
        self.spi.write(self.buffer)
        self.cs(1)
        
    def Clean(self):
        self.init_display()
        
    def WriteLine(self, s : str):
        import math
        global Row
        #print(f'print string is : {s}')
        length = len(s)
        #print(f'string length is : {length}')
        rownum= math.ceil(length / 30)
        #print(f'string rows : {rownum}')
        
        for i in range(rownum):
            i += 1
            print(s[30 * (i - 1) :30 * i])
            
            # 渲染头部
            self.fill_rect(0,0,350,15,0x0000)
            self.text("MicroPython",20,4,0xffff)
            self.fill_rect(1,1,3,13,0xffff)
            self.fill_rect(4,1,1,2,0xffff)
            self.fill_rect(5,1,2,13,0xffff)
            self.fill_rect(7,11,2,3,0xffff)
            self.fill_rect(8,1,2,13,0xffff)
            self.fill_rect(10,1,1,2,0xffff)
            self.fill_rect(11,1,3,13,0xffff)
            self.fill_rect(12,10,1,3,0x0000)

# 当i大于12行时屏幕开始滚动
            #print(f'screen rows : {Row}')
            
            if (Row > 10):
                self.scroll(0,-1 * 12)
                self.fill_rect(0,124,350, 10 ,0x0000)
                    # 渲染头部
                self.fill_rect(0,0,350,15,0x0000)
                self.text("MicroPython",20,4,0xffff)
                self.fill_rect(1,1,3,13,0xffff)
                self.fill_rect(4,1,1,2,0xffff)
                self.fill_rect(5,1,2,13,0xffff)
                self.fill_rect(7,11,2,3,0xffff)
                self.fill_rect(8,1,2,13,0xffff)
                self.fill_rect(10,1,1,2,0xffff)
                self.fill_rect(11,1,3,13,0xffff)
                self.fill_rect(12,10,1,3,0x0000)

if (Row < 10):
                self.text(s[30 * (i - 1) :30 * i], 0 , 15 + (Row-1) * 12 + 2 ,0xffff);
            else:
                self.text(s[30 * (i - 1) :30 * i], 0 , 15 + (10-1) * 12 + 2 ,0xffff);
            Row += 1
            
        self.show()
        #print('\n')
```
### 四、演示

---
参考资料：
[W5500-EVB-Pico | WIZnet Document System](https://docs.wiznet.io/Product/iEthernet/W5500/w5500-evb-pico)
[Pico LCD 1.14 - Waveshare Wiki](https://www.waveshare.com/wiki/Pico-LCD-1.14)

## 基础任务一

完成主控板W5500初始化(静态IP配置)，并能使用局域网电脑ping通，同时W5500可以ping通互联网站点；通过抓包软件(Wireshark、Sniffer等)抓取本地PC的ping报文，展示并分析。

### 一、W5500初始化并配置静态IP
根据WiZnet官方例程，配置静态IP先要查询当前网络网关（比如当前网络网关为192.168.1.1），使用`nic.ifconfig(('IP','子网掩码','网关','DNS'))`来配置w5500的静态ip地址。如下例程：

``` python
from machine import Pin,SPI

spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
nic = network.WIZNET5K() #spi,cs,reset pin  # network.WIZNET5K(spi,Pin(17),Pin(20))
nic.active(True)

# 设置动态ip
#nic.ifconfig('dhcp')

# 设置静态ip
nic.ifconfig(('192.168.1.107','255.255.255.0','192.168.1.1','8.8.8.8'))

while not nic.isconnected():
	time.sleep(1)
	
print(nic.ifconfig())
```

### 二、使用局域网电脑ping通W5500-EVB-Pico，并使用Wireshark抓包ping报文
ping命令通常被用来测试网络连接，可以帮助你确定电脑是否能够通过IP地址访问另一台电脑。当你执行ping命令时，它会发送一个小数据包到特定的IP地址上，并等待响应。如果连接成功，你会看到回复的时间以及数据包是否成功送达。

运行ping命令 地址192.168.1.108

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786259_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>Pasted image 20240217124650.png</strong> <em class="xg1">(440.7 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786259" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786259&handlekey=savephoto_786259">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

双击数据表列表，查看数据包各层详细信息。

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786263_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>Snipaste_2024-02-17_13-14-50.png</strong> <em class="xg1">(76.82 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786263" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786263&handlekey=savephoto_786263">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

互联网层IP包头部信息

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786264_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>Snipaste_2024-02-17_13-37-47.png</strong> <em class="xg1">(103.38 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786264" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786264&handlekey=savephoto_786264">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

传输层ICMP协议信息

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786265_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>Snipaste_2024-02-17_13-32-58.png</strong> <em class="xg1">(100.77 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786265" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786265&handlekey=savephoto_786265">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

### 三、使用W5500-EVB-Pico ping通局域网中的电脑
在gitee中找到一个uping的仓库,可以实现像电脑中的Ping命令一样在开发板中运行ping命令。由于该库中将ping信息输出到控制台上。再此也是做了小小的改动将信息输出到屏幕上，详见下节代码展示。

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786279" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786279&handlekey=savephoto_786279">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

### 四、代码展示

/basic_task1_code.py

``` python
from machine import Pin,SPI,RTC
import time,network,uping
import LCD_1inch14

#W5x00 chip init
def w5x00_init():
    spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
    nic = network.WIZNET5K() #spi,cs,reset pin  # spi,Pin(17),Pin(20)
    nic.active(True)
    
    # If you use the Dynamic IP(DHCP), you must use the "nic.ifconfig('dhcp')".
    #nic.ifconfig('dhcp')
    
    # If you use the Static IP, you must use the  "nic.ifconfig("IP","subnet","Gateway","DNS")".
    nic.ifconfig(('192.168.1.108','255.255.255.0','192.168.1.1','8.8.8.8'))
    
    while not nic.isconnected():
        time.sleep(1)
        #print(nic.regs())
    print(nic.ifconfig())

if __name__ == "__main__":
    lcd = LCD_1inch14.LCD()
    
    lcd.WriteLine('Follow Me 4')
    lcd.WriteLine('by shazhongjin')
    lcd.WriteLine('base task 1')
    
    w5x00_init()
    
    # 使用W5500 ping互联网站点 www.baidu.com
    ping = uping.Ping('www.baidu.com',COUNT=100,IsOutputScreen =True)
    ping.start()
```

/lib/uping.py

``` python
'''
Author: Yogile
Gitee: https://gitee.com/yogile
Date: 2022-07-16 17:41:38
LastEditors: Yogile
LastEditTime: 2022-07-16 18:04:28
Fork from: https://github.com/coffee-it/uPing
Description: Porting code to ESP32 chips
'''

import utime
import uselect
import uctypes
import usocket
import ustruct
import urandom
import micropython
import LCD_1inch14
urandom.seed(utime.ticks_us())

class Ping():
    """
    Create the ping calculating object exemplar
    """
    def __init__(self, HOST, SOURCE=None, COUNT=4, INTERVAL=1000, SIZE=64, TIMEOUT=5000, quiet=False, IsOutputScreen = False):
        self.HOST     = HOST
        self.COUNT    = COUNT
        self.TIMEOUT  = TIMEOUT
        self.INTERVAL = INTERVAL
        self.SIZE     = SIZE
        self.quiet    = quiet
        self.END      = None
        self.IsOutputScreen = IsOutputScreen
        
        # prepare packet
        assert SIZE >= 16, "pkt size too small"
        self._PKT = b'Q'*SIZE
        self.PKT_DESC = {
            "type": uctypes.UINT8 | 0,
            "code": uctypes.UINT8 | 1,
            "checksum": uctypes.UINT16 | 2,
            "id": uctypes.UINT16 | 4,
            "seq": uctypes.INT16 | 6,
            "timestamp": uctypes.UINT64 | 8,
        } # packet header descriptor
        h = uctypes.struct(uctypes.addressof(self._PKT), self.PKT_DESC, uctypes.BIG_ENDIAN)
        h.type = 8 # ICMP_ECHO_REQUEST
        h.code = 0
        h.checksum = 0
        h.id = urandom.getrandbits(16)
        h.seq = 1
        self.h = h

# init socket
        sock = usocket.socket(usocket.AF_INET, usocket.SOCK_RAW, 1)
        if SOURCE:
            src_addr = usocket.getaddrinfo(SOURCE, 1)[0][-1] # ip address
            sock.bind(src_addr)
        sock.setblocking(0)
        sock.settimeout(TIMEOUT/1000)
        addresses = usocket.getaddrinfo(HOST, 1) # [0][-1] # list of ip addresses
        assert addresses, "Can not take the IP address of host"
        self.CLIENT_IP = None
        for addr in addresses:
            try:
                sock.connect(addr[-1])
                self.CLIENT_IP = addr[-1][0]
            except:
                continue
        assert self.CLIENT_IP, "Connection failed"
        self.sock = sock

# [ COUNTERS ]
        self.seq_num = 1 # Next sequence number
        self.transmitted = 0
        self.received = 0
        self.seqs = None
        
        # LCD
        self.LCD = LCD_1inch14.LCD()
        
    
    def Output(self,msg :str):
        print(msg)
        
        if self.IsOutputScreen:
            self.LCD.WriteLine(msg) 
    
    def start(self):
        """
        Starting a ping cycle with the specified settings (like a interval, count e.t.c.)
        """
        t = self.INTERVAL
        finish = False

# [ Rtt metrics ]
        pongs = []
        min_rtt, max_rtt = None, None
        # [ Start over ]
        self.seq_num = 1
        self.transmitted = 0
        self.received = 0
        
        not self.quiet and self.Output("PING %s (%s): %u data bytes" % (self.HOST, self.CLIENT_IP, len(self._PKT)))

if self.seqs:
            self.seqs.extend(list(range(self.seq_num, self.COUNT + 1))) # [seq_num, seq_num + 1, ...., seq_num + n])
        else:
            self.seqs = list(range(self.seq_num, self.COUNT + 1)) # [1,2,...,count]

# Здесь нужно подвязаться на реальное время
        while self.seq_num <= self.seq_num + self.COUNT:
            if t >= self.INTERVAL:
                pong = self.ping()
                t = 0
                rtt = pong[1]
                if rtt:
                    pongs.append(rtt)
                    if not min_rtt or rtt <= min_rtt: min_rtt = round(rtt, 3)
                    if not max_rtt or rtt >= max_rtt: max_rtt = round(rtt, 3)
                if len(self.seqs) == 0:
                    finish = True
            if finish:
                break

utime.sleep_ms(1)
            t += 1

losses = round((self.transmitted - self.received) / self.transmitted) * 100
        avg_rtt = round(sum(pongs) / len(pongs), 3) if pongs else None
        from ucollections import namedtuple
        _result = namedtuple("result", ("tx", "rx", "losses", "min", "avg", "max"))
        result = _result(self.transmitted, self.received, losses, min_rtt, avg_rtt, max_rtt)
        if not self.quiet:
            self.Output(r'%u packets transmitted, %u packets received, %u packet loss' % (self.transmitted, self.received, losses))
            if avg_rtt: self.Output(r'round-trip min/avg/max = %r/%r/%r ms' % (min_rtt, avg_rtt, max_rtt))
        else:
            return result

def ping(self):
        """
        Send ping manually.
        Returns sequense number(int), round-trip time (ms, float), ttl
        """
        sock = self.sock
        if not self.seqs:
            self.seqs = []
            self.seqs.append(self.seq_num)
        seq, t_elasped, ttl = None, None, None

# header
        h = self.h
        h.checksum = 0
        h.seq = self.seq_num
        h.timestamp = utime.ticks_us()
        h.checksum = self.checksum(self._PKT)

try:
            # send packet
            if sock.send(self._PKT) == self.SIZE:
                self.transmitted += 1
            else:
                self.seqs.remove(self.seq_num)
            self.seq_num += 1

# recv packet
            while 1:
                resp = sock.recv(self.SIZE + 20) # ICMP header and payload + IP header
                resp_mv = memoryview(resp)
                h2 = uctypes.struct(uctypes.addressof(resp_mv[20:]), self.PKT_DESC, uctypes.BIG_ENDIAN)
                seq = h2.seq
                if h2.type==0 and h2.id==h.id and (seq in self.seqs): # 0: ICMP_ECHO_REPLY
                    t_elasped = (utime.ticks_us()-h2.timestamp) / 1000
                    self.seqs.remove(seq)
                    if h2.checksum == self.checksum(resp[24:]): # except IP header and a part of ICMP header (type, code, checksum)
                        ttl = ustruct.unpack('!B', resp_mv[8:9])[0] # time-to-live
                        self.received += 1
                        not self.quiet and self.Output("%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms" % (len(resp[12:]), self.CLIENT_IP, seq, ttl, t_elasped))
                        break
                    else:
                        not self.quiet and self.Output("Payload checksum doesnt match")
                        t_elasped = None
                        break

except Exception as identifier:
            import errno
            if identifier.args[0] == errno.EPIPE:
                self.Output("Client connection unexpectedly closed")
                pass
            elif identifier.args[0] == errno.EBADF:
                self.Output("Bad file descriptor.")
                pass
            else:
                raise identifier
        return (seq, t_elasped, ttl)
    
    async def ping_async(self):
        """
        Send ping manually by async.
        Returns sequense number(int), round-trip time (ms, float), ttl
        """
        sock = self.sock
        if not self.seqs:
            self.seqs = []
            self.seqs.append(self.seq_num)
        seq, t_elasped, ttl = None, None, None

# header
        h = self.h
        h.checksum = 0
        h.seq = self.seq_num
        h.timestamp = utime.ticks_us()
        h.checksum = self.checksum(self._PKT)

try:
            # send packet
            if sock.send(self._PKT) == self.SIZE:
                self.transmitted += 1
            else:
                self.seqs.remove(self.seq_num)
            self.seq_num += 1

# recv packet
            while 1:
                resp = sock.recv(self.SIZE + 20) # ICMP header and payload + IP header
                resp_mv = memoryview(resp)
                h2 = uctypes.struct(uctypes.addressof(resp_mv[20:]), self.PKT_DESC, uctypes.BIG_ENDIAN)
                seq = h2.seq
                if h2.type==0 and h2.id==h.id and (seq in self.seqs): # 0: ICMP_ECHO_REPLY
                    t_elasped = (utime.ticks_us()-h2.timestamp) / 1000
                    self.seqs.remove(seq)
                    if h2.checksum == self.checksum(resp[24:]): # except IP header and a part of ICMP header (type, code, checksum)
                        ttl = ustruct.unpack('!B', resp_mv[8:9])[0] # time-to-live
                        self.received += 1
                        not self.quiet and self.Output("%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms" % (len(resp[12:]), self.CLIENT_IP, seq, ttl, t_elasped))
                        break
                    else:
                        not self.quiet and self.Output("Payload checksum doesnt match")
                        t_elasped = None
                        break

except Exception as identifier:
            import errno
            if identifier.args[0] == errno.EPIPE:
                self.Output("Client connection unexpectedly closed")
                pass
            elif identifier.args[0] == errno.EBADF:
                self.Output("Bad file descriptor.")
                pass
            else:
                raise identifier
        self.END = (seq, t_elasped, ttl)

def checksum(self, data):
        if len(data) & 0x1: # Odd number of bytes
            data += b'\0'
        cs = 0
        for pos in range(0, len(data), 2):
            b1 = data[pos]
            b2 = data[pos + 1]
            cs += (b1 << 8) + b2
        while cs >= 0x10000:
            cs = (cs & 0xffff) + (cs >> 16)
        cs = ~cs & 0xffff
        return cs

def getEND_async(self):
        """
        get ping_async() result
        """
        return self.END

def __enter__(self):
        return self

def __exit__(self, type, value, traceback):
        self.close()

def close(self):
        self.sock.close()
```

### 五、演示

<div id="player_13056f04db15d23721687675e57b68381" class="localvideo-box"></div><div class="localvideo-title">fw4基础任务一</div><br/>
<div id="player_4724f024b7b39bf014110a9840f301d52" class="localvideo-box"></div><div class="localvideo-title">fw4基础任务一2</div><br/>

---
参考资料：
[Wiznet/RP2040-HAT-MicroPython (github.com)](https://github.com/Wiznet/RP2040-HAT-MicroPython)
[uPing: 使用 MicroPython 适配 ESP32 的 ping 函数，支持异步。 (gitee.com)](https://gitee.com/Yogile/uPing)

## 基础任务二

主控板建立TCPIP或UDP服务器，局域网PC使用TCPIP或UDP客户端进行连接并发送数据，主控板接收到数据后，送液晶屏显示(没有则通过串口打印显示)；通过抓包软件抓取交互报文，展示并分析。(TCP和UDP二选一，或者全都操作)

### 一、硬件介绍

| 硬件 | 链接、资料 | 图片 |
| ---- | ---- | ---- |
| W5500-EVB-Pico | [W5500-EVB-PICO WIZnet 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824) </br> [W5500-EVB-Pico WIZnet Document System](https://docs.wiznet.io/Product/iEthernet/W5500/w5500-evb-pico) | ![](https://docs.wiznet.io/assets/images/w5500_evb_pico_side-da676c5d9c41adedc0469b9f1810b81b.png) |
| GROVE SHIELD FOR PI PICO V1.0 | [103100142 Seeed Technology Co., Ltd 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103100142/13688265) | ![](https://mm.digikey.com/Volume0/opasdata/d220001/medias/images/3764/MFG_103100142.jpg) |
| Grove - AHT20 I2C Industrial Grade Temperature&Humidity SensorF QT PY/XIAO | [101990644 Seeed Technology Co., Ltd 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/101990644/11681294) </br> [Grove - AHT20 I2C Industrial Grade Temperature&Humidity Sensor Seeed Studio Wiki](https://wiki.seeedstudio.com/Grove-AHT20-I2C-Industrial-Grade-Temperature&Humidity-Sensor/) | ![](https://files.seeedstudio.com/wiki/Grove-AHT20_I2C_Industrial_Grade_Temperature_and_Humidity_Sensor/101990644_4_.png) |

### 二、实现功能
本项目使用microdot来实现webapi服务器，来控制GPIO设备以及发送和接收数据。
1. 使用API来控制板载LED灯的开关；
2. 使用API获得Grove - AHT20设备的温度与湿度信息；
3. 使用API来实现一个聊天室的功能；

### 三、抓包分析
使用Wireshark软件对里聊天过程进行抓包分析，点击发送消息时向服务端发送一个Post请求将文本框中的内容发送给服务端。如下图所示可以看到数据包Info中 一个POST方法 路径为/send-message。通过查看Hypertext Transfer Protocol（应用层的信息）包括请求方法、请求路径、请求的完整路径等信息。在Javascript object Notation:application/json Json数据信息可以看到发送的内容“follow me4”

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786283_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>fw4基础二抓包1.png</strong> <em class="xg1">(427.78 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786283" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786283&handlekey=savephoto_786283">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

下图为客户端接收消息的数据包，可以看到Hypertext Transfer Protocol（应用层的信息）返回的200状态码，以及Json数据中返回的信息“Server received: follow me4”。

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786282_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>fw4基础二抓包2.png</strong> <em class="xg1">(347.19 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786282" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786282&handlekey=savephoto_786282">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

### 三、代码展示

/basic_task2_code.py

``` python
import network,os,time
from usocket import socket
from machine import Pin,SPI,SoftSPI
from microdot import Microdot

app = Microdot()
led = Pin(25, Pin.OUT)
addr = ''

#w5500 初始化
def w5x00_init():
    spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
    nic = network.WIZNET5K(spi,Pin(17),Pin(20)) #spi,cs,reset pin  # network.WIZNET5K(spi,Pin(17),Pin(20))
    nic.active(True)
    
    # If you use the Dynamic IP(DHCP), you must use the "nic.ifconfig('dhcp')".
    nic.ifconfig('dhcp')
    
    # If you use the Static IP, you must use the  "nic.ifconfig("IP","subnet","Gateway","DNS")".
    #nic.ifconfig(('192.168.2.107','255.255.255.0','192.168.2.1','8.8.8.8'))
    
    while not nic.isconnected():
        time.sleep(1)
        #print(nic.regs())
    print(nic.ifconfig())
    
    # Get connection IP
    global addr
    addr = nic.ifconfig()[0]

def get_temp_humidity(debug = False):
    import ahtx0
    from machine import I2C,SPI
    
    i2c = I2C(0, scl=Pin(9), sda=Pin(8))
    # 获取传感器对象
    sensor = ahtx0.AHT20(i2c)
    
    if debug:
        print("\nTemperature: %0.2f C" % sensor.temperature)
        print("Humidity: %0.2f %%" % sensor.relative_humidity)    
    
    return sensor.temperature,sensor.relative_humidity

def app_init():

@app.route('/')
    def index(request):
        return 'Hello, world!'
    
    # 开灯
    @app.route('/ledon', methods=['GET'])
    def ledon(request):
        print("LED ON")
        led.value(1)
        return 'LED ON'
    
    # 关灯
    @app.route('/ledoff', methods=['GET'])
    def ledoff(request):
        print("LED OFF")
        led.value(0)
        return 'LED OFF'
    
    # 返回一次当前温度和湿度
    @app.route('/temp-humidity')
    def temp_humidity(request):
        temperature, humidity = get_temp_humidity()
        return {'temperature': temperature, 'humidity': humidity}
    
    # 通过页面更新持续返回当前温度和湿度
    @app.route('/temp-humidity2')
    def temp_humidity(request):
        from microdot import send_file
        return send_file('/page/Temperature and Humidity.html')
    
    # 聊天室
    @app.route('/chat')
    def send_message(request):
        from microdot import send_file
        return send_file('/page/ChatRoom.html')
    
    @app.route('/send-message', methods=['POST'])
    def send_message(request):
        message = request.json.get('message', '')
        
        # 处理消息，例如，在这里你可以调用其他函数来响应不同的命令或消息
        print(f'received sessage: {message}')
        
        response_message = 'Server received: ' + message
        return {'message': response_message}
    
    # 404页
    @app.errorhandler(404)
    def not_found(request):
        return {'error': 'resource not found'}, 404
    
    app.run(host=addr, port=5000, debug=False, ssl=None)

if __name__ == "__main__":
    w5x00_init()
    app_init()
```

/page/ChatRoom.html

``` html
<!DOCTYPE html>
<html>
<head>
    <title>Chat Room</title>
    <style>
        #chatBox {
            height: 300px;
            overflow: auto;
            background: #f0f0f0;
            padding: 15px;
            display: flex;
            flex-direction: column;
        }
        .message {
            padding: 5px 10px;
            margin: 5px;
            border-radius: 5px;
            max-width: 80%;
        }
        .user-message {
            align-self: flex-end;
            background-color: #dcf8c6;
        }
        .server-message {
            align-self: flex-start;
            background-color: #fff;
        }
    </style>
    <script>
	function sendMessage() {
        var input = document.getElementById('messageInput');
        var message = input.value.trim();
        if(message) {
            displayMessage(message, 'user-message'); // 显示用户消息
            input.value = ''; // 清空输入框
            fetch('/send-message', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json'
                },
                body: JSON.stringify({message: message})
            }).then(function(response) {
                return response.json();
            }).then(function(data) {
                displayMessage(data.message, 'server-message'); // 显示服务端消息
            });
        }
    }
    function displayMessage(text, className) {
	    var chatBox = document.getElementById('chatBox');
	    var messageElem = document.createElement('div');
	    messageElem.textContent = text;
	    messageElem.classList.add('message', className);
	    chatBox.appendChild(messageElem);
	    chatBox.scrollTop = chatBox.scrollHeight; // 滚动到最新消息
    }
    </script>
</head>
<body>
    <h1>Chat Room</h1>
    <p>Follow me 4 by shazhongjin</p>
    <div id="chatBox" style="height: 300px; overflow: auto; background: #f0f0f0; padding: 15px;">
        
    </div>
    <input type="text" id="messageInput" placeholder="Type your message"/>
    <button onclick="sendMessage()">Send</button>
</body>
</html>

```

/page/Temperature and Humidity.html

``` html
<!DOCTYPE html>
<html>
<head>
    <title>Temperature and Humidity</title>
    <script>
    function fetchTempHumidity() {
        fetch('/temp-humidity').then(function(response) {
            return response.json();
        }).then(function(data) {
            document.getElementById('temperature').textContent = data.temperature;
            document.getElementById('humidity').textContent = data.humidity;
        });
    }
    // 每秒执行一次fetchTempHumidity函数
    setInterval(fetchTempHumidity, 1000);
    </script>
</head>
<body>
    <p>Temperature: <span id="temperature">--</span></p>
    <p>Humidity: <span id="humidity">--</span></p>
</body>
</html>
```

### 四、演示

<div id="player_eae6acf9bf12349e9820963e59ee1ab23" class="localvideo-box"></div><div class="localvideo-title">fw4基础任务二(1)</div><br/>

---
参考资料：
[Installation — Microdot documentation](https://microdot.readthedocs.io/en/v1/intro.html)

## 进阶任务
从NTP服务器(注意数据交互格式的解析)同步时间，获取时间送显示屏(串口)显示。

### 一、硬件介绍

| 硬件                          | 链接、资料 | 图片 |
| ----------------------------- | ---- | ---- |
| W5500-EVB-Pico                |  [W5500-EVB-PICO WIZnet 开发板 套 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824) </br> [W5500-EVB-Pico WIZnet Document System](https://docs.wiznet.io/Product/iEthernet/W5500/w5500-evb-pico) | ![](https://docs.wiznet.io/assets/images/w5500_evb_pico_side-da676c5d9c41adedc0469b9f1810b81b.png) |
| Pico RTC DS3231 | [103030398 Seeed Technology Co., Ltd 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103030398/14317036) </br> [Pico RTC DS3231 - Waveshare Wiki](https://www.waveshare.com/wiki/Pico-RTC-DS3231) | ![](https://www.waveshare.com/media/catalog/product/cache/1/image/800x800/9df78eab33525d08d6e5fb8d27136e95/p/i/pico-rtc-ds3231-1_4.jpg) |
| Pico LCD 1.14   | [103030400 Seeed Technology Co., Ltd 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103030400/14317043) </br> [Pico LCD 1.14 - Waveshare Wiki](https://www.waveshare.com/wiki/Pico-LCD-1.14) | ![](https://www.waveshare.com/media/catalog/product/cache/1/image/800x800/9df78eab33525d08d6e5fb8d27136e95/p/i/pico-lcd-1.14-2_4.jpg) |

接线图1

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786286" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786286&handlekey=savephoto_786286">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

接线图2

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786284" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786284&handlekey=savephoto_786284">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

**Pico RTC DS3231**
 Pico-RTC-DS3231 是专为 Raspberry Pi Pico 设计的 RTC 扩展模块。它集成了高精度RTC芯片DS3231，并使用I2C总线进行通信。由于采用可堆叠设计，可以连接更多的外部传感器。
 
 - 标准树莓派Pico接头，支持树莓派Pico系列。
- 板载高精度RTC芯片DS3231，带备用电池座。
    - 实时时钟可计算秒、分、小时、月、月、星期几和年份，闰年补偿有效期至 2100 年。
    - 可选格式：24 小时或 12 小时，带 AM/PM 指示器。
    - 2 x 可编程闹钟。
- 提供在线文档（Raspberry Pi、Pico、C/C++ 和 MicroPython 示例演示）。

![](https://www.waveshare.com/w/upload/4/40/Pico-RTC-DS3231-details-inter.jpg)

至于为什么会使用DS3231呢？但是主要想搭配这屏幕一起叠高高。一看引脚排列就傻眼了 使用引脚GPIO20、21。GPIO16-21都被w5500使用，所以没办法只能杜邦线来连接DS3231以及屏幕了。

### 二、驱动RTC ds3231

这里使用了微雪示例代码 ds3231.py ，对ds3231设置时间。

``` python
#!/usr/bin/python
# -*- coding: utf-8 -*-
from machine import Pin, I2C
import time
import binascii

#    the first version use i2c1
#I2C_PORT = 1
#I2C_SDA = 6
#I2C_SCL = 7

#    the new version use i2c0,if it dont work,try to uncomment the line 14 and comment line 17
#    it should solder the R3 with 0R resistor if want to use alarm function,please refer to the Sch file on waveshare Pico-RTC-DS3231 wiki
#    https://www.waveshare.net/w/upload/0/08/Pico-RTC-DS3231_Sch.pdf
I2C_PORT = 0
I2C_SDA = 20
I2C_SCL = 21
ALARM_PIN = 3

class ds3231(object):
#            13:45:00 Mon 24 May 2021
#  the register value is the binary-coded decimal (BCD) format
#               sec min hour week day month year
    NowTime = b'\x00\x45\x13\x02\x24\x05\x21'
    w  = ["Sunday","Monday","Tuesday","Wednesday","Thursday","Friday","Saturday"];
    address = 0x68
    start_reg = 0x00
    alarm1_reg = 0x07
    control_reg = 0x0e
    status_reg = 0x0f

def __init__(self,i2c_port,i2c_scl,i2c_sda):
        self.bus = I2C(i2c_port,scl=Pin(i2c_scl),sda=Pin(i2c_sda))

def set_time(self,new_time):
        hour = new_time[0] + new_time[1]
        minute = new_time[3] + new_time[4]
        second = new_time[6] + new_time[7]
        week = "0" + str(self.w.index(new_time.split(",",2)[1])+1)
        year = new_time.split(",",2)[2][2] + new_time.split(",",2)[2][3]
        month = new_time.split(",",2)[2][5] + new_time.split(",",2)[2][6]
        day = new_time.split(",",2)[2][8] + new_time.split(",",2)[2][9]
        now_time = binascii.unhexlify((second + " " + minute + " " + hour + " " + week + " " + day + " " + month + " " + year).replace(' ',''))
        #print(binascii.unhexlify((second + " " + minute + " " + hour + " " + week + " " + day + " " + month + " " + year).replace(' ','')))
        #print(self.NowTime)
        self.bus.writeto_mem(int(self.address),int(self.start_reg),now_time)

def read_time(self):
        t = self.bus.readfrom_mem(int(self.address),int(self.start_reg),7)
        a = t[0]&0x7F  #second
        b = t[1]&0x7F  #minute
        c = t[2]&0x3F  #hour
        d = t[3]&0x07  #week
        e = t[4]&0x3F  #day
        f = t[5]&0x1F  #month
        print("20%x/%02x/%02x %02x:%02x:%02x %s" %(t[6],t[5],t[4],t[2],t[1],t[0],self.w[t[3]-1]))
        
    def set_alarm_time(self,alarm_time):

#    init the alarm pin
        self.alarm_pin = Pin(ALARM_PIN,Pin.IN,Pin.PULL_UP)

#    set alarm irq
        self.alarm_pin.irq(lambda pin: print("alarm1 time is up"), Pin.IRQ_FALLING)

#    enable the alarm1 reg
        self.bus.writeto_mem(int(self.address),int(self.control_reg),b'\x05')

#    convert to the BCD format
        hour = alarm_time[0] + alarm_time[1]
        minute = alarm_time[3] + alarm_time[4]
        second = alarm_time[6] + alarm_time[7]
        date = alarm_time.split(",",2)[2][8] + alarm_time.split(",",2)[2][9]
        now_time = binascii.unhexlify((second + " " + minute + " " + hour +  " " + date).replace(' ',''))

#    write alarm time to alarm1 reg
        self.bus.writeto_mem(int(self.address),int(self.alarm1_reg),now_time)

if __name__ == '__main__':
    rtc = ds3231(I2C_PORT,I2C_SCL,I2C_SDA)
    rtc.set_time('13:45:50,Monday,2021-05-24')
    rtc.read_time()
    rtc.set_alarm_time('13:45:55,Monday,2021-05-24')
```

### 三、代码展示

advanced_task_code.py

``` python
import network,time,LCD_1inch14,ds3231
from machine import Pin,SPI

addr = ''

#w5500 初始化
def w5x00_init():
    spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
    nic = network.WIZNET5K() #spi,cs,reset pin  # network.WIZNET5K(spi,Pin(17),Pin(20))
    nic.active(True)
    
    # If you use the Dynamic IP(DHCP), you must use the "nic.ifconfig('dhcp')".
    nic.ifconfig('dhcp')
    
    # If you use the Static IP, you must use the  "nic.ifconfig("IP","subnet","Gateway","DNS")".
    #nic.ifconfig(('192.168.1.108','255.255.255.0','192.168.1.1','8.8.8.8'))
    
    while not nic.isconnected():
        time.sleep(1)
        #print(nic.regs())
    print(nic.ifconfig())
    
    # Get connection IP
    global addr
    addr = nic.ifconfig()[0]

def sync_ntp():
    import ntptime,machine,time
    
    I2C_PORT = 1
    I2C_SDA = 26
    I2C_SCL = 27
    weekday  = ["Sunday","Monday","Tuesday","Wednesday","Thursday","Friday","Saturday"];
    
    print("start sync ntp time")
    
    # 使用DS3231RTC
    rtc = ds3231.ds3231(I2C_PORT,I2C_SCL,I2C_SDA)
    
    try:
        ntptime.host = 'ntp1.aliyun.com'  # 可选，ntp服务器，默认是"pool.ntp.org" 这里使用阿里服务器
        ntptime.settime()  # 修改设备时间,到这就已经设置好了
        
        sysrtc = machine.RTC()
        
        localtime_now=time.time()+8*3600
        localtime_now=time.localtime(localtime_now)
        sysrtc.datetime((localtime_now[0],localtime_now[1],localtime_now[2],localtime_now[6],localtime_now[3],localtime_now[4],localtime_now[5],localtime_now[7]))

year, month, day, weekday, hours, minutes, seconds = sysrtc.datetime()[0:7]
        # 构造24小时制格式的时间字符串，确保小时、分钟和秒都是两位数
        time_str = "{:02d}:{:02d}:{:02d}".format(hours, minutes, seconds)
        # 构造日期字符串
        date_str = "{:04d}-{:02d}-{:02d}".format(year, month, day)
        # 星期的字典
        weekdays = {
            0: "Monday",
            1: "Tuesday",
            2: "Wednesday",
            3: "Thursday",
            4: "Friday",
            5: "Saturday",
            6: "Sunday"
        }
        # 获取星期几
        weekday_str = weekdays.get(weekday, "")
   
        # 设置DS3231RTC时间
        rtc.set_time(time_str + "," + weekday_str + "," + date_str)       
                
        print("success sync ntp time")
    except Exception as e:
        print("error sync ntp time",repr(e))
    
if __name__ == "__main__":
    # 初始化网络
    w5x00_init()

# 同步ntp时间
    sync_ntp()
    
    # 初始化LCD
    LCD = LCD_1inch14.LCD()
    
    rtc = ds3231.ds3231(1,27,26)
    
    while True:
        print(f'The SystemRTC time is :         {time.localtime()[0]}-{time.localtime()[1]}-{time.localtime()[2]} {time.localtime()[3]}:{time.localtime()[4]}:{time.localtime()[5]}          The DS3231RTC time is :         {rtc.read_time()[0]}-{rtc.read_time()[1]}-{rtc.read_time()[2]} {rtc.read_time()[3]}:{rtc.read_time()[4]}:{rtc.read_time()[5]} {rtc.read_time()[6]}')
        LCD.WriteLine(f'The SystemRTC time is :         {time.localtime()[0]}-{time.localtime()[1]}-{time.localtime()[2]} {time.localtime()[3]}:{time.localtime()[4]}:{time.localtime()[5]}          The DS3231RTC time is :         {rtc.read_time()[0]}-{rtc.read_time()[1]}-{rtc.read_time()[2]} {rtc.read_time()[3]}:{rtc.read_time()[4]}:{rtc.read_time()[5]} {rtc.read_time()[6]}')
        LCD.Clean()
```

/lib/ds3231.py 来自于参考资料连接中微雪例程中文件未做改动，也可本项目提供的源码。

### 四、演示

---
参考资料：
[Pico RTC DS3231 - Waveshare Wiki](https://www.waveshare.com/wiki/Pico-RTC-DS3231)

## 终极任务二
■  **终极任务二：**使用外部存储器，组建简易FTP文件服务器，并能正常上传下载文件。

### 一、硬件介绍

本项目中使用到的硬件有：

| 硬件                          | 链接、资料 | 图片 |
| ----------------------------- | ---- | ---- |
| W5500-EVB-Pico                |  [W5500-EVB-PICO WIZnet 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/wiznet/W5500-EVB-PICO/16515824) </br> [W5500-EVB-Pico WIZnet Document System](https://docs.wiznet.io/Product/iEthernet/W5500/w5500-evb-pico) | ![](https://docs.wiznet.io/assets/images/w5500_evb_pico_side-da676c5d9c41adedc0469b9f1810b81b.png) |
| GROVE SHIELD FOR PI PICO V1.0 | [103100142 Seeed Technology Co., Ltd 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/seeed-technology-co-ltd/103100142/13688265) | ![](https://mm.digikey.com/Volume0/opasdata/d220001/medias/images/3764/MFG_103100142.jpg) |
| MICROSD CARD BFF QT PY/XIAO   | [5683 Adafruit Industries LLC 开发板 套件 编程器 DigiKey](https://www.digikey.cn/zh/products/detail/adafruit-industries-llc/5683/18073185) </br> [Overview Adafruit microSD Card BFF Adafruit Learning System](https://learn.adafruit.com/adafruit-microsd-card-bff) | ![](https://cdn-learn.adafruit.com/guides/cropped_images/000/003/748/medium640/5683-00.jpg?1676318485) |

接线图1

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786288_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>w5500 sdcard 接线图.png</strong> <em class="xg1">(4.71 MB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786288" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786288&handlekey=savephoto_786288">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

接线图2

<ignore_js_op>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786287" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786287&handlekey=savephoto_786287">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

### 二、设计思路
1. 首先我们需要使用W5500-EVB-Pico连接到网络中。
2. 将sd卡挂载到根目录。

#### 读取SDCard
1. 点击包管理工具，安装sdcard库如下图所示。

<ignore_js_op>

<div class="tip tip_4 aimg_tip" id="aimg_786290_menu" style="position: absolute; display: none" disautofocus="true">
<div class="xs0">
<p><strong>Pasted image 20240118135434.png</strong> <em class="xg1">(29.31 KB, 下载次数: 0)</em></p>
<p>
<a href="javascript:;" class="loginf" >下载附件</a>

<a href="javascript:;" onclick="showWindow(this.id, this.getAttribute('url'), 'get', 0);" id="savephoto_786290" url="home.php?mod=spacecp&ac=album&op=saveforumphoto&aid=786290&handlekey=savephoto_786290">保存到相册</a>

</p>

</div>
<div class="tip_horn"></div>
</div>

</ignore_js_op>

2. 使用代码来挂载sd卡。

``` python
    import os
    from sdcard import SDCard
    from machine import Pin,SPI,SoftSPI
    
    spi = SPI(1,2_000_000,sck=Pin(10), mosi=Pin(11), miso=Pin(12))
    #spi = SoftSPI(2_000_000,sck=Pin(10), mosi=Pin(11), miso=Pin(12))
    
    # 获取sdcard对象
    sd = SDCard(spi, Pin(13))
    # 挂载SD卡到/sd目录
    os.mount(sd, '/sd')
```

#### 实现FTP文件服务器
造轮子是不可能的这辈子都不可能，既然不造轮子在github上查找下有没有MicroPython FTP库，还真有来自[robert-hh](https://github.com/robert-hh)大神的[FTP-Server-for-ESP8266-ESP32-and-PYBD](https://github.com/robert-hh/FTP-Server-for-ESP8266-ESP32-and-PYBD/tree/master)库。使用库中的uftpd.py文件，该函数使用network.WLAN来获得连接信息。但本项目使用的是w5500控制器来实现网络连接，还需要对start函数稍加小改动。

``` python
# start listening for ftp connections on port 21
def start(port=21, verbose=0, splash=True):
    global ftpsockets, datasocket
    global verbose_l
    global client_list
    global client_busy

alloc_emergency_exception_buf(100)
    verbose_l = verbose
    client_list = []
    client_busy = False

for interface in [network.AP_IF, network.STA_IF]:
        wlan = network.WLAN(interface)
        if not wlan.active():
            continue

ifconfig = wlan.ifconfig()
        addr = socket.getaddrinfo(ifconfig[0], port)
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(addr[0][4])
        sock.listen(1)
        sock.setsockopt(socket.SOL_SOCKET,
                        _SO_REGISTER_HANDLER,
                        lambda s : accept_ftp_connect(s, ifconfig[0]))
        ftpsockets.append(sock)
        if splash:
            print("FTP server started on {}:{}".format(ifconfig[0], port))

datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    datasocket.bind(('0.0.0.0', _DATA_PORT))
    datasocket.listen(1)
    datasocket.settimeout(10)

def restart(port=21, verbose=0, splash=True):
    stop()
    sleep_ms(200)
    start(port, verbose, splash)
```

改动思路：
1. 在uftpd.py 的 start函数中，将WLAN改为WIZNET5K，从WIZNET5K对象中获得网络连接信息。

``` python
# start listening for ftp connections on port 21
def start(port=21, verbose=0, splash=True):
    global ftpsockets, datasocket
    global verbose_l
    global client_list
    global client_busy

alloc_emergency_exception_buf(100)
    verbose_l = verbose
    client_list = []
    client_busy = False

nic = network.WIZNET5K()
	if not nic.active():
		continue

ifconfig = nic.ifconfig()
	addr = socket.getaddrinfo(ifconfig[0], port)
	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
	sock.bind(addr[0][4])
	sock.listen(1)
	sock.setsockopt(socket.SOL_SOCKET,
					_SO_REGISTER_HANDLER,
					lambda s : accept_ftp_connect(s, ifconfig[0]))
	ftpsockets.append(sock)
	if splash:
		print("FTP server started on {}:{}".format(ifconfig[0], port))

datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    datasocket.bind(('0.0.0.0', _DATA_PORT))
    datasocket.listen(1)
    datasocket.settimeout(10)

def restart(port=21, verbose=0, splash=True):
    stop()
    sleep_ms(200)
    start(port, verbose, splash)
```

修改后在执行过程中一直无法成功创建FTP服务，经过分析找到了问题的原因。当在uftpd.py文件中再此执行 `network.WIZNET5K()` 时似乎他创建了一个新的nic对象，而不是使用 `network.WLAN()` 时返回的是当前的网络连接对象。所以在判断 `not nic.active()` 结果永远为True。程序无法继续运行下去。

2. 不妨换个思路在uftpd.start函数中获得网络对象的目的为获得当前网络连接分配的IP地址。干脆让start函数传入ip参数。这样程序就能顺利执行。也不会卡在`network.WIZNET5K()`这个问题上了。

``` python
# start listening for ftp connections on port 21
def start(host='0.0.0.0',port=21, verbose=0, splash=True):
    global ftpsockets, datasocket
    global verbose_l
    global client_list
    global client_busy

alloc_emergency_exception_buf(100)
    verbose_l = verbose
    client_list = []
    client_busy = False

addr = socket.getaddrinfo(host, port)
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(addr[0][4])
    sock.listen(1)
    sock.setsockopt(socket.SOL_SOCKET,
                    _SO_REGISTER_HANDLER,
                    lambda s : accept_ftp_connect(s,host))
    ftpsockets.append(sock)
    if splash:
        print("FTP server started on {}:{}".format(host, port))

datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    datasocket.bind(('0.0.0.0', _DATA_PORT))
    datasocket.listen(1)
    datasocket.settimeout(10)

def restart(host='0.0.0.0',port=21, verbose=0, splash=True):
    stop()
    sleep_ms(200)
    start(port, verbose, splash)
```

### 三、代码展示

/ultimate_task2_code.py

``` python
import network,time
from machine import Pin,SPI

addr = ''

#w5500 初始化
def w5x00_init():
    spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
    nic = network.WIZNET5K(spi,Pin(17),Pin(20)) #spi,cs,reset pin  # network.WIZNET5K(spi,Pin(17),Pin(20))
    nic.active(True)
    
    # If you use the Dynamic IP(DHCP), you must use the "nic.ifconfig('dhcp')".
    nic.ifconfig('dhcp')
    
    # If you use the Static IP, you must use the  "nic.ifconfig("IP","subnet","Gateway","DNS")".
    #nic.ifconfig(('192.168.2.107','255.255.255.0','192.168.2.1','8.8.8.8'))
    
    while not nic.isconnected():
        time.sleep(1)
        #print(nic.regs())
    print(nic.ifconfig())
    
    # Get connection IP
    global addr
    addr = nic.ifconfig()[0]

def sdcard_init(path='/sd'):
    import os
    from sdcard import SDCard
    from machine import Pin,SPI,SoftSPI
    
    spi = SPI(1,2_000_000,sck=Pin(10), mosi=Pin(11), miso=Pin(12))
    #spi = SoftSPI(2_000_000,sck=Pin(10), mosi=Pin(11), miso=Pin(12))
    
    # 获取sdcard对象
    sd = SDCard(spi, Pin(13))
    # 挂载SD卡到/sd目录
    os.mount(sd, path)
    print(f'success mounted sdcard in \'{path}\'')

if __name__ == "__main__":
    # 初始化网络
    w5x00_init()
    
    # 初始化并挂在sd卡
    sdcard_init()
    
    import uftpd
    uftpd.start(host = addr,port = 21,verbose = 0)
```

/lib/uftp.py

``` python
#
# Small ftp server for ESP8266 Micropython
# Based on the work of chrisgp - Christopher Popp and pfalcon - Paul Sokolovsky
#
# The server accepts passive mode only. It runs in background.
# Start the server with:
#
# import uftpd
# uftpd.start([port = 21][, verbose = level])
#
# port is the port number (default 21)
# verbose controls the level of printed activity messages, values 0, 1, 2
#
# Copyright (c) 2016 Christopher Popp (initial ftp server framework)
# Copyright (c) 2016 Paul Sokolovsky (background execution control structure)
# Copyright (c) 2016 Robert Hammelrath (putting the pieces together and a
# few extensions)
# Copyright (c) 2020 Jan Wieck Use separate FTP servers per socket for STA + AP mode
# Copyright (c) 2021 JD Smith Use a preallocated buffer and improve error handling.
# Distributed under MIT License
#
import socket
import network
import uos
import gc
import sys
import errno
from time import sleep_ms, localtime
from micropython import alloc_emergency_exception_buf

# constant definitions
_CHUNK_SIZE = const(1024)
_SO_REGISTER_HANDLER = const(20)
_COMMAND_TIMEOUT = const(300)
_DATA_TIMEOUT = const(100)
_DATA_PORT = const(13333)

# Global variables
ftpsockets = []
datasocket = None
client_list = []
verbose_l = 0
client_busy = False
# Interfaces: (IP-Address (string), IP-Address (integer), Netmask (integer))

_month_name = ("", "Jan", "Feb", "Mar", "Apr", "May", "Jun",
               "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")

class FTP_client:

def __init__(self, ftpsocket, local_addr):
        self.command_client, self.remote_addr = ftpsocket.accept()
        self.remote_addr = self.remote_addr[0]
        self.command_client.settimeout(_COMMAND_TIMEOUT)
        log_msg(1, "FTP Command connection from:", self.remote_addr)
        self.command_client.setsockopt(socket.SOL_SOCKET,
                                       _SO_REGISTER_HANDLER,
                                       self.exec_ftp_command)
        self.command_client.sendall("220 Hello, this is the {}.\r\n".format(sys.platform))
        self.cwd = '/'
        self.fromname = None
#        self.logged_in = False
        self.act_data_addr = self.remote_addr
        self.DATA_PORT = 20
        self.active = True
        self.pasv_data_addr = local_addr

def send_list_data(self, path, data_client, full):
        try:
            for fname in uos.listdir(path):
                data_client.sendall(self.make_description(path, fname, full))
        except Exception as e:  # path may be a file name or pattern
            path, pattern = self.split_path(path)
            try:
                for fname in uos.listdir(path):
                    if self.fncmp(fname, pattern):
                        data_client.sendall(
                            self.make_description(path, fname, full))
            except:
                pass

def make_description(self, path, fname, full):
        global _month_name
        if full:
            stat = uos.stat(self.get_absolute_path(path, fname))
            file_permissions = ("drwxr-xr-x"
                                if (stat[0] & 0o170000 == 0o040000)
                                else "-rw-r--r--")
            file_size = stat[6]
            tm = stat[7] & 0xffffffff
            tm = localtime(tm if tm < 0x80000000 else tm - 0x100000000)
            if tm[0] != localtime()[0]:
                description = "{} 1 owner group {:>10} {} {:2} {:>5} {}\r\n".\
                    format(file_permissions, file_size,
                        _month_name[tm[1]], tm[2], tm[0], fname)
            else:
                description = "{} 1 owner group {:>10} {} {:2} {:02}:{:02} {}\r\n".\
                    format(file_permissions, file_size,
                        _month_name[tm[1]], tm[2], tm[3], tm[4], fname)
        else:
            description = fname + "\r\n"
        return description

def send_file_data(self, path, data_client):
        buffer = bytearray(_CHUNK_SIZE)
        mv = memoryview(buffer)
        with open(path, "rb") as file:
            bytes_read = file.readinto(buffer)
            while bytes_read > 0:
                data_client.write(mv[0:bytes_read])
                bytes_read = file.readinto(buffer)
            data_client.close()

def save_file_data(self, path, data_client, mode):
        buffer = bytearray(_CHUNK_SIZE)
        mv = memoryview(buffer)
        with open(path, mode) as file:
            bytes_read = data_client.readinto(buffer)
            while bytes_read > 0:
                file.write(mv[0:bytes_read])
                bytes_read = data_client.readinto(buffer)
            data_client.close()

def get_absolute_path(self, cwd, payload):
        # Just a few special cases "..", "." and ""
        # If payload start's with /, set cwd to /
        # and consider the remainder a relative path
        if payload.startswith('/'):
            cwd = "/"
        for token in payload.split("/"):
            if token == '..':
                cwd = self.split_path(cwd)[0]
            elif token != '.' and token != '':
                if cwd == '/':
                    cwd += token
                else:
                    cwd = cwd + '/' + token
        return cwd

def split_path(self, path):  # instead of path.rpartition('/')
        tail = path.split('/')[-1]
        head = path[:-(len(tail) + 1)]
        return ('/' if head == '' else head, tail)

# compare fname against pattern. Pattern may contain
    # the wildcards ? and *.
    def fncmp(self, fname, pattern):
        pi = 0
        si = 0
        while pi < len(pattern) and si < len(fname):
            if (fname[si] == pattern[pi]) or (pattern[pi] == '?'):
                si += 1
                pi += 1
            else:
                if pattern[pi] == '*':  # recurse
                    if pi == len(pattern.rstrip("*?")):  # only wildcards left
                        return True
                    while si < len(fname):
                        if self.fncmp(fname[si:], pattern[pi + 1:]):
                            return True
                        else:
                            si += 1
                    return False
                else:
                    return False
        if pi == len(pattern.rstrip("*")) and si == len(fname):
            return True
        else:
            return False

def open_dataclient(self):
        if self.active:  # active mode
            data_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            data_client.settimeout(_DATA_TIMEOUT)
            data_client.connect((self.act_data_addr, self.DATA_PORT))
            log_msg(1, "FTP Data connection with:", self.act_data_addr)
        else:  # passive mode
            data_client, data_addr = datasocket.accept()
            log_msg(1, "FTP Data connection with:", data_addr[0])
        return data_client

def exec_ftp_command(self, cl):
        global datasocket
        global client_busy
        global my_ip_addr

try:
            gc.collect()

try:
                data = cl.readline().decode("utf-8").rstrip("\r\n")
            except OSError:
                # treat an error as QUIT situation.
                data = ""

if len(data) <= 0:
                # No data, close
                # This part is NOT CLEAN; there is still a chance that a
                # closing data connection will be signalled as closing
                # command connection
                log_msg(1, "*** No data, assume QUIT")
                close_client(cl)
                return

if client_busy:  # check if another client is busy
                cl.sendall("400 Device busy.\r\n")  # tell so the remote client
                return  # and quit
            client_busy = True  # now it's my turn

# check for log-in state may done here, like
            # if self.logged_in == False and not command in\
            #    ("USER", "PASS", "QUIT"):
            #    cl.sendall("530 Not logged in.\r\n")
            #    return

command = data.split()[0].upper()
            payload = data[len(command):].lstrip()  # partition is missing
            path = self.get_absolute_path(self.cwd, payload)
            log_msg(1, "Command={}, Payload={}".format(command, payload))

if command == "USER":
                # self.logged_in = True
                cl.sendall("230 Logged in.\r\n")
                # If you want to see a password,return
                #   "331 Need password.\r\n" instead
                # If you want to reject an user, return
                #   "530 Not logged in.\r\n"
            elif command == "PASS":
                # you may check here for a valid password and return
                # "530 Not logged in.\r\n" in case it's wrong
                # self.logged_in = True
                cl.sendall("230 Logged in.\r\n")
            elif command == "SYST":
                cl.sendall("215 UNIX Type: L8\r\n")
            elif command in ("TYPE", "NOOP", "ABOR"):  # just accept & ignore
                cl.sendall('200 OK\r\n')
            elif command == "QUIT":
                cl.sendall('221 Bye.\r\n')
                close_client(cl)
            elif command == "PWD" or command == "XPWD":
                cl.sendall('257 "{}"\r\n'.format(self.cwd))
            elif command == "CWD" or command == "XCWD":
                try:
                    if (uos.stat(path)[0] & 0o170000) == 0o040000:
                        self.cwd = path
                        cl.sendall('250 OK\r\n')
                    else:
                        cl.sendall('550 Fail\r\n')
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "PASV":
                cl.sendall('227 Entering Passive Mode ({},{},{}).\r\n'.format(
                    self.pasv_data_addr.replace('.', ','),
                    _DATA_PORT >> 8, _DATA_PORT % 256))
                self.active = False
            elif command == "PORT":
                items = payload.split(",")
                if len(items) >= 6:
                    self.act_data_addr = '.'.join(items[:4])
                    if self.act_data_addr == "127.0.1.1":
                        # replace by command session addr
                        self.act_data_addr = self.remote_addr
                    self.DATA_PORT = int(items[4]) * 256 + int(items[5])
                    cl.sendall('200 OK\r\n')
                    self.active = True
                else:
                    cl.sendall('504 Fail\r\n')
            elif command == "LIST" or command == "NLST":
                if payload.startswith("-"):
                    option = payload.split()[0].lower()
                    path = self.get_absolute_path(
                            self.cwd, payload[len(option):].lstrip())
                else:
                    option = ""
                try:
                    data_client = self.open_dataclient()
                    cl.sendall("150 Directory listing:\r\n")
                    self.send_list_data(path, data_client,
                                        command == "LIST" or 'l' in option)
                    cl.sendall("226 Done.\r\n")
                    data_client.close()
                except:
                    cl.sendall('550 Fail\r\n')
                    if data_client is not None:
                        data_client.close()
            elif command == "RETR":
                try:
                    data_client = self.open_dataclient()
                    cl.sendall("150 Opened data connection.\r\n")
                    self.send_file_data(path, data_client)
                    # if the next statement is reached,
                    # the data_client was closed.
                    data_client = None
                    cl.sendall("226 Done.\r\n")
                except:
                    cl.sendall('550 Fail\r\n')
                    if data_client is not None:
                        data_client.close()
            elif command == "STOR" or command == "APPE":
                try:
                    data_client = self.open_dataclient()
                    cl.sendall("150 Opened data connection.\r\n")
                    self.save_file_data(path, data_client,
                                        "wb" if command == "STOR" else "ab")
                    # if the next statement is reached,
                    # the data_client was closed.
                    data_client = None
                    cl.sendall("226 Done.\r\n")
                except:
                    cl.sendall('550 Fail\r\n')
                    if data_client is not None:
                        data_client.close()
            elif command == "SIZE":
                try:
                    cl.sendall('213 {}\r\n'.format(uos.stat(path)[6]))
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "MDTM":
                try:
                    tm=localtime(uos.stat(path)[8])
                    cl.sendall('213 {:04d}{:02d}{:02d}{:02d}{:02d}{:02d}\r\n'.format(*tm[0:6]))
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "STAT":
                if payload == "":
                    cl.sendall("211-Connected to ({})\r\n"
                               "    Data address ({})\r\n"
                               "    TYPE: Binary STRU: File MODE: Stream\r\n"
                               "    Session timeout {}\r\n"
                               "211 Client count is {}\r\n".format(
                                self.remote_addr, self.pasv_data_addr,
                                _COMMAND_TIMEOUT, len(client_list)))
                else:
                    cl.sendall("213-Directory listing:\r\n")
                    self.send_list_data(path, cl, True)
                    cl.sendall("213 Done.\r\n")
            elif command == "DELE":
                try:
                    uos.remove(path)
                    cl.sendall('250 OK\r\n')
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "RNFR":
                try:
                    # just test if the name exists, exception if not
                    uos.stat(path)
                    self.fromname = path
                    cl.sendall("350 Rename from\r\n")
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "RNTO":
                    try:
                        uos.rename(self.fromname, path)
                        cl.sendall('250 OK\r\n')
                    except:
                        cl.sendall('550 Fail\r\n')
                    self.fromname = None
            elif command == "CDUP" or command == "XCUP":
                self.cwd = self.get_absolute_path(self.cwd, "..")
                cl.sendall('250 OK\r\n')
            elif command == "RMD" or command == "XRMD":
                try:
                    uos.rmdir(path)
                    cl.sendall('250 OK\r\n')
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "MKD" or command == "XMKD":
                try:
                    uos.mkdir(path)
                    cl.sendall('250 OK\r\n')
                except:
                    cl.sendall('550 Fail\r\n')
            elif command == "SITE":
                try:
                    exec(payload.replace('\0','\n'))
                    cl.sendall('250 OK\r\n')
                except:
                    cl.sendall('550 Fail\r\n')
            else:
                cl.sendall("502 Unsupported command.\r\n")
                # log_msg(2,
                #  "Unsupported command {} with payload {}".format(command,
                #  payload))
        except OSError as err:
            if verbose_l > 0:
                log_msg(1, "Exception in exec_ftp_command:")
                sys.print_exception(err)
            if err.errno in (errno.ECONNABORTED, errno.ENOTCONN):
                close_client(cl)
        # handle unexpected errors
        except Exception as err:
            log_msg(1, "Exception in exec_ftp_command: {}".format(err))
        # tidy up before leaving
        client_busy = False

def log_msg(level, *args):
    global verbose_l
    if verbose_l >= level:
        print(*args)

# close client and remove it from the list
def close_client(cl):
    cl.setsockopt(socket.SOL_SOCKET, _SO_REGISTER_HANDLER, None)
    cl.close()
    for i, client in enumerate(client_list):
        if client.command_client == cl:
            del client_list<i>
            break

def accept_ftp_connect(ftpsocket, local_addr):
    # Accept new calls for the server
    try:
        client_list.append(FTP_client(ftpsocket, local_addr))
    except:
        log_msg(1, "Attempt to connect failed")
        # try at least to reject
        try:
            temp_client, temp_addr = ftpsocket.accept()
            temp_client.close()
        except:
            pass

def num_ip(ip):
    items = ip.split(".")
    return (int(items[0]) << 24 | int(items[1]) << 16 |
            int(items[2]) << 8 | int(items[3]))

def stop():
    global ftpsockets, datasocket
    global client_list
    global client_busy

for client in client_list:
        client.command_client.setsockopt(socket.SOL_SOCKET,
                                         _SO_REGISTER_HANDLER, None)
        client.command_client.close()
    del client_list
    client_list = []
    client_busy = False
    for sock in ftpsockets:
        sock.setsockopt(socket.SOL_SOCKET, _SO_REGISTER_HANDLER, None)
        sock.close()
    ftpsockets = []
    if datasocket is not None:
        datasocket.close()
        datasocket = None

# start listening for ftp connections on port 21
def start(host='0.0.0.0',port=21, verbose=0, splash=True):
    global ftpsockets, datasocket
    global verbose_l
    global client_list
    global client_busy

alloc_emergency_exception_buf(100)
    verbose_l = verbose
    client_list = []
    client_busy = False

addr = socket.getaddrinfo(host, port)
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(addr[0][4])
    sock.listen(1)
    sock.setsockopt(socket.SOL_SOCKET,
                    _SO_REGISTER_HANDLER,
                    lambda s : accept_ftp_connect(s,host))
    ftpsockets.append(sock)
    if splash:
        print("FTP server started on {}:{}".format(host, port))

datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    datasocket.bind(('0.0.0.0', _DATA_PORT))
    datasocket.listen(1)
    datasocket.settimeout(10)

def restart(host='0.0.0.0',port=21, verbose=0, splash=True):
    stop()
    sleep_ms(200)
    start(port, verbose, splash)

#start(splash=True)
```

### 四、演示

<div id="player_379e2ac7056f7dd4d7f35efa6543f5ee5" class="localvideo-box"></div><div class="localvideo-title">fw4终极任务二</div><br/>

---
参考资料：
[robert-hh/FTP-Server-for-ESP8266-ESP32-and-PYBD：MicroPython 平台上用于 ESP8266/ESP32/PYBD 的小型 FTP 服务器 (github.com)](https://github.com/robert-hh/FTP-Server-for-ESP8266-ESP32-and-PYBD/tree/master)

# 总结
本次项目中深深了解了自己小白水平，比较深刻印象是在终极任务二的最后读写sd卡，每次通过FTP写入sd卡的时候都会返回个异常导致写入不成功。但使用Thonny传入文件时无任何问题，以及在Follow me第三期时也使用操作了microsd卡设备均没有存在问题。到最后经过不断的测试发现了问题所在自己没有理解硬件SPI与软SPI的差异以及如何选用各种GPIO接口，错误的使用了SPI（0）这个已经被w5500使用导致连接sd卡写入报错。尽管这感觉是十分的基础的内容，也通过一次次尝试中找到了方法和思路。 感谢得捷、感谢EEWorld，希望能多多参与学习活动，也希望Follow me活动越来越好。

Jacktang

用杜邦线来连接DS3231和屏幕也是一样的正常显示，，，

shazhongjin

Jacktang 发表于 2024-2-18 07:33 用杜邦线来连接DS3231和屏幕也是一样的正常显示，，，

嗯我索性就全部用杜邦线连接了，当时咋没想到DS3231可以跟屏幕插在一块

Henry-0755

写得好棒，排版也很整齐，学习了

chejm

感谢楼主的无私分享，楼主辛苦啦，楼主总结的非常到位，值得收藏学习

【得捷电子Follow me第4期】成果展示汇总帖 [复制链接]

最新回复

点评

浏览过的版块