Day 18 - Event Bus - 讓 IDD Status Changed 優雅地執行吧

17th鐵人賽

JenLiu

2025-08-18 19:52:57

144 瀏覽

分享至

截至目前為止，IddStatusChanged 類別還遠算不上真正意義的完成，因為它的職責是將有變化的狀態主動傳送給 GATT Client，但還沒有任何程式碼對保存 IDS 狀態的 Config.idd_status_changed_flags 有任何干涉，所以第一步是先提供方法來改變 Config.idd_status_changed_flags。

1. 新增與重設狀態

IDD Status Changed 的 Flags 欄位的每個位元所代表的涵義如下：

Flags fit Bit	Definition	Description
0	Therapy Control State Changed	If this bit is set, the therapy control state of the Insulin Delivery Device changed.
1	Operational State Changed	If this bit is set, the operational state of the Insulin Delivery Device changed.
2	Reservoir Status Changed	If this bit is set, the status of the insulin reservoir changed (caused by a reservoir change or the delivery of insulin).
3	Annunciation Status Changed	If this bit is set, a new annunciation was created by the Server application.
4	Total Daily Insulin Status Changed	If this bit is set, the total daily insulin amount changed due to a bolus or basal delivery. The bit shall be set at the end of an effective delivery.
5	Active Basal Rate Status Changed	If this bit is set, the current basal rate changed due to a new basal rate value (e.g., caused by a changed basal rate profile, reaching of a time block with another basal rate value or by a TBR).
6	Active Bolus Status Changed	If this bit is set, a new bolus was initiated or the status of current Active Bolus changed.
7	History Event Recorded	If this bit is set, a new event has been recorded in the history.
All other bits	RFU	None

當 IDS server 的狀態改變時，Flags 相應的位元便會被設定。而無論這個位元是由 0 轉 1，或由 1 轉 0，系統都會送出一個 IDD Status Changed 的 Indicate 給 GATT Client。根據這個特性，程式可以這樣設計：

class IddStatusChanged:
    def _add_flags(self, flags: int) -> bool:
        """設定 flags 指定的欄位，若有欄位被改變，則回傳 True；否則 False。"""

        old = self._config.idd_status_changed_flags
        self._config.idd_status_changed_flags |= flags
        return old != self._config.idd_status_changed_flags

有了設定功能後，當然會有對應的重設功能：

def _reset_flags(self, flags: int) -> bool:
    """重設 flags 指定的欄位，若有欄位被改變，則回傳 True；否則 False。"""

    old = self._config.idd_status_changed_flags
    self._config.idd_status_changed_flags &= ~flags
    return old != self._config.idd_status_changed_flags

2. 與其它部件的溝通

_add_flags() 和 _reset_flags() 已經完成了，但問題是其它部件怎麼使用它們呢？會改變狀態的時機是散佈在 IDS 系統的各處，如：

Bolus 和 Basal 等注射系統
每天零時
異常狀態
IDS 系統的各運作流程

要讓其它組件直接呼叫 IddStatusChanged 類別的成員方法嗎？如若目前運作的是支援 E2E-Protection 的 E2EIddStatusChanged 呢？

有許多方法可以解決此問題，比如簡單地，定義兩個全域變數，讓它們儲存相關函數，以後需要改變狀態時，都藉由這些全域變數來呼叫相應函數：

# server.py

class IdsServer:
    def __init__(self):
        ids_var.set_add_status_changed_fn(self._add_status_changed)
        ids_var.set_reset_status_changed_fn(self._reset_status_changed)

# ids_var.py

def set_add_status_changed_fn(fn):
    global fn_add_status_changed
    fn_add_status_changed = fn

def set_reset_status_changed_fn(fn):
    global fn_reset_status_changed
    fn_reset_status_changed = fn

# some_module.py

def some_fn():
    ids_var.fn_add_status_changed(ACTIVE_BOLUS_STATUS_CHANGED)

這樣姑且是不與實際的函數有直接相依，若以後想要使用新的函數，也可以輕易替換。不過，為了更好解耦合與擴充，咱們不使用此模式，而是使用 Event Bus。

3. Event Bus

Event Bus 的基本概念如下：

分為事件、發送者和接收者
事件：表示系統中的動作或狀態改變，比如按鈕被按下
發送者：觸發事件並將其發送到事件匯流排的組件，比如被點擊的按鈕
接收者：訂閱特定事件，並在事件發生時接收與處理的組件，比如按鈕被點擊後要做的事

_subscribers: dict[int, list] = None

def subscribe(event: int, handler):
    global _subscribers

    if not _subscribers:
        _subscribers = {}

    _subscribers.setdefault(event, []).append(handler)

def publish(event: int, *args):
    for handler in _subscribers.get(event, ()):
        handler(*args)

_subscribers 是一個字典，每個項目都是由事件和處理事件的函數的串列組成：

事件 1：[處理程序 1、處理程序 2、…]
事件 2：[處理程序 1、處理程序 2、…]

subscribe() 會由對特定事件感興趣的組件呼叫，將處裡事件的函數做為參數傳入。

publish() 則由觸發事件的組件呼叫，將事件發生時的相關資訊傳入此函數。

4. 使用 Event Bus 完善 IDD Status Changed

首先，咱們先定義 IDD Status Changed 會使用到的 Event Bus 事件：

# args:
# flags: int，欲增加的位元組合，由 IDD Status Changed 的 Flags 所規定。
EVENT_IDD_STATUS_ADDING = 0

