將水錶連接到智能家居

曾幾何時,家庭自動化系統(通常被稱為“智能家居”)非常昂貴,只有富人才能買得起。 如今,您可以在市場上找到相當便宜的套件,配有傳感器、按鈕/開關和執行器,用於控制照明、插座、通風、供水和其他消費者。 即使是最狡猾的 DIY 人也可以涉足美容領域,並以低廉的價格為智能家居組裝設備。

將水錶連接到智能家居

通常,所提出的設備是傳感器或執行器。 它們可以輕鬆實現“當運動傳感器被觸發時,打開燈”或“出口附近的開關關閉整個公寓的燈”等場景。 但不知怎的,遙測技術並沒有解決問題。 充其量,它是溫度和濕度的圖表,或者特定插座的瞬時功率。

我最近安裝了帶脈衝輸出的水錶。 每經過一升水,簧片開關就會被激活並閉合觸點。 剩下要做的唯一一件事就是緊緊抓住電線並嘗試從中受益。 例如,按小時和星期幾分析用水量。 好吧,如果公寓裡有多個升水器,那麼在一個屏幕上查看所有當前指示器比用手電筒爬進難以到達的壁龕更方便。

下面是我基於 ESP8266 的設備版本,它對水錶的脈衝進行計數,並通過 MQTT 將讀數發送到智能家居服務器。 我們將使用 uasyncio 庫在 micropython 中進行編程。 在創建固件時,我遇到了一些有趣的困難,我也將在本文中討論這些困難。 去!

駕駛

將水錶連接到智能家居

整個電路的核心是 ESP8266 微控制器上的模塊。 ESP-12 原本是計劃中的,但結果證明我的有缺陷。 我們必須對可用的 ESP-07 模塊感到滿意。 幸運的是,它們在引腳和功能方面都是相同的,唯一的區別在於天線 - ESP-12 具有內置天線,而 ESP-07 具有外置天線。 不過,即使沒有WiFi天線,我家浴室裡的信號也能正常接收。

標準模塊接線:

  • 帶上拉電阻和電容器的複位按鈕(儘管兩者都已位於模塊內部)
  • 使能信號(CH_PD)上拉至電源
  • GPIO15 被拉至地。 這只是開始時需要的,但我仍然沒有任何東西可以附加到這條腿上;我不再需要它

為了讓模塊進入固件模式,需要將GPIO2短路到地,為了更方便,我提供了一個Boot按鈕。 在正常情況下,該引腳被上拉至電源。

僅在操作開始時檢查 GPIO2 線路的狀態 - 通電時或複位後立即檢查。 因此,模塊要么照常啟動,要么進入固件模式。 加載後,該引腳可用作常規 GPIO。 好吧,由於那裡已經有一個按鈕,您可以為其附加一些有用的功能。

對於編程和調試,我將使用 UART,它輸出到梳子。 必要時,我只需在那裡連接一個 USB-UART 適配器即可。 您只需要記住該模塊由 3.3V 供電。 如果忘記將適配器切換到該電壓並提供 5V,則模塊很可能會燒毀。

我在浴室裡用電沒有問題 - 插座距離電錶大約一米,所以我將使用 220V 供電。 作為電源我會有一個小的 塊HLK-PM03 由騰星機器人設計。 就我個人而言,我對模擬和電力電子器件不太熟悉,但這裡有一個裝在小盒子裡的現成電源。

為了指示操作模式,我提供了一個連接到 GPIO2 的 LED。 不過,我沒有拆焊它,因為…… ESP-07 模塊已經有一個 LED,並且也連接到 GPIO2。 但讓它在板上,以防萬一我想將這個 LED 輸出到機箱上。

讓我們繼續最有趣的部分。 水錶沒有邏輯;你不能向他們詢問當前的讀數。 我們唯一可用的是脈衝——每升關閉簧片開關的觸點。 我的簧片開關輸出連接到 GPIO12/GPIO13。 我將在模塊內以編程方式啟用上拉電阻。

