树莓派Python multiprocessing.Array报TypeError问题排查
问题原因
你遇到的TypeError: len() of unsized object根源有两处:
- Python3中
map()返回的是惰性迭代器,不是可直接计算长度的序列对象。你在splitToDisplay里写np.array(map(float, arrToDisplay))时,numpy拿到的是未展开的迭代器,生成的数组没有固定长度,multiprocessing.Array初始化时需要读取传入序列的长度分配共享内存,拿不到长度就直接抛出错误。 - 代码里还存在多进程逻辑错误、GPIO操作冲突、按键逻辑bug、缺少多进程入口保护等多个问题,就算修完上面的类型错误,程序也没法正常运行。
具体修复方案
- 首先修复
splitToDisplay函数的类型问题,返回原生Python列表,不要直接传numpy数组给共享内存,同时处理空格转义避免转float报错:
def splitToDisplay(toDisplay): arrToDisplay = list(toDisplay) for i in range(len(arrToDisplay)): if arrToDisplay[i] == ".": arrToDisplay[i-1] = arrToDisplay[i-1] + arrToDisplay[i] while "." in arrToDisplay: arrToDisplay.remove(".") # 补全4位长度,空格转0.0避免float转换报错 while len(arrToDisplay) < 4: arrToDisplay.append(" ") return [float(c) if c != " " else 0.0 for c in arrToDisplay]
- 修复按键读取的索引bug,第四列按键错误取了第一个字符:
def readLine(line, characters): GPIO.output(line, GPIO.HIGH) if(GPIO.input(C1) == 1): print(characters[0]) user_input.append(characters[0]) elif(GPIO.input(C2) == 1): print(characters[1]) user_input.append(characters[1]) elif(GPIO.input(C3) == 1): print(characters[2]) user_input.append(characters[2]) elif(GPIO.input(C4) == 1): # 原来这里写的characters[0]是错的 print(characters[3]) user_input.append(characters[3]) GPIO.output(line, GPIO.LOW)
- 修正共享内存初始化逻辑,不要在顶层直接传处理后的不确定长度序列,直接指定长度初始化4位双精度共享数组:
# 初始化4个double类型元素的共享数组,初始值全为0.0 displayArr = multiprocessing.Array('d', 4)
- 修正两个子进程的逻辑,不要在子进程里给局部变量赋值,要直接读写共享内存,同时把所有GPIO数码管操作都放在display进程里,避免跨进程操作GPIO冲突:
def display(displayArr): while True: # 从共享内存读取当前要显示的数值 current_digits = [] for i in range(4): val = displayArr[i] current_digits.append(str(int(val)) if val != 0 else " ") showDisplay(current_digits) def keypad(displayArr): while True: enterPasscode() # 把新输入的密码写入共享内存 display_vals = splitToDisplay(user_input) for idx, val in enumerate(display_vals): displayArr[idx] = val if user_input == code: print("Correct Passcode!") break else: print("Incorrect Passcode")
- 修正主程序入口,增加多进程必须的主模块判断,避免子进程重复初始化GPIO,退出时自动清理GPIO状态:
if __name__ == "__main__": displayProcess = multiprocessing.Process(target=display, args=(displayArr,)) keypadIntake = multiprocessing.Process(target=keypad, args=(displayArr,)) try: keypadIntake.start() displayProcess.start() keypadIntake.join() displayProcess.terminate() GPIO.cleanup() except KeyboardInterrupt: print("\nApplication Stopped!") GPIO.cleanup()
注意:删掉原来主程序里
keypadIntake.start()之后直接调用showDisplay的代码,所有数码管刷新逻辑都放在display子进程里执行,避免多进程同时操作GPIO引脚导致状态混乱。
内容的提问来源于stack exchange,提问作者MissMJRocks
相关产品推荐
相关产品推荐

