本文目录导读：
我来提供一个运动数据汇总脚本的实现方案,这个脚本可以自动汇总各种运动数据（如跑步、游泳、骑行等）。
基础版本 - 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
主要功能
- 数据导入: 支持手动输入、CSV文件、Strava API
- 汇总统计: 按周/月/年汇总
- 类型分类: 按运动类型统计
- 性能指标: 距离、时间、卡路里、心率等
- 数据导出: JSON/CSV格式
- 可视化: 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、添加预测功能等。