最初,我忘記提供電阻器 R8 和 R9,而我的開發板版本沒有它們。 但由於我已經將圖表發布給大家看,所以值得糾正這個疏忽。 需要電阻器,以免在固件出現故障並將引腳設置為 3.3 時燒毀端口,並且簧片開關將該線短路到地(電阻器最大電流為 1000V/3.3Ohm = XNUMXmA)。

是時候考慮一​​下如果停電了該怎麼辦。 第一個選項是在開始時向服務器請求初始計數器值。 但這需要使交換協議變得非常複雜。 此外,在這種情況下設備的性能取決於服務器的狀態。 如果斷電後(或稍後啟動)服務器沒有啟動,水錶將無法請求初始值,無法正常工作。

因此,我決定在通過I2C連接的存儲芯片中實現保存計數器值。 我對閃存的大小沒有任何特殊要求——只需要保存2個數字(根據冷熱水錶的升數)。 即使是最小的模塊也可以。 但需要注意記錄週期的數量。 對於大多數模塊來說,這是 100 萬個週期,有些模塊可達 XNUMX 萬個週期。

看起來一百萬已經很多了。 但在我住的公寓的四年裡,我消耗了4多立方米的水,也就是500萬升! 以及閃存中的 500 萬條記錄。 那隻是冷水。 當然,您可以每隔幾年重新焊接一次芯片,但事實證明有 FRAM 芯片。 從編程的角度來看,這與 I500C EEPROM 相同,只是重寫週期數非常多(數億)。 只是我仍然無法去商店購買這種微電路,所以目前通常的 2LC24 就可以了。

印刷電路板

最初,我計劃在家製作電路板。 因此,該板被設計為單面的。 但在用激光烙鐵和阻焊膜花了一個小時後(沒有它就有點不正常),我仍然決定從中國訂購電路板。

將水錶連接到智能家居

幾乎在訂購該板之前,我意識到除了閃存芯片之外,我還可以將其他有用的東西連接到 I2C 總線,例如顯示器。 究竟要輸出什麼仍然是一個問題,但需要在板上進行佈線。 好吧,由於我要從工廠訂購主板,所以沒有必要將自己限制在單面板上,因此 I2C 線是主板背面唯一的線路。

單向佈線還存在一個大問題。 因為電路板被繪製為單面,因此軌道和 SMD 元件計劃放置在一側,輸出元件、連接器和電源放置在另一側。 一個月後,當我收到電路板時,我忘記了最初的計劃,並將所有組件焊接在正面。 只有當焊接電源時,才發現正負極接反了。 我不得不和跳線一起耕種。 在上圖中,我已經改變了接線,但是接地通過啟動按鈕的引腳從板的一個部分轉移到另一個部分(儘管可以在第二層上畫一條軌道)。

結果是這樣的

將水錶連接到智能家居

住房

下一步是身體。 如果您有 3D 打印機,這不是問題。 我並沒有太費心——我只是畫了一個合適尺寸的盒子,並在合適的地方做了切口。 蓋子通過小自攻螺釘固定在主體上。

將水錶連接到智能家居

我已經提到過“啟動”按鈕可以用作通用按鈕 - 因此我們將其顯示在前面板上。 為此,我在按鈕所在的位置畫了一個特殊的“井”。

將水錶連接到智能家居

外殼內部還有螺柱,用於安裝電路板並用單個 M3 螺釘固定(電路板上沒有更多空間)

當我打印案例的第一個樣本版本時,我已經選擇了顯示器。 標準的兩行讀卡器不適合這種情況,但底部有一個 OLED 顯示屏 SSD1306 128×32。 它有點小,但我不必每天盯著它——這對我來說太多了。

弄清楚這種方式以及如何從其佈線,我決定將顯示器粘在外殼的中間。 當然,人體工學效果低於標準——按鈕位於頂部,顯示屏位於底部。 但我已經說過,連接顯示器的想法來得太晚了,我懶得重新連接電路板來移動按鈕。

設備已組裝完畢。 將顯示模塊用熱膠粘在鼻涕上

