JJY_Toolkit/WaveClock/RTCClockApp.py

from Debug import Debug
from machine import Pin
from machine import RTC
import utime as time
from tm1637 import TM1637
from TimeSource import TimeSource
from TimeSyncer import TimeSyncer
from micropython import schedule
import _thread
from machine import idle

MODE_TIME = 0
MODE_DATE = 1
MODE_WDAY = 2
MODE_SEC  = 3

WEEKDAYS=("non","tuE","uEd","tHu","Fri","SAt","Sun")

class RTCClockApp(Debug):
    """RTCを用いた時計アプリケーションクラス"""

    def __init__(self, display_dev: TM1637, time_source: TimeSource, time_sync: TimeSyncer, mode_select_pin: int, force_sync_pin: int):
        """
        コンストラクタ
        Args:
            display_dev: TM1637のインスタンス
            time_source: TimeSourceのインスタンス
            time_sync: TimeSyncerのインスタンス
            mode_select_pin: 表示モード切替スイッチが接続されているGPIO番号
            force_sync_pin: 強制受信スイッチが接続されているGPIO番号
        """
        self.disp = display_dev
        self.timesrc = time_source
        self.timesync = time_sync
        # 継続フラグ
        self._continue = False
        # 1つ前の秒
        self.prev_sec = -1
        # 初期表示
        self.disp.show_str("----")
        # RTCに現在日時が設定されているか？
        self.setup_rtc = False
        # TimeSourceにコールバック関数を登録
        self.timesrc.add_callback(self.adjust_rtc)
        # 表示モード
        self.mode = MODE_TIME
        # 1つ前の表示モード
        self.prev_mode = self.mode
        # キースイッチ設定
        self._last_keydown_times = {} # キー押下時間を格納する辞書
        self.mode_select = Pin(mode_select_pin, Pin.IN, pull=Pin.PULL_UP)
        self.force_sync = Pin(force_sync_pin, mode=Pin.IN, pull=Pin.PULL_UP)
        self.mode_select.irq(self._key_event_handler,trigger=Pin.IRQ_FALLING)
        self.force_sync.irq(self._key_event_handler,trigger=Pin.IRQ_FALLING)
    
    def _key_event_handler(self, p):
        """スイッチ押下時の割り込みハンドラ"""
        key_id = id(p)
        now = time.ticks_ms()
        last_time = self._last_keydown_times.get(key_id, 0)
        if time.ticks_diff(now, last_time) > 500:    # チャタリング防止: 500ms以上経過していたら
            self._last_keydown_times[key_id] = now
            schedule(self._key_down, p)
    
    def _key_down(self, p):
        """キーダウンイベント処理（非同期コンテキストで実行）"""
        if p == self.mode_select:       # 表示モード選択
            self.mode = (self.mode + 1) % 4
        elif p == self.force_sync:      # 強制受信
            self.timesync.sync_start()

    def adjust_rtc(self, args):
        """
        RTCに現在日時を設定するコールバック関数
        Args:
            args: (jjy_time, ticks_ms)
            jjy_time: RTCに書き込むUNIXエポックタイム
            ticks_ms: この時間を受信したtime.ticks_ms()
        """
        jjy_time, ticks_ms = (args)

        rtc = RTC()
        elapsed_ms = time.ticks_diff(time.ticks_ms(), ticks_ms)
        rtc_time = jjy_time + (elapsed_ms // 1000) + 1
        now = time.localtime(rtc_time)
        # 次の秒が来るまでスリープしてタイミングを合わせる
        time.sleep_ms(1000 - (elapsed_ms % 1000))
        rtc.datetime((now[0], now[1], now[2], now[6], now[3], now[4], now[5], 0))
        self.dprint("Set RTC to %d/%d/%d, %d:%d:%d" % (now[0], now[1], now[2], now[3], now[4], now[5]))

        self.setup_rtc = True
    
    def _update_display(self):
        """
        表示の更新を行うスレッド
        """
        while self._continue:
            time.sleep_ms(50)      # 50ms
            now = time.localtime()
            sec = now[5]
            if self.prev_sec != sec or self.prev_mode != self.mode:
                self.prev_sec = sec
                self.prev_mode = self.mode
                # 表示更新
                disp_str = "----"
                if self.setup_rtc:
                    if self.mode == MODE_TIME:
                        if now[5] % 2 == 0:
                            disp_str = "%2d:%02d" % (now[3],now[4])
                        else:
                            disp_str = "%2d%02d" % (now[3],now[4])
                    elif self.mode == MODE_DATE:
                        disp_str = "%2d%2d" % (now[1], now[2])
                    elif self.mode == MODE_WDAY:
                        disp_str = " %s" % (WEEKDAYS[now[6]])
                    elif self.mode == MODE_SEC:
                        disp_str = "%2d%2d" % (now[4], now[5])
                self.disp.show_str(disp_str)
    
    def run(self):
        """
        アプリケーションメインループ
        """
        self._continue = True   # スレッド起動
        _thread.start_new_thread(self._update_display,())
        try:
            while True:
                idle()
        except Exception as e:   # なにか起きたら停止する
            self.dprint(f"Main loop error: {e}")
        finally:
            self._continue = False