459|1

21

帖子

2

TA的资源

一粒金砂(中级)

楼主
 

【得捷电子Follow me第4期】终极任务 [复制链接]

  本帖最后由 eew_uscYT9 于 2024-2-21 19:42 编辑

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

通过网上开源代码,建立了ftp服务器,代码如下:
#
# Small ftp server for ESP8266 ans ESP32 Micropython
#
# Based on the work of chrisgp - Christopher Popp and pfalcon - Paul Sokolovsky
#
# The server accepts passive mode only.
# It runs in foreground and quits, when it receives a quit command
# Start the server with:
#
# import ftp
#
# Copyright (c) 2016 Christopher Popp (initial ftp server framework)
# Copyright (c) 2016 Robert Hammelrath (putting the pieces together
# and a few extensions)
# Distributed under MIT License
#
import socket
import network
import uos
import gc
#from usocket import socket
from machine import Pin,SPI,UART,PWM
import time, network,framebuf

''' static netinfo
'''
ip = '192.168.1.20'
sn = '255.255.255.0'
gw = '192.168.1.1'
dns= '8.8.8.8'

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

netinfo=(ip, sn, gw, dns)

localip = ''
localport = 8000
listen_info = (localip, localport)

conn_flag = False

def w5x00_init():
    global localip
  
    spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
    nic = network.WIZNET5K(spi,Pin(17),Pin(20))
    nic.active(True)
    # use dhcp, if fail use static netinfo
    #try:
    #    nic.ifconfig('dhcp')
    #except:
    nic.ifconfig(netinfo)
    localip = nic.ifconfig()[0]
    print('ip :', nic.ifconfig()[0])
    print('sn :', nic.ifconfig()[1])
    print('gw :', nic.ifconfig()[2])
    print('dns:', nic.ifconfig()[3])
 
    
    while not nic.isconnected():
        time.sleep(1)
#         print(nic.regs())
        print('no link')
        
    return nic
        


def send_list_data(path, dataclient, full):
    try:  # whether path is a directory name
        for fname in sorted(uos.listdir(path), key=str.lower):
            dataclient.sendall(make_description(path, fname, full))
    except:  # path may be a file name or pattern
        pattern = path.split("/")[-1]
        path = path[:-(len(pattern) + 1)]
        if path == "":
            path = "/"
        for fname in sorted(uos.listdir(path), key=str.lower):
            if fncmp(fname, pattern):
                dataclient.sendall(make_description(path, fname, full))


def make_description(path, fname, full):
    if full:
        stat = uos.stat(get_absolute_path(path, fname))
        file_permissions = ("drwxr-xr-x"
                            if (stat[0] & 0o170000 == 0o040000)
                            else "-rw-r--r--")
        file_size = stat[6]
        description = "{}    1 owner group {:>10} Jan 1 2000 {}\r\n".format(
                file_permissions, file_size, fname)
    else:
        description = fname + "\r\n"
    return description


def send_file_data(path, dataclient):
    with open(path, "rb") as file:
        chunk = file.read(512)
        while len(chunk) > 0:
            dataclient.sendall(chunk)
            chunk = file.read(512)


def save_file_data(path, dataclient):
    with open(path, "wb") as file:
        chunk = dataclient.recv(512)
        while len(chunk) > 0:
            file.write(chunk)
            chunk = dataclient.recv(512)