將水錶連接到智能家居

將水錶連接到智能家居

最終結果可以在KDPV上看到

插入

讓我們繼續討論軟件部分。 對於像這樣的小工藝品,我非常喜歡使用Python(微蟒) - 代碼非常緊湊且易於理解。 幸運的是,無需深入到寄存器級別來擠出微秒 - 一切都可以通過 Python 完成。

看起來一切都很簡單,但又不是很簡單——該設備有幾個獨立的功能:

  • 用戶按下按鈕並查看顯示屏
  • 升刻度並更新閃存中的值
  • 模塊監測WiFi信號,必要時重新連接
  • 好吧,如果沒有閃爍的燈泡,這是不可能的

如果某個功能由於某種原因被卡住,您不能假設另一個功能不起作用。 我已經在其他項目中使用了仙人掌,但現在我仍然看到“由於當時顯示器正在更新而錯過了另一升”或“當模塊連接到時用戶無法執行任何操作”之類的故障。無線上網。 ” 當然,有些事情可以通過中斷來完成,但您可能會遇到持續時間、調用嵌套或變量的非原子更改方面的限制。 嗯,做所有事情的代碼很快就會變得一團糟。

В 更嚴肅的項目 我使用了經典的搶占式多任務處理和 FreeRTOS,但在這種情況下,該模型更合適 協程和 uasync 庫 。 而且,協程的 Python 實現簡直令人驚嘆 - 對於程序員來說,一切都簡單方便地完成。 只需編寫您自己的邏輯即可,只需告訴我您可以在哪些位置之間切換流即可。

我建議將先發製人和競爭性多任務處理之間的差異作為選修科目進行研究。 現在讓我們最後繼續討論代碼。

#####################################
# Counter class - implements a single water counter on specified pin
#####################################
class Counter():
    debounce_ms = const(25)
    
    def __init__(self, pin_num, value_storage):
        self._value_storage = value_storage
        
        self._value = self._value_storage.read()
        self._value_changed = False

        self._pin = Pin(pin_num, Pin.IN, Pin.PULL_UP)

        loop = asyncio.get_event_loop()
        loop.create_task(self._switchcheck())  # Thread runs forever

每個計數器都由 Counter 類的一個實例處理。 首先,從EEPROM(value_storage)中減去初始計數器值——這就是實現斷電後恢復的方法。

該引腳通過內置上拉至電源進行初始化:如果簧片開關閉合,則線路為零,如果線路打開,則將其上拉至電源,控制器讀取 XNUMX。

這裡還啟動了一個單獨的任務,它將輪詢 pin。 每個計數器將運行自己的任務。 這是她的代碼

    """ Poll pin and advance value when another litre passed """
    async def _switchcheck(self):
        last_checked_pin_state = self._pin.value()  # Get initial state

        # Poll for a pin change
        while True:
            state = self._pin.value()
            if state != last_checked_pin_state:
                # State has changed: act on it now.
                last_checked_pin_state = state
                if state == 0:
                    self._another_litre_passed()

            # Ignore further state changes until switch has settled
            await asyncio.sleep_ms(Counter.debounce_ms)

需要 25ms 的延遲來過濾接觸反彈,同時調節任務喚醒的頻率(當該任務處於睡眠狀態時,其他任務正在運行)。 該功能每 25 毫秒喚醒一次,檢查引腳,如果簧片開關觸點閉合,則另一升液體已通過流量計,需要對此進行處理。

    def _another_litre_passed(self):
        self._value += 1
        self._value_changed = True

        self._value_storage.write(self._value)

處理下一升是微不足道的——計數器只是增加。 好吧,最好將新值寫入閃存驅動器。

為了便於使用,提供了“訪問器”

    def value(self):
        self._value_changed = False
        return self._value

    def set_value(self, value):
        self._value = value
        self._value_changed = False

好吧,現在讓我們利用 Python 和 uasync 庫的優勢,製作一個可等待的計數器對象(我們如何將其翻譯成俄語?您可以期待的那個嗎?)

    def __await__(self):
        while not self._value_changed:
            yield from asyncio.sleep(0)

        return self.value()

    __iter__ = __await__  

