#!/usr/bin/env python3
import os
import re
from pathlib import Path
from datetime import datetime
from getpass import getpass
import logging
import garth
from garth.exc import GarthException, GarthHTTPError
from fit_tool.fit_file import FitFile
from fit_tool.fit_file_builder import FitFileBuilder
from fit_tool.profile.messages.record_message import RecordMessage, RecordTemperatureField
from fit_tool.profile.messages.session_message import SessionMessage
from fit_tool.profile.messages.lap_message import LapMessage
from fit_tool.profile.messages.file_id_message import FileIdMessage
# === Logging ===
SCRIPT_DIR = Path(__file__).resolve().parent
log_file_path = SCRIPT_DIR / "myWhoosh2Garmin.log"
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
file_handler = logging.FileHandler(log_file_path)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# === Pfade ===
MYWHOOSH_PATH = Path("Download-Path")
BACKUP_PATH = MYWHOOSH_PATH / "backup"
BACKUP_PATH.mkdir(exist_ok=True)
# === Garmin Auth ===
TOKENS_PATH = SCRIPT_DIR / '.garth'
def authenticate_to_garmin():
try:
if TOKENS_PATH.exists():
garth.resume(TOKENS_PATH)
else:
username = input("Garmin Benutzername: ")
password = getpass("Garmin Passwort: ")
garth.login(username, password)
garth.save(TOKENS_PATH)
logger.info(f"Authenticated as: {garth.client.username}")
except GarthException as e:
logger.error(f"Garmin Auth Fehler: {e}")
exit(1)
# === Hilfsfunktionen ===
def calculate_avg(values):
return round(sum(values)/len(values),1) if values else 0
def append_value(values, message, field_name):
val = getattr(message, field_name, None)
if val is not None:
values.append(val)
def reset_values():
return [], [], []
def get_most_recent_fit_file(folder: Path) -> Path:
fit_files = list(folder.glob("*.fit"))
if not fit_files:
return None
return max(fit_files, key=os.path.getctime)
def generate_new_filename(fit_file: Path) -> Path:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return BACKUP_PATH / f"{fit_file.stem}_{timestamp}.fit"
# === Hauptlogik ===
def cleanup_fit_file(fit_file_path: Path, new_file_path: Path):
builder = FitFileBuilder()
fit_file = FitFile.from_file(str(fit_file_path))
cadence_values, power_values, hr_values = reset_values()
for record in fit_file.records:
msg = record.message
if isinstance(msg, FileIdMessage):
# Override manufacturer/product but keep other fields
msg.manufacturer = 1
msg.product = 1836
if isinstance(msg, LapMessage):
continue
if isinstance(msg, RecordMessage):
msg.remove_field(RecordTemperatureField.ID)
append_value(cadence_values, msg, "cadence")
append_value(power_values, msg, "power")
append_value(hr_values, msg, "heart_rate")
if isinstance(msg, SessionMessage):
if not msg.avg_cadence:
msg.avg_cadence = calculate_avg(cadence_values)
if not msg.avg_power:
msg.avg_power = calculate_avg(power_values)
if not msg.avg_heart_rate:
msg.avg_heart_rate = calculate_avg(hr_values)
cadence_values, power_values, hr_values = reset_values()
builder.add(msg)
# Datei bauen und speichern
builder.build().to_file(str(new_file_path))
logger.info(f"Bereinigte Datei gespeichert: {new_file_path}")
# === Main ===
def main():
authenticate_to_garmin()
latest_fit = get_most_recent_fit_file(MYWHOOSH_PATH)
if not latest_fit:
logger.error(f"Keine FIT-Dateien in {MYWHOOSH_PATH} gefunden.")
exit(1)
logger.info(f"Neueste Datei: {latest_fit}")
new_file = generate_new_filename(latest_fit)
try:
cleanup_fit_file(latest_fit, new_file)
except Exception as e:
logger.error(f"Fehler beim Bereinigen: {e}")
exit(1)
# Upload zu Garmin
try:
with open(new_file, "rb") as f:
garth.client.upload(f)
logger.info(f"Erfolgreich hochgeladen: {new_file}")
except GarthHTTPError:
logger.info("Aktivität möglicherweise schon bei Garmin Connect vorhanden.")
if __name__ == "__main__":
main()