def get_absolute_path(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 == '..':
            if cwd != '/':
                cwd = '/'.join(cwd.split('/')[:-1])
                if cwd == '':
                    cwd = '/'
        elif token != '.' and token != '':
            if cwd == '/':
                cwd += token
            else:
                cwd = cwd + '/' + token
    return cwd


# compare fname against pattern. Pattern may contain
# wildcards ? and *.
def fncmp(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 + 1) == len(pattern):
                    return True
                while si < len(fname):
                    if 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 ftpserver(net, port=21, timeout=None):

    DATA_PORT = 13333

    ftpsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    ftpsocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    ftpsocket.bind(socket.getaddrinfo("0.0.0.0", port)[0][4])
    datasocket.bind(socket.getaddrinfo("0.0.0.0", DATA_PORT)[0][4])

    ftpsocket.listen(1)
    ftpsocket.settimeout(timeout)
    datasocket.listen(1)
    datasocket.settimeout(None)

    msg_250_OK = '250 OK\r\n'
    msg_550_fail = '550 Failed\r\n'
    addr = net.ifconfig()[0]
    print("FTP Server started on ", addr)
    try:
        dataclient = None
        fromname = None
        do_run = True
        while do_run:
            cl, remote_addr = ftpsocket.accept()
            cl.settimeout(300)
            cwd = '/'
            try:
                # print("FTP connection from:", remote_addr)
                cl.sendall("220 Hello, this is Micropython.\r\n")
                while True:
                    gc.collect()
                    data = cl.readline().decode("utf-8").rstrip("\r\n")
                    if len(data) <= 0:
                        print("Client disappeared")
                        do_run = False
                        break

                    command = data.split(" ")[0].upper()
                    payload = data[len(command):].lstrip()

                    path = get_absolute_path(cwd, payload)

                    print("Command={}, Payload={}".format(command, payload))

                    if command == "USER":
                        cl.sendall("230 Logged in.\r\n")
                    elif command == "SYST":
                        cl.sendall("215 UNIX Type: L8\r\n")
                    elif command == "NOOP":
                        cl.sendall("200 OK\r\n")
                    elif command == "FEAT":
                        cl.sendall("211 no-features\r\n")
                    elif command == "PWD" or command == "XPWD":
                        cl.sendall('257 "{}"\r\n'.format(cwd))
                    elif command == "CWD" or command == "XCWD":
                        try:
                            files = uos.listdir(path)
                            cwd = path
                            cl.sendall(msg_250_OK)
                        except:
                            cl.sendall(msg_550_fail)
                    elif command == "CDUP":
                        cwd = get_absolute_path(cwd, "..")
                        cl.sendall(msg_250_OK)
                    elif command == "TYPE":
                        # probably should switch between binary and not
                        cl.sendall('200 Transfer mode set\r\n')
                    elif command == "SIZE":
                        try:
                            size = uos.stat(path)[6]
                            cl.sendall('213 {}\r\n'.format(size))
                        except:
                            cl.sendall(msg_550_fail)
                    elif command == "QUIT":
                        cl.sendall('221 Bye.\r\n')
                        do_run = False
                        break
                    elif command == "PASV":
                        cl.sendall('227 Entering Passive Mode ({},{},{}).\r\n'.
                                   format(addr.replace('.', ','), DATA_PORT >> 8,
                                          DATA_PORT % 256))
                        dataclient, data_addr = datasocket.accept()
                        print("FTP Data connection from:", data_addr)
                        DATA_PORT = 13333
                        active = False
                    elif command == "PORT":
                        items = payload.split(",")
                        if len(items) >= 6:
                            data_addr = '.'.join(items[:4])
                            # replace by command session addr
                            if data_addr == "127.0.1.1":
                                data_addr = remote_addr
                            DATA_PORT = int(items[4]) * 256 + int(items[5])
                            dataclient = socket.socket(socket.AF_INET,
                                                       socket.SOCK_STREAM)
                            dataclient.settimeout(10)
                            dataclient.connect((data_addr, DATA_PORT))
                            print("FTP Data connection with:", data_addr)
                            cl.sendall('200 OK\r\n')
                            active = True
                        else:
                            cl.sendall('504 Fail\r\n')
                    elif command == "LIST" or command == "NLST":
                        if not payload.startswith("-"):
                            place = path
                        else:
                            place = cwd
                        try:
                            cl.sendall("150 Here comes the directory listing.\r\n")
                            send_list_data(place, dataclient,
                                           command == "LIST" or payload == "-l")
                            cl.sendall("226 Listed.\r\n")
                        except:
                            cl.sendall(msg_550_fail)
                        if dataclient is not None:
                            dataclient.close()
                            dataclient = None
                    elif command == "RETR":
                        try:
                            cl.sendall("150 Opening data connection.\r\n")
                            send_file_data(path, dataclient)
                            cl.sendall("226 Transfer complete.\r\n")
                        except:
                            cl.sendall(msg_550_fail)
                        if dataclient is not None:
                            dataclient.close()
                            dataclient = None
                    elif command == "STOR":
                        try:
                            cl.sendall("150 Ok to send data.\r\n")
                            save_file_data(path, dataclient)
                            cl.sendall("226 Transfer complete.\r\n")
                        except:
                            cl.sendall(msg_550_fail)
                        if dataclient is not None:
                            dataclient.close()
                            dataclient = None
                    elif command == "DELE":
                        try:
                            uos.remove(path)
                            cl.sendall(msg_250_OK)
                        except:
                            cl.sendall(msg_550_fail)
                    elif command == "RMD" or command == "XRMD":
                        try:
                            uos.rmdir(path)
                            cl.sendall(msg_250_OK)
                        except:
                            cl.sendall(msg_550_fail)
                    elif command == "MKD" or command == "XMKD":
                        try:
                            uos.mkdir(path)
                            cl.sendall(msg_250_OK)
                        except:
                            cl.sendall(msg_550_fail)
                    elif command == "RNFR":
                            fromname = path
                            cl.sendall("350 Rename from\r\n")
                    elif command == "RNTO":
                            if fromname is not None:
                                try:
                                    uos.rename(fromname, path)
                                    cl.sendall(msg_250_OK)
                                except:
                                    cl.sendall(msg_550_fail)
                            else:
                                cl.sendall(msg_550_fail)
                            fromname = None
                    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"
                                       "211 TYPE: Binary STRU: File MODE:"
                                       " Stream\r\n".format(
                                           remote_addr[0], addr))
                        else:
                            cl.sendall("213-Directory listing:\r\n")
                            send_list_data(path, cl, True)
                            cl.sendall("213 Done.\r\n")
                    else:
                        cl.sendall("502 Unsupported command.\r\n")
                        print("Unsupported command {} with payload {}".format(
                            command, payload))
            except Exception as err:
                print(err)

            finally:
                cl.close()
                cl = None
    except Exception as e:
        print(e)
    finally:
        datasocket.close()
        ftpsocket.close()
        if dataclient is not None:
            dataclient.close()