這是一個非常方便的函數,它會等待計數器值更新 - 該函數會不時喚醒並檢查 _value_changed 標誌。 該函數最酷的一點是,調用代碼可以在調用該函數時進入休眠狀態,並休眠直到收到新值。

打擾了怎麼辦?是的,此時你可以對我進行惡搞,說你自己說過關於打擾的事情,但實際上你做了一個愚蠢的民意調查。 實際上中斷是我嘗試的第一件事。 在 ESP8266 中,您可以組織邊沿中斷,甚至可以用 Python 編寫該中斷的處理程序。 在此中斷中,可以更新變量的值。 如果計數器是一個從設備,那麼這可能就足夠了——一個等待直到被請求該值的設備。

不幸的是(或者幸運的是?)我的設備處於活動狀態,它本身必須通過 MQTT 協議發送消息並將數據寫入 EEPROM。 這裡限制發揮作用 - 您不能在中斷中分配內存並使用大堆棧,這意味著您可以忘記通過網絡發送消息。 有像 micropython.schedule() 這樣的包允許你“盡快”運行某些函數,但問題出現了,“有什麼意義?” 如果我們現在正在發送某種消息,然後出現中斷並破壞了變量的值,該怎麼辦? 或者,例如,一個新的計數器值從服務器到達,而我們尚未寫下舊的計數器值。 一般來說,您需要阻止同步或以不同的方式擺脫它。

有時會發生 RuntimeError: Schedule stack full crashs,誰知道為什麼?

通過顯式輪詢和 uasync,在這種情況下,它在某種程度上變得更加美觀和可靠

我將 EEPROM 的工作帶到了一個小班

class EEPROM():
    i2c_addr = const(80)

    def __init__(self, i2c):
        self.i2c = i2c
        self.i2c_buf = bytearray(4) # Avoid creation/destruction of the buffer on each call


    def read(self, eeprom_addr):
        self.i2c.readfrom_mem_into(self.i2c_addr, eeprom_addr, self.i2c_buf, addrsize=16)
        return ustruct.unpack_from("<I", self.i2c_buf)[0]    
        
    
    def write(self, eeprom_addr, value):
        ustruct.pack_into("<I", self.i2c_buf, 0, value)
        self.i2c.writeto_mem(self.i2c_addr, eeprom_addr, self.i2c_buf, addrsize=16)

在Python中,直接使用字節是很困難的,但寫入內存的是字節。 我必須使用 ustruct 庫來隔離整數和字節之間的轉換。

為了不每次都傳輸I2C對象和內存單元的地址,我把它全部包裝在一個小而方便的經典中

class EEPROMValue():
    def __init__(self, i2c, eeprom_addr):
        self._eeprom = EEPROM(i2c)
        self._eeprom_addr = eeprom_addr
        

    def read(self):
        return self._eeprom.read(self._eeprom_addr)


    def write(self, value):
        self._eeprom.write(self._eeprom_addr, value)

I2C 對象本身是使用這些參數創建的

i2c = I2C(freq=400000, scl=Pin(5), sda=Pin(4))

我們來到了最有趣的部分——通過 MQTT 與服務器通信的實現。 好吧,不需要實現協議本身 - 我在互聯網上找到了它 現成的異步實現。 這就是我們將要使用的。

所有最有趣的東西都集中在 CounterMQTTClient 類中,該類基於 MQTTClient 庫。 我們先從外圍開始

#####################################
# Class handles both counters and sends their status to MQTT
#####################################
class CounterMQTTClient(MQTTClient):

    blue_led = Pin(2, Pin.OUT, value = 1)
    button = Pin(0, Pin.IN)

    hot_counter = Counter(12, EEPROMValue(i2c, EEPROM_ADDR_HOT_VALUE))
    cold_counter = Counter(13, EEPROMValue(i2c, EEPROM_ADDR_COLD_VALUE))

