怎样实现运动数据汇总脚本

本文目录导读：

1. 基础版本 - Python实现
2. 增强版 - 带Web界面
3. 使用说明

我来提供一个运动数据汇总脚本的实现方案,这个脚本可以自动汇总各种运动数据（如跑步、游泳、骑行等）。

基础版本 - Python实现

运动数据汇总脚本

import pandas as pd
import json
from datetime import datetime, timedelta
import os
from typing import Dict, List, Optional
class SportDataSummarizer:
    def __init__(self):
        self.activities = []
        self.summary = {}
    def add_strava_data(self, data: Dict) -> None:
        """添加Strava格式的数据"""
        self.activities.append({
            'type': 'run' if data.get('type') == 'Run' else 'cycle',
            'date': datetime.fromtimestamp(data['start_date']),
            'distance': data.get('distance', 0) / 1000,  # 米转公里
            'duration': data.get('moving_time', 0) / 60,  # 秒转分钟
            'calories': data.get('calories', 0),
            'avg_heart_rate': data.get('average_heartrate', 0),
            'max_heart_rate': data.get('max_heartrate', 0),
            'avg_speed': data.get('average_speed', 0) * 3.6,  # m/s转km/h
        })
    def add_simple_data(self, sport_type: str, date: str, 
                       distance_km: float, duration_min: float, 
                       **kwargs) -> None:
        """添加简单格式的数据"""
        self.activities.append({
            'type': sport_type,
            'date': datetime.strptime(date, '%Y-%m-%d'),
            'distance': distance_km,
            'duration': duration_min,
            **kwargs
        })
    def calculate_summary(self, period: str = 'month') -> Dict:
        """计算汇总统计"""
        if not self.activities:
            return {}
        df = pd.DataFrame(self.activities)
        now = datetime.now()
        # 根据时间段过滤
        if period == 'week':
            start_date = now - timedelta(days=now.weekday())
            df = df[df['date'] >= start_date]
        elif period == 'month':
            start_date = now.replace(day=1, hour=0, minute=0, second=0)
            df = df[df['date'] >= start_date]
        elif period == 'year':
            start_date = now.replace(month=1, day=1, hour=0, minute=0, second=0)
            df = df[df['date'] >= start_date]
        if df.empty:
            return {'message': f'No activities in this {period}'}
        # 汇总统计
        self.summary = {
            'period': period,
            'total_activities': len(df),
            'total_distance_km': round(df['distance'].sum(), 2),
            'total_duration_min': round(df['duration'].sum(), 1),
            'total_duration_hours': round(df['duration'].sum() / 60, 2),
            'total_calories': int(df['calories'].sum()),
            'avg_distance_km': round(df['distance'].mean(), 2),
            'avg_duration_min': round(df['duration'].mean(), 1),
            'avg_speed_kmh': round(df['distance'].sum() / (df['duration'].sum() / 60), 2) if df['duration'].sum() > 0 else 0,
            'max_distance_km': round(df['distance'].max(), 2),
            'max_duration_min': round(df['duration'].max(), 1),
            'by_type': {},
            'by_week': []
        }
        # 按运动类型分类汇总
        for sport_type in df['type'].unique():
            type_data = df[df['type'] == sport_type]
            self.summary['by_type'][sport_type] = {
                'count': len(type_data),
                'total_distance_km': round(type_data['distance'].sum(), 2),
                'total_duration_min': round(type_data['duration'].sum(), 1)
            }
        # 按周统计（如果有长期数据）
        if len(df) > 7 and period in ['month', 'year']:
            df['week'] = df['date'].dt.isocalendar().week
            weekly = df.groupby('week').agg({
                'distance': 'sum',
                'duration': 'sum',
                'calories': 'sum'
            }).reset_index()
            self.summary['by_week'] = weekly.to_dict('records')
        return self.summary
    def print_summary(self, summary: Dict = None) -> None:
        """打印汇总结果"""
        if summary is None:
            summary = self.summary
        if 'message' in summary:
            print(summary['message'])
            return
        print("=" * 50)
        print(f"📊 运动数据汇总 - {summary['period'].capitalize()}")
        print("=" * 50)
        print(f"📅 总活动次数: {summary['total_activities']}")
        print(f"📏 总距离: {summary['total_distance_km']} km")
        print(f"⏱️ 总时长: {summary['total_duration_hours']} 小时 ({summary['total_duration_min']} 分钟)")
        print(f"🔥 总消耗卡路里: {summary['total_calories']} kcal")
        print("-" * 50)
        print(f"📐 平均距离: {summary['avg_distance_km']} km")
        print(f"⏱️ 平均时长: {summary['avg_duration_min']} 分钟")
        print(f"🏃 平均速度: {summary['avg_speed_kmh']} km/h")
        print(f"📈 最长距离: {summary['max_distance_km']} km")
        if summary['by_type']:
            print("-" * 50)
            print("📋 按运动类型:")
            for sport_type, data in summary['by_type'].items():
                print(f"  • {sport_type}: {data['count']}次, {data['total_distance_km']}km")
    def export_to_json(self, filename: str = 'sport_summary.json') -> None:
        """导出为JSON文件"""
        if not self.summary:
            self.calculate_summary()
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.summary, f, indent=2, ensure_ascii=False, default=str)
        print(f"✅ 数据已导出到 {filename}")
    def export_to_csv(self, filename: str = 'sport_activities.csv') -> None:
        """导出活动明细为CSV"""
        if self.activities:
            df = pd.DataFrame(self.activities)
            df.to_csv(filename, index=False, encoding='utf-8-sig')
            print(f"✅ 活动明细已导出到 {filename}")