if __name__ == "__main__":
    
    nic = w5x00_init()
    ftpserver(nic)


启动后,先与w5500建立网络连接,然后在设置ftp

 

在window的文件夹内搜索ftp服务器

 

 

 

访问成功

点赞 关注
 
 

回复
举报

21

帖子

2

TA的资源

一粒金砂(中级)

沙发
 

   

 

能够上传和下载文件

image.png (20.68 KB, 下载次数: 0)

image.png
 
 
 

回复
您需要登录后才可以回帖 登录 | 注册

查找数据手册?

EEWorld Datasheet 技术支持

相关文章 更多>>
推荐帖子
PCB设计中格点的设置

PCB设计中格点的设置合理的使用格点系统,能是我们在PCB设计中起到事半功倍的作用。但何谓合理呢? 很多人认为格点设置的越小越好 ...

STN-LCD彩屏模块

摘要:本文介绍了彩色STN-LCD模块的内部结构、主要器件,以及设计选用要求。 关键词:彩色STN-LCD模块 LCM LED 电荷泵 升压 ...

寻迹小车设计

一. 起因、目标及车体设计 一、 概述以自制方式为主制作一辆低成本的寻迹小车,用于学习嵌入式应用(单片机),这个项目有两 ...

Stellaris资料整理贴

//--------------------------------------2010.3.25---------------------------------------- 1. Stellaris大全(不断更新) ...

0-5V转换-20mA V/I电路分析

60110

使用MSP430G2 LaunchPad开发板连接步进电机

如何使用MSP430 LaunchPad开发板连接一个步进电机。 MSP-EXP430G2是德州仪器(TI)提供的开发工具,又名LaunchPad,用于学习和练 ...

#AI挑战营终点站# rv1106数字识别模型部署

本帖最后由 打破传统 于 2024-5-29 10:39 编辑 参考https://wiki.luckfox.com/zh/Luckfox-Pico/Luckfox-Pico-SDK 进行了SD ...

测评汇总:拥抱AIGC 应用ChatGPT和OpenAI API

活动详情:【拥抱AIGC 应用ChatGPT和OpenAI API】更新至 2024-10-10测评报告汇总:@皓月光兮非自明 《拥抱AIGC》四、OpenAI与GPT ...

【Follow me第二季第4期】任务一:搭建环境并开启第一步Blink三色LED / 串口打印He...

本帖最后由 mingzhe123 于 2024-12-1 12:14 编辑 2024年得捷电子和eeworld论坛联合举办的Follow me活动第二季已经进行到第4期 ...

如何用FPGA平台编写一个信号的2048点DFT,不使用fft的ip核

不使用fft的ip核实现信号2048点DFT

关闭
站长推荐上一条 1/8 下一条

 
EEWorld订阅号

 
EEWorld服务号

 
汽车开发圈

About Us 关于我们 客户服务 联系方式 器件索引 网站地图 最新更新 手机版

站点相关: 国产芯 安防电子 汽车电子 手机便携 工业控制 家用电子 医疗电子 测试测量 网络通信 物联网

北京市海淀区中关村大街18号B座15层1530室 电话:(010)82350740 邮编:100190

电子工程世界版权所有 京B2-20211791 京ICP备10001474号-1 电信业务审批[2006]字第258号函 京公网安备 11010802033920号 Copyright © 2005-2025 EEWORLD.com.cn, Inc. All rights reserved
快速回复 返回顶部 返回列表