您可以在此處創建和配置燈泡引腳和按鈕,以及冷熱水錶對象。

有了初始化,一切就不再那麼微不足道了

    def __init__(self):
        self.internet_outage = True
        self.internet_outages = 0
        self.internet_outage_start = ticks_ms()

        with open("config.txt") as config_file:
            config['ssid'] = config_file.readline().rstrip()
            config['wifi_pw'] = config_file.readline().rstrip()
            config['server'] = config_file.readline().rstrip()
            config['client_id'] = config_file.readline().rstrip()
            self._mqtt_cold_water_theme = config_file.readline().rstrip()
            self._mqtt_hot_water_theme = config_file.readline().rstrip()
            self._mqtt_debug_water_theme = config_file.readline().rstrip()

        config['subs_cb'] = self.mqtt_msg_handler
        config['wifi_coro'] = self.wifi_connection_handler
        config['connect_coro'] = self.mqtt_connection_handler
        config['clean'] = False
        config['clean_init'] = False
        super().__init__(config)

        loop = asyncio.get_event_loop()
        loop.create_task(self._heartbeat())
        loop.create_task(self._counter_coro(self.cold_counter, self._mqtt_cold_water_theme))
        loop.create_task(self._counter_coro(self.hot_counter, self._mqtt_hot_water_theme))
        loop.create_task(self._display_coro())

為了設置 mqtt_as 庫的操作參數,使用了一個包含不同設置的大字典 - config。 大多數默認設置對我們來說都很好,但許多設置需要明確設置。 為了不直接在代碼中寫入設置,我將它們存儲在文本文件config.txt中。 這允許您更改代碼,無論設置如何,以及鉚接多個具有不同參數的相同設備。

最後一段代碼啟動幾個協程來服務系統的各種功能。 例如,這是一個為計數器提供服務的協程

    async def _counter_coro(self, counter, topic):
        # Publish initial value
        value = counter.value()
        await self.publish(topic, str(value))

        # Publish each new value
        while True:
            value = await counter
            await self.publish_msg(topic, str(value))

協程循環等待新的計數器值,一旦出現,就會通過 MQTT 協議發送消息。 即使沒有水流過計數器,第一段代碼也會發送初始值。

基類 MQTTClient 為自身提供服務,發起 WiFi 連接並在連接丟失時重新連接。 當WiFi連接狀態發生變化時,庫通過調用wifi_connection_handler通知我們

    async def wifi_connection_handler(self, state):
        self.internet_outage = not state
        if state:
            self.dprint('WiFi is up.')
            duration = ticks_diff(ticks_ms(), self.internet_outage_start) // 1000
            await self.publish_debug_msg('ReconnectedAfter', duration)
        else:
            self.internet_outages += 1
            self.internet_outage_start = ticks_ms()
            self.dprint('WiFi is down.')
            
        await asyncio.sleep(0)

該函數是從示例中誠實複製的。 在這種情況下,它會計算中斷次數 (internet_outages) 及其持續時間。 當連接恢復時,空閒時間會發送到服務器。

順便說一句,最後一次睡眠只需要使函數異步 - 在庫中它是通過await 調用的,並且只有主體包含另一個await 的函數才能被調用。

除了連接 WiFi 之外,您還需要建立與 MQTT 代理(服務器)的連接。 庫也這樣做,當建立連接時我們有機會做一些有用的事情

    async def mqtt_connection_handler(self, client):
        await client.subscribe(self._mqtt_cold_water_theme)
        await client.subscribe(self._mqtt_hot_water_theme)

這裡我們訂閱了幾條消息——服務器現在能夠通過發送相應的消息來設置當前的計數器值。

    def mqtt_msg_handler(self, topic, msg):
        topicstr = str(topic, 'utf8')
        self.dprint("Received MQTT message topic={}, msg={}".format(topicstr, msg))

        if topicstr == self._mqtt_cold_water_theme:
            self.cold_counter.set_value(int(msg))

        if topicstr == self._mqtt_hot_water_theme:
            self.hot_counter.set_value(int(msg))

