---
# System prepended metadata

title: 智慧電網監控系統：十年免月租的 Things Mobile 多模全球數據傳輸方案
tags: [創客, Micropython, NBIoT, MQTT]

---

---
disqus: ahb0222
GA : G-VF9ZT413CG
---

> [color=#40f1ef][name=LHB阿好伯, 2021/07/16][:earth_africa:](https://www.facebook.com/LHB0222/)

[TOC]

****


智慧電網監控系統：十年免月租的 Things Mobile 多模數據傳輸方案
採用 Raspberry Pi Pico 微控制器
搭配 PZEM-004T 電力監測模組進行精確的電力參數採
並通過 OLED 顯示屏實現即時數據
通過軟硬體設計支持多模通信，實現了 WiFi 和4G網絡（通過 EC800K 4G 模組）的自動切換
![20240728_222612](https://hackmd.io/_uploads/HyFBTC7tR.jpg)
![image](https://hackmd.io/_uploads/H1REgJVKC.png)

這種設計大大提高了數據傳輸的可靠性和適應性
能夠在各種網絡環境下保持穩定運行
[採用了 Things Mobile 的 IoT SIM 卡](https://www.thingsmobile.com/business)
![Snipaste_2024-07-28_22-29-22](https://hackmd.io/_uploads/S1ygRR7YA.png)

這不僅實現了十年免月租
![Snipaste_2024-07-28_22-31-50](https://hackmd.io/_uploads/rkCZA07KR.png)

還支持全球165+國家的網絡覆蓋
![image](https://hackmd.io/_uploads/SJIEACXKC.png)

系統採用彈性計費模式，只對實際使用的數據量收費
![Snipaste_2024-07-28_22-28-12](https://hackmd.io/_uploads/BJWW0R7tA.png)

極大地優化了長期運營成本。
這個解決方案的優勢在於其高度的靈活性和可擴展性
它可以輕易適應不同的監控需求
從小型智能家居到大規模工業應用


| 特徵 | 中華電信 NB-IoT | 中華電信 LTE-M | Things Mobile |
|-----|----------------|----------------|---------------|
| 資費方案 | 物聯NB-A/B/C型 | 物聯M1-A/B/C型 | 單一方案 |
| 月租費 | 10元/25元/60元 | 20元/40元/80元 | 無月租費 |
| 啟用費 | 未提供資訊 | 未提供資訊 | 約150元（5美元） |
| 國內數據內含量 | 5MB/15MB/40MB | 5MB/15MB/40MB | 可無內含量 |
| 超量費率（每MB） | 約1.536元 | 約3.072元 | 約3.9元（0.12美元） |
| 數據收費上限 | 750元（含月租費） | 750元（含月租費） | 無上限(可設定) |
| 網路技術 | NB-IoT | LTE-M | 多種網路技術 |
| 適用地區 | 台灣 | 台灣 | 全球165+國家 |
| 合約限制 | 無資料 | 無資料 | 無合約限制 |


```python=
import network
import time
from machine import Pin, I2C, ADC, UART, WDT
from ssd1306 import SSD1306_I2C
from umqtt.simple import MQTTClient
import ujson
from pzem import PZEM

# WiFi 设置
WIFI_SSID = "WIFI_SSID"
WIFI_PASSWORD = "WIFI_PASSWORD"

# MQTT 设置
MQTT_BROKER = "MQTT_BROKER"
MQTT_PORT = 1883
MQTT_USER = ""
MQTT_PASSWORD = ""
MQTT_CLIENT_ID = "RP2040Client"
MQTT_TOPIC = "MQTT_TOPIC"
WILL_TOPIC = "WILL_TOPIC"
WILL_MESSAGE = b"offline"

# OLED 设置
OLED_WIDTH = 128
OLED_HEIGHT = 64
OLED_ADDR = 0x3C
OLED_I2C_SCL = 17  # SCL -> GP17
OLED_I2C_SDA = 16  # SDA -> GP16

# LED 设置
LED_PIN = 0  # GPIO0

# PZEM-004T 设置
PZEM_UART_ID = 1
PZEM_TX_PIN = 8  # GP8
PZEM_RX_PIN = 9  # GP9

# EC800K 设置
EC800K_UART_ID = 0
EC800K_TX_PIN = 12  # GP12
EC800K_RX_PIN = 13  # GP13


wifi = None
mqtt_client = None
oled = None
pzem = None
ec800k = None
wdt = None
led = None

def scan_i2c():
    i2c = I2C(0, scl=Pin(17), sda=Pin(16))
    devices = i2c.scan()
    if devices:
        for device in devices:
            print(f"I2C device found at address: 0x{device:02x}")
    else:
        print("No I2C device found")

def init_led():
    global led
    led = Pin(LED_PIN, Pin.OUT)

def init_wifi():
    global wifi
    wifi = network.WLAN(network.STA_IF)
    wifi.active(True)
    wifi.connect(WIFI_SSID, WIFI_PASSWORD)
    
    max_attempts = 10
    for attempt in range(max_attempts):
        if wifi.isconnected():
            print("Connected to WiFi")
            return True
        print(f"Attempting to connect to WiFi... ({attempt+1}/{max_attempts})")
        time.sleep(1)
        wdt.feed()
    
    print("Failed to connect to WiFi")
    return False

def init_mqtt():
    global mqtt_client
    mqtt_client = MQTTClient(MQTT_CLIENT_ID, MQTT_BROKER, MQTT_PORT, MQTT_USER, MQTT_PASSWORD, keepalive=60)
    mqtt_client.set_last_will(WILL_TOPIC, WILL_MESSAGE, retain=True, qos=1)
    try:
        mqtt_client.connect()
        print("Connected to MQTT broker")
        return True
    except:
        print("Failed to connect to MQTT broker")
        return False

def init_oled():
    global oled
    try:
        i2c_dev = I2C(0, scl=Pin(OLED_I2C_SCL), sda=Pin(OLED_I2C_SDA), freq=400000)
        i2c_addr = [hex(ii) for ii in i2c_dev.scan()]  
        if not i2c_addr:
            print('未找到I2C')
            return None
        
        print(f"I2C 地址: {i2c_addr[0]}")
        print(f"I2C 配置: {i2c_dev}")
        
        oled = SSD1306_I2C(OLED_WIDTH, OLED_HEIGHT, i2c_dev)
        print("OLED 初始化成功")
        return oled
    except Exception as e:
        print(f"OLED 初始化錯誤: {e}")
        return None

def display_info(oled, data, internal_temp):
    if oled:
        oled.fill(0)  
        oled.text(f"V:{data['voltage']:.3f} V", 0, 0)
        oled.text(f"I:{data['current']:.3f} A", 0, 10)
        oled.text(f"P:{data['power']:.3f} W", 0, 20)
        oled.text(f"E:{data['energy']:.3f} Wh", 0, 30)
        oled.text(f"F:{data['frequency']:.1f}Hz;", 0, 40)
        oled.text(f"Temp:{internal_temp:.1f}", 0, 50)
        oled.text(f"PF:{data['pf']:.2f}", 72, 40)
        oled.show()

def init_pzem():
    global pzem
    try:
        uart = UART(PZEM_UART_ID, baudrate=9600, tx=Pin(PZEM_TX_PIN), rx=Pin(PZEM_RX_PIN))
        pzem = PZEM(uart=uart)
        if pzem.readAddress():  
            print("PZEM initialized successfully")
        else:
            print("Failed to communicate with PZEM")
            pzem = None
    except Exception as e:
        print(f"Error initializing PZEM: {e}")
        pzem = None
        
def init_ec800k():
    global ec800k
    wdt.feed()
    try:
        ec800k = UART(EC800K_UART_ID, baudrate=115200, tx=Pin(EC800K_TX_PIN), rx=Pin(EC800K_RX_PIN))
        commands = [
            "AT+CGATT=0",
            "AT+CGATT=1",
            "AT+COPS=0",
            'AT+QICSGP=1,1,"TM"',
            "AT+QIACT=1",
            'AT+QMTOPEN=0,"broker.emqx.io",1883',
            'AT+QMTCONN=0,"mqttx_7afa236d","",""'
        ]
        
        for cmd in commands:
            ec800k.write(cmd + "\r\n")
            time.sleep(2) 
            wdt.feed()
            response = ec800k.read()
            print(f"Command: {cmd}")
            print(f"Response: {response}")
            if b'OK' not in response and b'QMTOPEN: 0,0' not in response and b'QMTCONN: 0,0,0' not in response:
                print(f"EC800K initialization failed at command: {cmd}")
                return False
        
        print("EC800K initialized successfully")
        return True
    except Exception as e:
        print(f"Error initializing EC800K: {e}")
        return False

def read_pzem():
    if pzem is not None:
        pzem.uart.read() 
        if pzem.read():
            return {
                "voltage": pzem.getVoltage(),
                "current": pzem.getCurrent(),
                "power": pzem.getActivePower(),
                "energy": pzem.getActiveEnergy(),
                "frequency": pzem.getFrequency(),
                "pf": pzem.getPowerFactor()
            }
    return None


def send_data_via_ec800k(topic, payload):
    try:
        command = f'AT+QMTPUBEX=0,1,1,0,"{topic}",{len(payload)}\r\n'
        ec800k.write(command)
        time.sleep(0.5)
        response = ec800k.read()
        print(f"QMTPUBEX command response: {response}")
        
        if b'>' in response:
            ec800k.write(payload + '\r\n')
            time.sleep(1)
            response = ec800k.read()
            print(f"Payload send response: {response}")
            if b'+QMTPUBEX: 0,0,0' in response:
                print("Data sent successfully via EC800K")
                return True
            else:
                print("Failed to send data via EC800K")
                return False
        else:
            print("Did not receive '>' prompt")
            return False
    except Exception as e:
        print(f"Error sending data via EC800K: {e}")
        return False

def send_data(topic, payload):
    if wifi.isconnected():
        try:
            mqtt_client.publish(topic, payload)
            print(f"Data sent via MQTT: {payload}")
        except:
            print("Failed to send data via MQTT")
    else:
        send_data_via_ec800k(topic, payload)
        print(f"Data sent via EC800K: {payload}")

def read_internal_temperature():
    adc = ADC(4)  # ADC for internal temperature sensor
    raw = adc.read_u16()
    voltage = (raw * 3.3) / 65535
    temperature = 27 - (voltage - 0.706) / 0.001721
    return temperature

def read_and_display_pzem_data():
    internal_temp = read_internal_temperature()
    if pzem is not None:
        pzem_data = read_pzem()
        if pzem_data:
            if oled:
                display_info(oled, pzem_data, internal_temp)
            else:
                print("OLED update skipped")
            
            pzem_data['ConnectionType'] = 'WiFi' if wifi.isconnected() else 'EC800K'
            pzem_data['internal_temp'] = internal_temp
            send_data("ahb0222/TM/908/PZEM", ujson.dumps(pzem_data))
            return pzem_data, internal_temp
        else:
            print("Failed to read PZEM data")
    else:
        print("PZEM not initialized, skipping data reading")
    return None, internal_temp

def main():
    global wdt
    
    init_led()
    wdt = WDT(timeout=8000)  # 8 seconds timeout
    
    oled = init_oled()
    if oled:
        display_info(oled, {"voltage": 0, "current": 0, "power": 0, "energy": 0, "frequency": 0, "pf": 0}, 0)
        time.sleep(2)  
    
    scan_i2c()  
    
    wifi_connected = False
    mqtt_connected = False
    ec800k_initialized = False
    
    while True:
        try:
            wdt.feed()
            
            if not wifi_connected:
                print("Attempting to connect to WiFi...")
                wifi_connected = init_wifi()
                if wifi_connected:
                    mqtt_connected = init_mqtt()
                    if not mqtt_connected:
                        print("MQTT initialization failed, switching to EC800K")
                        init_ec800k()
                        ec800k_initialized = True
                else:
                    print("WiFi connection failed, switching to EC800K")
                    init_ec800k()
                    ec800k_initialized = True
            
            if wifi_connected and not wifi.isconnected():
                print("WiFi connection lost. Attempting to reconnect...")
                wifi_connected = init_wifi()
                if wifi_connected:
                    mqtt_connected = init_mqtt()
                else:
                    print("WiFi reconnection failed, switching to EC800K")
                    init_ec800k()
                    ec800k_initialized = True
            
            if pzem is None:
                print("Attempting to initialize PZEM...")
                init_pzem()
            
            pzem_data, internal_temp = read_and_display_pzem_data()
            
            if pzem_data:
                temp_data = {
                    "tamp": time.time(),
                    "temp": internal_temp,
                    "ConnectionType": 'WiFi' if wifi_connected else 'EC800K'
                }
                if wifi_connected and mqtt_connected:
                    try:
                        #send_data(MQTT_TOPIC, ujson.dumps(temp_data))
                        print("")
                    except Exception as e:
                        print(f"Error sending data via MQTT: {e}")
                        mqtt_connected = False
                elif ec800k_initialized:
                    try:
                        #send_data_via_ec800k(MQTT_TOPIC, ujson.dumps(temp_data))
                        print("")
                    except Exception as e:
                        print(f"Error sending data via EC800K: {e}")
                        ec800k_initialized = False
                else:
                    print("No available connection method")
            
            led.toggle()
            time.sleep(2)
        
        except Exception as e:
            print(f"Error in main loop: {e}")
            

if __name__ == "__main__":
    main()


```
🌟

🌟全文可以至下方連結觀看或是補充


全文分享至

https://www.facebook.com/LHB0222/

https://www.instagram.com/ahb0222/

有疑問想討論的都歡迎於下方留言

喜歡的幫我分享給所有的朋友 \o/

有所錯誤歡迎指教

# [:page_with_curl: 全部文章列表](https://hackmd.io/@LHB-0222/AllWritings)



![](https://i.imgur.com/nHEcVmm.jpg)


