ESP8266 MicroPython项目报错：TypeError: 函数需4个参数却传入5个

我正在开发基于MicroPython的ESP8266 WiFi连接项目，目标是当手机连接ESP8266创建的AP后，打印手机发出的DNS请求。但运行脚本时触发以下错误：

>>> import main
trying to load wifi credentials from ./wifi.creds
./wifi.creds does not exist 
starting captive portal
Waiting for access Point To turn on
#12 ets_task(4020f510, 29, 3fff9428, 10)
AP mode Configured: ('210.210.210.1', '255.255.255.0', '210.210.210.1', '210.210.210.1')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "main.py", line 5, in <module>
  File "captive_portal.py", line 37, in start
  File "captive_portal.py", line 93, in captive_portal
  File "captive_dns.py", line 55, in __init__
TypeError: function takes 4 positional arguments but 5 were given

项目相关脚本 #

main.py #

from captive_portal import CaptivePortal

portal = CaptivePortal()

portal.start()

captive_portal.py #

import network
import uerrno
import uos as os
import utime as time
import gc 
import uselect as select
from captive_dns import DNSServer

class CaptivePortal():
    AP_IP = "210.210.210.1"
    CRED_FILE = "./wifi.creds"
    MAX_CONN_ATTEMPTS = 10
    
    def __init__(self, essid = None):
        self.local_ip = self.AP_IP
        self.sta_if = network.WLAN(network.STA_IF)
        self.ap_if = network.WLAN(network.AP_IF)

        if essid is None:
            essid = "FAKEAP"
        self.essid = essid  

        self.dns_server = None
        self.poller = select.poll()  

        self.ssid = None
        self.password = None

    def start(self):
        # 关闭station接口强制重连
        self.sta_if.active(False)
        if not self.try_connect_from_file():
            self.captive_portal()

    def connect_to_wifi(self):
        print(
            "尝试使用SSID '{:s}' 和密码 {:s} 连接".format(
                self.ssid, self.password
            )
        )   
        # 初始化连接
        self.sta_if.active(True)
        self.sta_if.connect(self.ssid, self.password)

        attempts = 0 
        while attempts < self.MAX_CONN_ATTEMPTS:
            if not self.sta_if.isconnected():
                print("连接中...")
                time.sleep(2)
                attempts += 1
            else:
                print("已连接到 {:s}".format(self.ssid))
                self.local_ip = self.sta_if.ifconfig()[0] 
                self.write_creds(self.ssid, self.password)
                return True   

        print("连接 {:s} 失败，密码 {:s}。WLAN状态码={:d}".format(
            self.ssid, self.password, self.sta_if.status()
        ))        
        # 清除无效凭证，关闭station模式
        self.ssid = self.password = None
        self.sta_if.active(False)
        return False

    def write_creds(self, ssid, password):
        open(self.CRED_FILE, 'wb').write(b','.join([ssid, password]))
        print("已将凭证写入 {:s}".format(self.CRED_FILE))

    def start_access_point(self):
        # 有时需要先关闭AP才能正常启动
        self.ap_if.active(False)
        while not self.ap_if.active():
            print("等待AP启动...")
            self.ap_if.active(True)
            time.sleep(4)
        # 设置DNS服务器重定向所有客户端流量
        self.ap_if.ifconfig((self.local_ip, "255.255.255.0", self.local_ip, self.local_ip))  
        self.ap_if.config(essid=self.essid, authmode=network.AUTH_OPEN)
        print("AP模式已配置：", self.ap_if.ifconfig())

    def captive_portal(self):
        print("启动captive portal") 
        self.start_access_point()

        if self.dns_server is None:
            self.dns_server = DNSServer(self.poller, self.local_ip)
            print("DNS服务器已配置")

        try:
            while True:
                gc.collect()
                # 检查socket事件并处理
                for response in self.poller.ipoll(1000):
                    sock, events, *others = response
                    self.handle_dns(sock, events, others)
        except KeyboardInterrupt:
            print("Captive portal已停止")
            self.cleanup()

    def handle_dns(self, sock, event, others):
        if sock is self.dns_server.sock:
            # 忽略UDP套接字挂起事件
            if events == select.POLLHUP:
                return True
            self.dns_server.handle(sock, event, others)
            return True
        return False

    def cleanup(self):
        print("资源清理中")
        if self.dns_server:
            self.dns_server.stop(self.poller)
        gc.collect()                

    def try_connect_from_file(self):
        print("尝试从 {:s} 加载WiFi凭证".format(self.CRED_FILE))
        try:
            os.stat(self.CRED_FILE)
        except OSError as e:
            if e.args[0] == uerrno.ENOENT:
                print("{:s} 文件不存在".format(self.CRED_FILE))
                return False

        contents = open(self.CRED_FILE, 'rb').read().split(b',')
        if len(contents) == 2:
            self.ssid, self.password = contents
        else:
            print("凭证文件无效：", contents)
            return False

        if not self.connect_to_wifi():
            print("使用凭证连接WiFi失败，启动Captive portal")
            os.remove(self.CRED_FILE)
            return False

        return True            

