Gobidev/GadgetbridgeMqtt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adjust daily steps query to start at 4am

Browse Source
This commit is contained in:
Oliver Großkloß 2025-11-21 00:25:06 +01:00
parent 43930562cb
commit 4db6a97733
1 changed files with 80 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
main.py
View File

@ -56,6 +56,20 @@ class GadgetbridgeMQTTPublisher:
"password": os.getenv("MQTT_PASSWORD", ""),
}
def get_day_start_timestamp(self) -> int:
"""Get the timestamp for the start of the current day (4am)"""
now = datetime.now()
today = now.date()
# Day starts at 4am
day_start_time = datetime.combine(today, datetime.min.time()).replace(hour=4)
# If current time is before 4am, we're still in "yesterday's" day
if now.hour < 4:
day_start_time -= timedelta(days=1)
return int(day_start_time.timestamp())
def initialize_sensors(self):
"""Initialize sensor definitions after device_name is available"""
self.sensors = [
@ -166,6 +180,14 @@ class GadgetbridgeMQTTPublisher:
"state_class": "measurement",
"query": self.query_total_sleep_duration,
},
{
"name": "Server Time",
"unique_id": "server_time",
"state_topic": f"gadgetbridge/{self.device_name}/server_time",
"icon": "mdi:clock-outline",
"device_class": "timestamp",
"query": self.query_server_time,
},
]
async def publish_home_assistant_discovery(
@ -211,36 +233,58 @@ class GadgetbridgeMQTTPublisher:
)
def query_daily_steps(self, cursor) -> Any:
today = datetime.now().date()
today_start = int(datetime.combine(today, datetime.min.time()).timestamp())
today_end = int(datetime.combine(today, datetime.max.time()).timestamp())
day_start_ts = self.get_day_start_timestamp()
now_ts = int(datetime.now().timestamp())
cursor.execute(
"SELECT SUM(STEPS) FROM XIAOMI_ACTIVITY_SAMPLE WHERE TIMESTAMP >= ? AND TIMESTAMP <= ?",
(today_start, today_end),
(day_start_ts, now_ts),
)
return cursor.fetchone()[0] or 0
def query_weekly_steps(self, cursor) -> Any:
today = datetime.now().date()
now = datetime.now()
today = now.date()
# Week starts on Monday at 4am
week_start = today - timedelta(days=today.weekday())
week_start_ts = int(
datetime.combine(week_start, datetime.min.time()).timestamp()
)
week_start_time = datetime.combine(week_start, datetime.min.time()).replace(hour=4)
# If we're before 4am on Monday, the week actually started last Monday
if today.weekday() == 0 and now.hour < 4:
week_start_time -= timedelta(days=7)
week_start_ts = int(week_start_time.timestamp())
now_ts = int(now.timestamp())
cursor.execute(
"SELECT SUM(STEPS) FROM XIAOMI_ACTIVITY_SAMPLE WHERE TIMESTAMP >= ?",
(week_start_ts,),
"SELECT SUM(STEPS) FROM XIAOMI_ACTIVITY_SAMPLE WHERE TIMESTAMP >= ? AND TIMESTAMP <= ?",
(week_start_ts, now_ts),
)
return cursor.fetchone()[0] or 0
def query_monthly_steps(self, cursor) -> Any:
today = datetime.now().date()
now = datetime.now()
today = now.date()
# Month starts on 1st at 4am
month_start = today.replace(day=1)
month_start_ts = int(
datetime.combine(month_start, datetime.min.time()).timestamp()
)
month_start_time = datetime.combine(month_start, datetime.min.time()).replace(hour=4)
# If we're before 4am on the 1st, the month actually started last month's 1st
if today.day == 1 and now.hour < 4:
# Go back to previous month
if month_start.month == 1:
month_start_time = month_start_time.replace(year=month_start.year - 1, month=12)
else:
month_start_time = month_start_time.replace(month=month_start.month - 1)
month_start_ts = int(month_start_time.timestamp())
now_ts = int(now.timestamp())
cursor.execute(
"SELECT SUM(STEPS) FROM XIAOMI_ACTIVITY_SAMPLE WHERE TIMESTAMP >= ?",
(month_start_ts,),
"SELECT SUM(STEPS) FROM XIAOMI_ACTIVITY_SAMPLE WHERE TIMESTAMP >= ? AND TIMESTAMP <= ?",
(month_start_ts, now_ts),
)
return cursor.fetchone()[0] or 0
@ -266,29 +310,41 @@ class GadgetbridgeMQTTPublisher:
return row[0] if row else None
def query_hr_resting(self, cursor) -> Any:
day_start_ts = self.get_day_start_timestamp()
cursor.execute(
"SELECT HR_RESTING FROM XIAOMI_DAILY_SUMMARY_SAMPLE ORDER BY TIMESTAMP DESC LIMIT 1"
"SELECT HR_RESTING FROM XIAOMI_DAILY_SUMMARY_SAMPLE WHERE TIMESTAMP >= ? ORDER BY TIMESTAMP DESC LIMIT 1",
(day_start_ts,)
)
row = cursor.fetchone()
return row[0] if row else None
def query_hr_max(self, cursor) -> Any:
day_start_ts = self.get_day_start_timestamp()
cursor.execute(
"SELECT HR_MAX FROM XIAOMI_DAILY_SUMMARY_SAMPLE ORDER BY TIMESTAMP DESC LIMIT 1"
"SELECT HR_MAX FROM XIAOMI_DAILY_SUMMARY_SAMPLE WHERE TIMESTAMP >= ? ORDER BY TIMESTAMP DESC LIMIT 1",
(day_start_ts,)
)
row = cursor.fetchone()
return row[0] if row else None
def query_hr_avg(self, cursor) -> Any:
day_start_ts = self.get_day_start_timestamp()
cursor.execute(
"SELECT HR_AVG FROM XIAOMI_DAILY_SUMMARY_SAMPLE ORDER BY TIMESTAMP DESC LIMIT 1"
"SELECT HR_AVG FROM XIAOMI_DAILY_SUMMARY_SAMPLE WHERE TIMESTAMP >= ? ORDER BY TIMESTAMP DESC LIMIT 1",
(day_start_ts,)
)
row = cursor.fetchone()
return row[0] if row else None
def query_calories(self, cursor) -> Any:
day_start_ts = self.get_day_start_timestamp()
cursor.execute(
"SELECT CALORIES FROM XIAOMI_DAILY_SUMMARY_SAMPLE ORDER BY TIMESTAMP DESC LIMIT 1"
"SELECT CALORIES FROM XIAOMI_DAILY_SUMMARY_SAMPLE WHERE TIMESTAMP >= ? ORDER BY TIMESTAMP DESC LIMIT 1",
(day_start_ts,)
)
row = cursor.fetchone()
return row[0] if row else None
@ -309,6 +365,10 @@ class GadgetbridgeMQTTPublisher:
# Convert minutes to hours, round to 2 decimals
return round(row[0] / 60, 2) if row and row[0] is not None else None
def query_server_time(self, cursor) -> Any:
"""Return current server time in ISO 8601 format for Home Assistant timestamp"""
return datetime.now().isoformat()
def get_sensor_data(self) -> Dict[str, Any]:
"""Query all sensors and return their values as a dict"""
if not os.path.exists(self.db_path):