# args:
# flags: int，欲重設的位元組合，由 IDD Status Changed 的 Flags 所規定。
EVENT_IDD_STATUS_RESETTING = 1

然後定義 IDD Status Changed 所需的 Flags 位元：

# IDD Status Changed
THERAPY_CONTROL_STATE_CHANGED = 0x0001
OPERATIONAL_STATE_CHANGED = 0x0002
RESERVOIR_STATUS_CHANGED = 0x0004
ANNUNCIATION_STATUS_CHANGED = 0x0008
TOTAL_DAILY_INSULIN_STATUS_CHANGED = 0x0010
ACTIVE_BASAL_RATE_STATUS_CHANGED = 0x0020
ACTIVE_BOLUS_STATUS_CHANGED = 0x0040
HISTORY_EVENT_RECORDED = 0x0080

有了 Event Bus 事件後，就是讓 IddStatusChanged 類別訂閱這些事件：

class IddStatusChanged:
    def __init__(self, config: config.Config):
        common.eventbus.subscribe(
            core.events.EVENT_IDD_STATUS_ADDING, self._on_status_adding
        )
        common.eventbus.subscribe(
            core.events.EVENT_IDD_STATUS_RESETTING, self._on_status_resetting
        )

    def _on_status_adding(self, flags: int):
        if self._add_flags(flags):
            ble.stack.BleTxScheduler().add(
                ble.stack.ACT_INDICATE, self.send_data, self.value_handle, None
            )

    def _on_status_resetting(self, flags: int):
        if self._reset_flags(flags):
            ble.stack.BleTxScheduler().add(
                ble.stack.ACT_INDICATE, self.send_data, self.value_handle, None
            )

咱們讓 IddStatusChanged 類別訂閱事件 EVENT_IDD_STATUS_ADDING 和 EVENT_IDD_STATUS_RESETTING，並且分別指派 _on_status_adding() 和 _on_status_resetting() 來處理事件。

這兩個方法裡只是直接呼叫先前已做好的 _add_flags() 和 _reset_flags()，然後利用第 14 天所做的排程器將資料排進傳送的佇列。

咱們可以稍微改一下第 15 天所做的計時器來試驗：

_timestamp = 0

def _on_timer(timer):
    # 遞增系統計數器，然後由系統安排主執行緒執行。
    global _timestamp
    _timestamp += 1
    micropython.schedule(_on_timestamp_changed, _timestamp)

def _on_timestamp_changed(timestamp: int):
    """此函數只能在主執行緒被呼叫，否則將有同步問題"""

    if timestamp % 10 == 0:
        common.logger.write("Timestamp: " + str(timestamp))

    if timestamp == 10:
        common.eventbus.publish(
            core.events.EVENT_IDD_STATUS_ADDING, core.events.ACTIVE_BOLUS_STATUS_CHANGED
        )

    elif timestamp == 15:
        common.eventbus.publish(
            core.events.EVENT_IDD_STATUS_RESETTING,
            core.events.ACTIVE_BOLUS_STATUS_CHANGED,
        )

當 IDS server 與 nRF Connect app 建立連線，並且正確設定 IDD Status Changed 的 CCCD 後，便可以在系統計數器為 10 和 15 秒時，收到相應的 Indicate。

至此，IDD Status Changed 就算是完成了，但還有一個問題：

若同一時間段，多次送出 EVENT_IDD_STATUS_ADDING 或 EVENT_IDD_STATUS_RESETTING，將會送出多個具相同值的 Indicate。

比如：

if timestamp == 10:
    common.eventbus.publish(
        core.events.EVENT_IDD_STATUS_ADDING, core.events.ACTIVE_BOLUS_STATUS_CHANGED
    )
    common.eventbus.publish(
        core.events.EVENT_IDD_STATUS_ADDING,
        core.events.ACTIVE_BASAL_RATE_STATUS_CHANGED,
    )

那麼將會收到 2 個 Flags 欄位為 0x0060 的 Indicate，這是因為 common.eventbus.publish() 並不是馬上將 Indicate 送出。要解決這問題，可以簡單地設立一個旗標 _scheduled：

class IddStatusChanged:
    def __init__(self, config: config.Config):
        # 因 Status Changed 被送出前可能會被多次設定，以此來避免重複送出相同值。
        self._scheduled = False

    def send_data(self, conn_handle: int | None, value_handle: int, arg):
        successful = super().send_data(conn_handle, value_handle, arg)
        self._scheduled = False
        return successful

    def _on_status_adding(self, flags: int):
        if self._add_flags(flags) and not self._scheduled:
            self._scheduled = True

            ble.stack.BleTxScheduler().add(
                ble.stack.ACT_INDICATE, self.send_data, self.value_handle, None
            )

    def _on_status_resetting(self, flags: int):
        if self._reset_flags(flags) and not self._scheduled:
            self._scheduled = True

            ble.stack.BleTxScheduler().add(
                ble.stack.ACT_INDICATE, self.send_data, self.value_handle, None
            )

不過這設計是否符合需求就不一定了。比如上面例子，也許需求是在第 10 秒時，先送出 ACTIVE_BOLUS_STATUS_CHANGED，然後再送出 ACTIVE_BASAL_RATE_STATUS_CHANGED，共 2 個 Indicate，那麼這設計就不符合要求了。

IDS 規格書並沒有說不能這樣，只是一般為了省電，會將多個 EVENT_IDD_STATUS_ADDING 和 EVENT_IDD_STATUS_RESETTING 合併，但一切還是要依需求為主。