captive_dns.py #

import usocket as socket
import gc

from server import Server

class DNSServer(Server):
    def __init__(self, poller, ip_addr):
        super().__init__(poller, 53, socket.SOCK_DGRAM, "DNS Server")
        self.ip_addr = ip_addr

    def handle(self, sock, event, others):
        # 服务器不会生成其他套接字，只处理自身套接字
        if sock is not self.sock:
            return

        # 解析DNS请求并返回响应
        try:
            data, sender = sock.recvfrom(1024)
            request = DNSQuery(data)

            print("转发 {:s} -> {:s}".format(request.domain, self.ip_addr))
            sock.sendto(request.answer(self.ip_addr), sender)

            # 帮助MicroPython管理内存
            del request
            gc.collect()
        except Exception as e:
            print("DNS服务器异常：", e)

server.py #

import usocket as socket
import uselect as select

class Server:
    def __init__(self, poller, sock_type, name):
        self.name = name
        # 创建指定类型的套接字：流(TCP)或数据报(UDP)
        self.sock = socket.socket(socket.AF_INET, sock_type)

        # 注册套接字以获取事件更新
        self.poller = poller
        self.poller.register(self.sock, select.POLLIN)

        addr = socket.getaddrinfo("0.0.0.0", port)[0][-1]
        # 允许在发送最后一个响应时处理新请求
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(addr)

        print(self.name , "正在监听", addr)

    def stop(self, poller):
        poller.unregister(self.sock)
        self.sock.close()
        print(self.name, "已停止")    

错误原因分析 #

报错TypeError: function takes 4 positional arguments but 5 were given出现在DNSServer初始化时调用super().__init__的位置，核心问题：

1. Server类的__init__方法定义参数为(self, poller, sock_type, name)，仅接受3个外部参数（除self外）
2. DNSServer的__init__中调用super().__init__(poller, 53, socket.SOCK_DGRAM, "DNS Server")，传入了4个外部参数，加上self共5个，与Server方法签名不匹配
3. server.py中使用了未定义的变量port，属于隐性错误

修复方案 #

1. 修改Server类的__init__方法，添加port参数 #

更新server.py：

import usocket as socket
import uselect as select

class Server:
    def __init__(self, poller, port, sock_type, name):
        self.name = name
        # 创建指定类型的套接字：流(TCP)或数据报(UDP)
        self.sock = socket.socket(socket.AF_INET, sock_type)

        # 注册套接字以获取事件更新
        self.poller = poller
        self.poller.register(self.sock, select.POLLIN)

        addr = socket.getaddrinfo("0.0.0.0", port)[0][-1]
        # 允许在发送最后一个响应时处理新请求
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(addr)

        print(self.name , "正在监听", addr)

    def stop(self, poller):
        poller.unregister(self.sock)
        self.sock.close()
        print(self.name, "已停止")    

2. 补充缺失的DNSQuery类 #

在captive_dns.py开头添加DNS请求解析类：

class DNSQuery:
    def __init__(self, data):
        self.data = data
        self.domain = ''
        # 解析DNS请求中的域名
        ptr = 12
        length = data[ptr]
        while length != 0:
            self.domain += data[ptr+1:ptr+1+length].decode('utf-8') + '.'
            ptr += length + 1
            length = data[ptr]
        self.domain = self.domain[:-1]  # 去掉末尾的点

    def answer(self, ip_addr):
        # 生成DNS响应，将域名指向指定IP
        packet = self.data[:2] + b'\x81\x80'
        packet += self.data[4:6] + self.data[4:6] + b'\x00\x00\x00\x00'
        packet += self.data[12:]
        packet += b'\xc0\x0c'
        packet += b'\x00\x01\x00\x01\x00\x00\x00\x3c\x00\x04'
        # 将IP地址转换为字节
        ip_bytes = bytes(map(int, ip_addr.split('.')))
        packet += ip_bytes
        return packet

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