該函數處理傳入的消息,並根據主題(消息標題)更新其中一個計數器的值

幾個輔助函數

    # Publish a message if WiFi and broker is up, else discard
    async def publish_msg(self, topic, msg):
        self.dprint("Publishing message on topic {}: {}".format(topic, msg))
        if not self.internet_outage:
            await self.publish(topic, msg)
        else:
            self.dprint("Message was not published - no internet connection")

如果連接建立,此函數會發送一條消息。 如果沒有連接,則該消息將被忽略。

這只是一個生成和發送調試消息的便捷功能。

    async def publish_debug_msg(self, subtopic, msg):
        await self.publish_msg("{}/{}".format(self._mqtt_debug_water_theme, subtopic), str(msg))

文字太多,我們還沒有閃爍 LED 燈。 這裡

    # Blink flash LED if WiFi down
    async def _heartbeat(self):
        while True:
            if self.internet_outage:
                self.blue_led(not self.blue_led()) # Fast blinking if no connection
                await asyncio.sleep_ms(200) 
            else:
                self.blue_led(0) # Rare blinking when connected
                await asyncio.sleep_ms(50)
                self.blue_led(1)
                await asyncio.sleep_ms(5000)

我提供了兩種閃爍模式。 如果連接丟失(或剛剛建立),設備將快速閃爍。 如果建立連接,設備每 2 秒閃爍一次。 如果需要,可以在這裡實現其他閃爍模式。

但LED只是縱容而已。 我們還瞄準了顯示器。

    async def _display_coro(self):
        display = SSD1306_I2C(128,32, i2c)
    
        while True:
            display.poweron()
            display.fill(0)
            display.text("COLD: {:.3f}".format(self.cold_counter.value() / 1000), 16, 4)
            display.text("HOT:  {:.3f}".format(self.hot_counter.value() / 1000), 16, 20)
            display.show()
            await asyncio.sleep(3)
            display.poweroff()

            while self.button():
                await asyncio.sleep_ms(20)

這就是我所說的——使用協程是多麼簡單和方便。 這個小功能描述了整個用戶體驗。 協程只需等待按鈕按下並打開顯示屏 3 秒鐘。 顯示屏顯示當前儀表讀數。

還剩下一些小事情。 這是(重新)啟動整個企業的函數。 主循環只是每分鐘發送一次各種調試信息。 總的來說,我照原樣引用——我認為沒有必要評論太多

   async def main(self):
        while True:
            try:
                await self._connect_to_WiFi()
                await self._run_main_loop()
                    
            except Exception as e:
                self.dprint('Global communication failure: ', e)
                await asyncio.sleep(20)

    async def _connect_to_WiFi(self):
        self.dprint('Connecting to WiFi and MQTT')
        sta_if = network.WLAN(network.STA_IF)
        sta_if.connect(config['ssid'], config['wifi_pw'])
        
        conn = False
        while not conn:
            await self.connect()
            conn = True

        self.dprint('Connected!')
        self.internet_outage = False

    async def _run_main_loop(self):
        # Loop forever
        mins = 0
        while True:
            gc.collect()  # For RAM stats.
            mem_free = gc.mem_free()
            mem_alloc = gc.mem_alloc()

            try:
                await self.publish_debug_msg("Uptime", mins)
                await self.publish_debug_msg("Repubs", self.REPUB_COUNT)
                await self.publish_debug_msg("Outages", self.internet_outages)
                await self.publish_debug_msg("MemFree", mem_free)
                await self.publish_debug_msg("MemAlloc", mem_alloc)
            except Exception as e:
                self.dprint("Exception occurred: ", e)
            mins += 1

            await asyncio.sleep(60)

好吧,還有一些設置和常量來完成描述

#####################################
# Constants and configuration
#####################################


config['keepalive'] = 60
config['clean'] = False
config['will'] = ('/ESP/Wemos/Water/LastWill', 'Goodbye cruel world!', False, 0)

MQTTClient.DEBUG = True