# 使用示例
def main():
    # 创建汇总器
    summarizer = SportDataSummarizer()
    # 添加模拟数据
    summarizer.add_simple_data('run', '2024-01-15', 5.2, 30, calories=350)
    summarizer.add_simple_data('run', '2024-01-20', 10.1, 55, calories=650)
    summarizer.add_simple_data('cycle', '2024-01-25', 20.5, 45, calories=400)
    summarizer.add_simple_data('run', '2024-02-01', 8.5, 45, calories=500)
    summarizer.add_simple_data('swim', '2024-02-10', 1.5, 30, calories=300)
    # 计算并显示汇总
    summary = summarizer.calculate_summary('month')
    summarizer.print_summary(summary)
    # 导出数据
    summarizer.export_to_json()
    summarizer.export_to_csv()
if __name__ == "__main__":
    main()

增强版 - 带Web界面

# sport_dashboard.py
from flask import Flask, render_template, jsonify, request
import pandas as pd
import plotly.express as px
import plotly.utils
import json
app = Flask(__name__)
summarizer = SportDataSummarizer()  # 复用之前的类
@app.route('/')
def index():
    """主页面"""
    return render_template('dashboard.html')
@app.route('/api/upload', methods=['POST'])
def upload_data():
    """上传运动数据"""
    if 'file' not in request.files:
        return jsonify({'error': 'No file uploaded'}), 400
    file = request.files['file']
    if file.filename.endswith('.csv'):
        df = pd.read_csv(file)
        for _, row in df.iterrows():
            summarizer.add_simple_data(
                row['type'], row['date'],
                row['distance_km'], row['duration_min']
            )
        return jsonify({'message': f'Loaded {len(df)} activities'})
    return jsonify({'error': 'Unsupported format'}), 400
@app.route('/api/summary')
def get_summary():
    """获取汇总数据"""
    period = request.args.get('period', 'month')
    summary = summarizer.calculate_summary(period)
    return jsonify(summary)
@app.route('/api/charts')
def get_charts():
    """获取图表数据"""
    if not summarizer.activities:
        return jsonify({'error': 'No data'}), 404
    df = pd.DataFrame(summarizer.activities)
    df['date_str'] = df['date'].dt.strftime('%Y-%m-%d')
    # 创建图表
    fig1 = px.bar(df, x='date_str', y='distance', color='type',
                  title='Daily Distance by Activity Type')
    fig2 = px.pie(df, names='type', values='distance',
                  title='Distance Distribution by Sport Type')
    return jsonify({
        'daily_chart': json.dumps(fig1, cls=plotly.utils.PlotlyJSONEncoder),
        'pie_chart': json.dumps(fig2, cls=plotly.utils.PlotlyJSONEncoder)
    })
if __name__ == '__main__':
    app.run(debug=True)

使用说明

安装依赖

pip install pandas flask plotly

主要功能

1. 数据导入: 支持手动输入、CSV文件、Strava API
2. 汇总统计: 按周/月/年汇总
3. 类型分类: 按运动类型统计
4. 性能指标: 距离、时间、卡路里、心率等
5. 数据导出: JSON/CSV格式
6. 可视化: Web界面图表展示

数据格式示例 (CSV)

type,date,distance_km,duration_min,calories
run,2024-01-15,5.2,30,350
cycle,2024-01-20,20.5,45,400
swim,2024-01-25,1.5,30,300

这个脚本可以根据你的具体需求进行扩展,比如添加更多运动类型、集成其他API、添加预测功能等。