File size: 4,497 Bytes
8a8a771
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
import json
import time
import random
import threading
from datetime import datetime
from typing import Dict, Optional
import numpy as np

class FitnessDataSimulator:
    def __init__(self, config):
        self.config = config
        self.running = False
        self.thread = None
        
        # Base patterns for realistic simulation
        self.sleep_pattern = {"base": 7.5, "variance": 1.5}
        self.steps_pattern = {"base": 8000, "variance": 3000}
        self.mood_pattern = {"base": 7, "variance": 2}
        self.calories_pattern = {"base": 400, "variance": 200}
        self.water_pattern = {"base": 6, "variance": 2}
        
        # Daily progression
        self.daily_progression = 0
    
    def generate_realistic_data(self) -> Dict:
        """Generate realistic fitness data with daily patterns"""
        hour = datetime.now().hour
        
        # Adjust patterns based on time of day
        steps_multiplier = self._get_activity_multiplier(hour)
        mood_adjustment = self._get_mood_adjustment(hour)
        
        data = {
            "timestamp": datetime.now().isoformat(),
            "sleep_hours": max(4, min(12, 
                np.random.normal(self.sleep_pattern["base"], self.sleep_pattern["variance"])
            )),
            "steps": max(0, int(
                np.random.normal(self.steps_pattern["base"] * steps_multiplier, self.steps_pattern["variance"])
            )),
            "mood_score": max(1, min(10, 
                np.random.normal(self.mood_pattern["base"] + mood_adjustment, self.mood_pattern["variance"])
            )),
            "calories_burned": max(0, int(
                np.random.normal(self.calories_pattern["base"] * steps_multiplier, self.calories_pattern["variance"])
            )),
            "water_intake": max(0, 
                np.random.normal(self.water_pattern["base"], self.water_pattern["variance"])
            ),
            "heart_rate": random.randint(60, 100),
            "active_minutes": random.randint(20, 120)
        }
        
        return data
    
    def _get_activity_multiplier(self, hour: int) -> float:
        """Get activity multiplier based on hour of day"""
        if 6 <= hour <= 9:  # Morning
            return 1.2
        elif 12 <= hour <= 14:  # Lunch
            return 1.1
        elif 17 <= hour <= 19:  # Evening
            return 1.3
        elif 22 <= hour or hour <= 5:  # Night
            return 0.3
        else:
            return 1.0
    
    def _get_mood_adjustment(self, hour: int) -> float:
        """Get mood adjustment based on hour of day"""
        if 7 <= hour <= 11:  # Morning
            return 0.5
        elif 14 <= hour <= 16:  # Afternoon dip
            return -0.3
        elif 18 <= hour <= 21:  # Evening
            return 0.3
        else:
            return 0
    
    def save_data(self, data: Dict):
        """Save data to JSON file"""
        try:
            with open(self.config.FIT_STREAM_FILE, 'w') as f:
                json.dump(data, f, indent=2)
        except Exception as e:
            print(f"Error saving data: {e}")
    
    def _simulate_loop(self):
        """Main simulation loop"""
        while self.running:
            try:
                data = self.generate_realistic_data()
                self.save_data(data)
                time.sleep(5)  # Update every 5 seconds
            except Exception as e:
                print(f"Simulation error: {e}")
                time.sleep(5)
    
    def start_simulation(self):
        """Start the data simulation"""
        if not self.running:
            self.running = True
            self.thread = threading.Thread(target=self._simulate_loop, daemon=True)
            self.thread.start()
            print("✅ Fitness data simulation started")
    
    def stop_simulation(self):
        """Stop the data simulation"""
        self.running = False
        if self.thread:
            self.thread.join()
        print("🛑 Fitness data simulation stopped")
    
    def get_latest_data(self) -> Optional[Dict]:
        """Get the latest fitness data"""
        try:
            with open(self.config.FIT_STREAM_FILE, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return None
        except Exception as e:
            print(f"Error reading data: {e}")
            return None