EEPROM_ADDR_HOT_VALUE = const(0)
EEPROM_ADDR_COLD_VALUE = const(4)

一切都是這樣開始的

client = CounterMQTTClient()
loop = asyncio.get_event_loop()
loop.run_until_complete(client.main())

我的記憶發生了一些事情

所以,所有的代碼都在那裡。 我使用 ampy 實用程序上傳文件 - 它允許您將它們上傳到內部(ESP-07 本身中的閃存驅動器),然後從程序中將其作為常規文件訪問。 我還上傳了我使用的 mqtt_as、uasyncio、ssd1306 和集合庫(在 mqtt_as 內部使用)。

我們啟動並...我們得到一個內存錯誤。 此外,我越是嘗試了解內存洩漏的確切位置,放置的調試打印越多,此錯誤出現的越早。 通過 Google 進行簡短搜索後,我了解到微控制器原則上只有 30 kB 內存,根本無法容納 65 kB 代碼(包括庫)。

但還有一條出路。 事實證明,micropython 並不直接從 .py 文件執行代碼 - 該文件首先被編譯。 而且,它直接在微控制器上編譯,轉換為字節碼,然後存儲在內存中。 那麼,為了讓編譯器工作,您還需要一定量的 RAM。

訣竅是使微控制器免於資源密集型編譯。 您可以在大型計算機上編譯文件並將現成的字節碼上傳到微控制器中。 為此,您需要下載 micropython 固件並構建 mpy-cross 實用程序.

我沒有編寫 Makefile,而是手動檢查並編譯了所有必要的文件(包括庫),如下所示

mpy-cross water_counter.py

剩下的就是上傳擴展名為 .mpy 的文件,不要忘記首先從設備的文件系統中刪除相應的 .py。

我在程序(IDE?)ESPlorer 中完成了所有開發。 它允許您將腳本上傳到微控制器並立即執行它們。 就我而言,所有對象的所有邏輯和創建都位於 water_counter.py (.mpy) 文件中。 但為了讓這一切自動啟動,開頭還必須有一個名為 main.py 的文件。 而且,它應該是完全的.py,而不是預編譯的.mpy。 這是它的瑣碎內容

import water_counter

我們啟動它 - 一切正常。 但可用內存小得驚人——大約 1kb。 我仍然計劃擴展設備的功能,而這個千字節顯然對我來說不夠。 但事實證明,這個案子也有出路。

事情是這樣的。 儘管這些文件被編譯為字節碼並駐留在內部文件系統上,但實際上它們仍然被加載到 RAM 中並從那裡執行。 但事實證明,micropython 可以直接從閃存執行字節碼,但為此你需要將其直接構建到固件中。 這並不困難,儘管在我的上網本上花了相當長的時間(只是我碰巧有 Linux)。

算法是這樣的:

  • Скачатьиустановить ESP開放SDK。 這個東西為 ESP8266 的程序組裝了編譯器和庫。 按照項目主頁上的說明組裝(我選擇了STANDALONE=yes設置)
  • 下載 微型蟒蛇排序
  • 將所需的庫放置在 micropython 樹內的 ports/esp8266/modules 中
  • 我們按照文件中的說明組裝固件 端口/esp8266/README.md
  • 我們將固件上傳到微控制器(我在 Windows 上使用 ESP8266Flasher 程序或 Python esptool 執行此操作)

就是這樣,現在“導入 ssd1306”將直接從固件中提取代碼,並且不會為此消耗 RAM。 通過這個技巧,我只將庫代碼上傳到固件中,而主程序代碼從文件系統執行。 這使您可以輕鬆修改程序,而無需重新編譯固件。 目前我有大約 8.5kb 的可用 RAM。 這將使我們能夠在未來實現許多不同的有用功能。 好吧,如果內存根本不夠,那麼你可以將主程序推送到固件中。

那麼我們現在應該做什麼呢?

好的,硬件被焊接,固件被寫入,盒子被打印,設備被粘在牆上,愉快地閃爍著一個燈泡。 但目前它還是一個黑匣子(無論是字面意思還是像徵意義),而且仍然沒什麼用處。 是時候對發送到服務器的 MQTT 消息執行某些操作了。

我的“智能家居”正在運轉 管家製度。 MQTT 模塊要么是開箱即用的,要么可以從附加市場輕鬆安裝 - 我不記得從哪裡得到它。 MQTT 不是一個自給自足的東西——你需要一個所謂的。 代理 - 接收、排序並將 MQTT 消息轉發給客戶端的服務器。 我使用 mosquitto,它(像 Majordomo)在同一台上網本上運行。

設備發送至少一次消息後,該值將立即出現在列表中。

將水錶連接到智能家居

這些值現在可以與系統對象關聯,它們可以在自動化腳本中使用並進行各種分析 - 所有這些都超出了本文的範圍。 我可以向任何感興趣的人推薦majordomo系統 鏡頭中的通道電子器件 — 一位朋友也在建造一個智能家居,並且清楚地談論瞭如何設置該系統。

我將向您展示幾張圖表。 這是每日價值的簡單圖表

將水錶連接到智能家居
可見晚上幾乎沒有人用水。 有幾次有人去廁所,反滲透過濾器似乎每晚會吸幾升水。 早上,消耗明顯增加。 我通常使用鍋爐裡的水,但後來我想洗澡,就暫時改用城市熱水——這在底部圖表中也清晰可見。

從這張圖中我了解到,上廁所需要6-7升水,洗澡需要20-30升水,洗碗大約需要20升水,洗澡需要160升水。 我的家人每天消耗大約 500-600 升。

對於特別好奇的人,可以查看每個單獨值的記錄

將水錶連接到智能家居

從這裡我了解到,當水龍頭打開時,水以大約每1秒5升的速度流動。

但這種形式的統計數據可能不太方便查看。 Majordomo 還可以按天、按周和按月查看消耗圖表。 例如,這是一個以條形表示的消耗圖

將水錶連接到智能家居

到目前為止我只有一周的數據。 一個月後,該圖表將更具指示性 - 每天都會有一個單獨的列。 由於對我手動輸入的值(最大的列)進行的調整,圖片略有損壞。 目前還不清楚我是否錯誤地設置了第一個值,幾乎少了一個立方體,或者這是否是固件中的錯誤,並且沒有計算所有升數。 需要更多時間。

圖表本身仍然需要一些魔法、粉飾、繪畫。 也許我還會構建一個內存消耗圖以用於調試目的 - 以防出現洩漏。 也許我會以某種方式展示沒有互聯網的時期。 目前,這一切都還停留在想法層面。

結論

今天我的公寓變得更加智能了。 有了這麼小的設備,我監控家裡的用水量會更方便。 如果說早些時候我對“我們一個月又消耗了很多水”感到憤怒,那麼現在我可以找到這種消耗的來源。

如果屏幕上的讀數距離儀表本身一米,有些人可能會覺得很奇怪。 但在不久的將來,我計劃搬到另一間公寓,那裡會有幾個立管,而且儀表本身很可能位於樓梯平台上。 所以遠程抄表設備將會非常有用。

我還計劃擴展該設備的功能。 我已經在研究電動閥門了。 現在,要將鍋爐切換為城市用水,我需要在一個難以觸及的位置轉動 3 個水龍頭。 如果用一個帶有相應指示的按鈕來完成此操作會方便得多。 當然,採取防洩漏保護措施是值得的。

在文章中,我描述了基於 ESP8266 的設備版本。 在我看來,我使用協程想出了一個非常有趣的 micropython 固件版本 - 簡單又漂亮。 我試圖描述我在競選期間遇到的許多細微差別和缺點。 也許我把所有事情描述得太詳細了;就我個人而言,作為一名讀者,我更容易跳過不必要的內容,而不是稍後再思考未說的內容。

一如既往,我願意接受建設性的批評。

源代碼
電路及闆卡
案例模型

來源: www.habr.com

添加評論