gaitsetpy.dataset

 18class DaphnetLoader(BaseDatasetLoader):
 19    """
 20    Daphnet dataset loader class.
 21    
 22    This class handles loading and processing of the Daphnet dataset for gait analysis.
 23    """
 24    
 25    def __init__(self):
 26        super().__init__(
 27            name="daphnet",
 28            description="Daphnet Freezing of Gait Dataset - Contains accelerometer data from subjects with Parkinson's disease"
 29        )
 30        self.metadata = {
 31            'sensors': ['shank', 'thigh', 'trunk'],
 32            'components': ['h_fd', 'v', 'h_l'],  # horizontal forward, vertical, horizontal lateral
 33            'sampling_frequency': 64,
 34            'annotations': {
 35                0: 'not_valid',
 36                1: 'no_freeze',
 37                2: 'freeze'
 38            }
 39        }
 40    
 41    def load_data(self, data_dir: str, **kwargs) -> Tuple[List[pd.DataFrame], List[str]]:
 42        """
 43        Load Daphnet dataset from the specified directory.
 44        
 45        Args:
 46            data_dir: Directory to store/find the dataset
 47            **kwargs: Additional arguments (unused for Daphnet)
 48            
 49        Returns:
 50            Tuple of (data_list, names_list)
 51        """
 52        # Download and extract if needed
 53        download_dataset("daphnet", data_dir)
 54        extract_dataset("daphnet", data_dir)
 55        
 56        file_path = os.path.join(data_dir, "dataset_fog_release/dataset")
 57        daphnet_data = []
 58        daphnet_names = []
 59        
 60        # Load all subject files
 61        for file in sorted(glob(os.path.join(file_path, "S*.txt"))):
 62            # Extract filename from path
 63            filename = os.path.basename(file)
 64            daphnet_names.append(filename)
 65            
 66            # Load CSV with proper column names
 67            column_names = [
 68                "time", "shank_h_fd", "shank_v", "shank_h_l", 
 69                "thigh_h_fd", "thigh_v", "thigh_h_l", 
 70                "trunk_h_fd", "trunk_v", "trunk_h_l", "annotations"
 71            ]
 72            
 73            df = pd.read_csv(file, sep=" ", names=column_names)
 74            
 75            # Set time as index
 76            df = df.set_index("time")
 77            
 78            # Calculate magnitude for each sensor
 79            df["thigh"] = np.sqrt(df["thigh_h_l"]**2 + df["thigh_v"]**2 + df["thigh_h_fd"]**2)
 80            df["shank"] = np.sqrt(df["shank_h_l"]**2 + df["shank_v"]**2 + df["shank_h_fd"]**2)
 81            df["trunk"] = np.sqrt(df["trunk_h_l"]**2 + df["trunk_v"]**2 + df["trunk_h_fd"]**2)
 82            
 83            # Reorder columns for consistency
 84            df = df[["shank", "shank_h_fd", "shank_v", "shank_h_l", 
 85                    "thigh", "thigh_h_fd", "thigh_v", "thigh_h_l", 
 86                    "trunk", "trunk_h_fd", "trunk_v", "trunk_h_l", "annotations"]]
 87            
 88            daphnet_data.append(df)
 89        
 90        # Store loaded data
 91        self.data = daphnet_data
 92        self.names = daphnet_names
 93        
 94        return daphnet_data, daphnet_names
 95    
 96    def create_sliding_windows(self, data: List[pd.DataFrame], names: List[str], 
 97                             window_size: int = 192, step_size: int = 32) -> List[Dict]:
 98        """
 99        Create sliding windows from the Daphnet dataset.
100        
101        Args:
102            data: List of DataFrames containing Daphnet data
103            names: List of names corresponding to the data
104            window_size: Size of the sliding window (default: 192)
105            step_size: Step size for the sliding window (default: 32)
106            
107        Returns:
108            List of dictionaries containing sliding windows for each DataFrame
109        """
110        windows_data = []
111        
112        for idx, df in enumerate(data):
113            # Filter out invalid data (annotations == 0)
114            df_filtered = df[df.annotations > 0]
115            
116            if df_filtered.empty:
117                continue
118                
119            windows = []
120            processed_columns = set()
121            
122            # Process each sensor column
123            for col in df_filtered.columns:
124                if col != "annotations" and col not in processed_columns:
125                    window_data = sliding_window(df_filtered[col], window_size, step_size)
126                    windows.append({"name": col, "data": window_data})
127                    processed_columns.add(col)
128            
129            # Include annotations separately
130            annotations_window = sliding_window(df_filtered["annotations"], window_size, step_size)
131            windows.append({"name": "annotations", "data": annotations_window})
132            
133            windows_data.append({"name": names[idx], "windows": windows})
134        
135        return windows_data
136    
137    def get_supported_formats(self) -> List[str]:
138        """
139        Get list of supported file formats for Daphnet dataset.
140        
141        Returns:
142            List of supported file extensions
143        """
144        return ['.txt']
145    
146    def get_sensor_info(self) -> Dict[str, List[str]]:
147        """
148        Get information about sensors in the dataset.
149        
150        Returns:
151            Dictionary containing sensor information
152        """
153        return {
154            'sensors': self.metadata['sensors'],
155            'components': self.metadata['components'],
156            'sampling_frequency': self.metadata['sampling_frequency']
157        }
158    
159    def get_annotation_info(self) -> Dict[int, str]:
160        """
161        Get information about annotations in the dataset.
162        
163        Returns:
164            Dictionary mapping annotation values to descriptions
165        """
166        return self.metadata['annotations']

 17class MobiFallLoader(BaseDatasetLoader):
 18    """
 19    MobiFall dataset loader class.
 20    
 21    This class handles loading and processing of the MobiFall dataset for gait analysis.
 22    """
 23    
 24    def __init__(self):
 25        super().__init__(
 26            name="mobifall",
 27            description="MobiFall Dataset - Contains accelerometer and gyroscope data for fall detection"
 28        )
 29        self.metadata = {
 30            'sensors': ['accelerometer', 'gyroscope'],
 31            'components': ['x', 'y', 'z'],
 32            'sampling_frequency': 100,  # Typical for MobiFall
 33            'activities': ['ADL', 'FALL']  # Activities of Daily Living and Falls
 34        }
 35    
 36    def load_data(self, data_dir: str, **kwargs) -> Tuple[List[pd.DataFrame], List[str]]:
 37        """
 38        Load MobiFall dataset from the specified directory.
 39        
 40        Args:
 41            data_dir: Directory to store/find the dataset
 42            **kwargs: Additional arguments (unused for MobiFall)
 43            
 44        Returns:
 45            Tuple of (data_list, names_list)
 46        """
 47        # TODO: Implement MobiFall data loading
 48        # This is a placeholder implementation
 49        print("MobiFall data loading is not yet implemented")
 50        return [], []
 51    
 52    def create_sliding_windows(self, data: List[pd.DataFrame], names: List[str], 
 53                             window_size: int = 192, step_size: int = 32) -> List[Dict]:
 54        """
 55        Create sliding windows from the MobiFall dataset.
 56        
 57        Args:
 58            data: List of DataFrames containing MobiFall data
 59            names: List of names corresponding to the data
 60            window_size: Size of the sliding window (default: 192)
 61            step_size: Step size for the sliding window (default: 32)
 62            
 63        Returns:
 64            List of dictionaries containing sliding windows for each DataFrame
 65        """
 66        # TODO: Implement MobiFall sliding window creation
 67        # This is a placeholder implementation
 68        print("MobiFall sliding window creation is not yet implemented")
 69        return []
 70    
 71    def get_supported_formats(self) -> List[str]:
 72        """
 73        Get list of supported file formats for MobiFall dataset.
 74        
 75        Returns:
 76            List of supported file extensions
 77        """
 78        return ['.csv', '.txt']
 79    
 80    def get_sensor_info(self) -> Dict[str, List[str]]:
 81        """
 82        Get information about sensors in the dataset.
 83        
 84        Returns:
 85            Dictionary containing sensor information
 86        """
 87        return {
 88            'sensors': self.metadata['sensors'],
 89            'components': self.metadata['components'],
 90            'sampling_frequency': self.metadata['sampling_frequency']
 91        }
 92    
 93    def get_activity_info(self) -> List[str]:
 94        """
 95        Get information about activities in the dataset.
 96        
 97        Returns:
 98            List of activity types
 99        """
100        return self.metadata['activities']

 17class ArduousLoader(BaseDatasetLoader):
 18    """
 19    Arduous dataset loader class.
 20    
 21    This class handles loading and processing of the Arduous dataset for gait analysis.
 22    """
 23    
 24    def __init__(self):
 25        super().__init__(
 26            name="arduous",
 27            description="Arduous Dataset - Contains multi-sensor wearable data for daily activity recognition"
 28        )
 29        self.metadata = {
 30            'sensors': ['accelerometer', 'gyroscope', 'magnetometer'],
 31            'components': ['x', 'y', 'z'],
 32            'sampling_frequency': 50,  # Typical for Arduous
 33            'activities': ['walking', 'running', 'sitting', 'standing', 'lying']
 34        }
 35    
 36    def load_data(self, data_dir: str, **kwargs) -> Tuple[List[pd.DataFrame], List[str]]:
 37        """
 38        Load Arduous dataset from the specified directory.
 39        
 40        Args:
 41            data_dir: Directory to store/find the dataset
 42            **kwargs: Additional arguments (unused for Arduous)
 43            
 44        Returns:
 45            Tuple of (data_list, names_list)
 46        """
 47        # TODO: Implement Arduous data loading
 48        # This is a placeholder implementation
 49        print("Arduous data loading is not yet implemented")
 50        return [], []
 51    
 52    def create_sliding_windows(self, data: List[pd.DataFrame], names: List[str], 
 53                             window_size: int = 192, step_size: int = 32) -> List[Dict]:
 54        """
 55        Create sliding windows from the Arduous dataset.
 56        
 57        Args:
 58            data: List of DataFrames containing Arduous data
 59            names: List of names corresponding to the data
 60            window_size: Size of the sliding window (default: 192)
 61            step_size: Step size for the sliding window (default: 32)
 62            
 63        Returns:
 64            List of dictionaries containing sliding windows for each DataFrame
 65        """
 66        # TODO: Implement Arduous sliding window creation
 67        # This is a placeholder implementation
 68        print("Arduous sliding window creation is not yet implemented")
 69        return []
 70    
 71    def get_supported_formats(self) -> List[str]:
 72        """
 73        Get list of supported file formats for Arduous dataset.
 74        
 75        Returns:
 76            List of supported file extensions
 77        """
 78        return ['.csv', '.txt']
 79    
 80    def get_sensor_info(self) -> Dict[str, List[str]]:
 81        """
 82        Get information about sensors in the dataset.
 83        
 84        Returns:
 85            Dictionary containing sensor information
 86        """
 87        return {
 88            'sensors': self.metadata['sensors'],
 89            'components': self.metadata['components'],
 90            'sampling_frequency': self.metadata['sampling_frequency']
 91        }
 92    
 93    def get_activity_info(self) -> List[str]:
 94        """
 95        Get information about activities in the dataset.
 96        
 97        Returns:
 98            List of activity types
 99        """
100        return self.metadata['activities']

 25class PhysioNetLoader(BaseDatasetLoader):
 26    """
 27    PhysioNet VGRF dataset loader class.
 28    
 29    This class handles loading and processing of the PhysioNet Gait in Parkinson's Disease dataset.
 30    The dataset contains vertical ground reaction force (VGRF) data from subjects with Parkinson's 
 31    disease and healthy controls.
 32    """
 33    
 34    def __init__(self):
 35        super().__init__(
 36            name="physionet",
 37            description="PhysioNet Gait in Parkinson's Disease Dataset - Contains VGRF data from subjects with Parkinson's disease and healthy controls"
 38        )
 39        self.metadata = {
 40            'sensors': ['VGRF_L1', 'VGRF_L2', 'VGRF_L3', 'VGRF_L4', 'VGRF_L5', 'VGRF_L6', 'VGRF_L7', 'VGRF_L8',
 41                       'VGRF_R1', 'VGRF_R2', 'VGRF_R3', 'VGRF_R4', 'VGRF_R5', 'VGRF_R6', 'VGRF_R7', 'VGRF_R8'],
 42            'sampling_frequency': 100,  # 100 Hz sampling frequency
 43            'subjects': {
 44                'Co': 'Control subjects',
 45                'Pt': 'Parkinson\'s disease patients'
 46            },
 47            'window_size': 600,  # 6 seconds at 100 Hz
 48            'url': 'https://physionet.org/files/gaitpdb/1.0.0/'
 49        }
 50        self.labels = []
 51        self.subject_types = []
 52    
 53    def _download_physionet_data(self, data_dir: str) -> str:
 54        """
 55        Download PhysioNet dataset if not already present.
 56        
 57        Args:
 58            data_dir: Directory to store the dataset
 59            
 60        Returns:
 61            Path to the downloaded/existing dataset directory
 62        """
 63        dataset_path = os.path.join(data_dir, "physionet_gaitpdb")
 64        
 65        if os.path.exists(dataset_path) and len(os.listdir(dataset_path)) > 0:
 66            print(f"PhysioNet dataset already exists at: {dataset_path}")
 67            return dataset_path
 68        
 69        os.makedirs(dataset_path, exist_ok=True)
 70        
 71        # Download the dataset files
 72        base_url = "https://physionet.org/files/gaitpdb/1.0.0/"
 73        
 74        # Get list of files (basic file names based on the reference)
 75        file_patterns = [
 76            # Control subjects - Ga prefix
 77            *[f"GaCo{i:02d}_{j:02d}.txt" for i in range(1, 18) for j in range(1, 3)],
 78            "GaCo22_01.txt", "GaCo22_10.txt",
 79            
 80            # Parkinson's patients - Ga prefix
 81            *[f"GaPt{i:02d}_{j:02d}.txt" for i in range(3, 10) for j in range(1, 3)],
 82            *[f"GaPt{i:02d}_{j:02d}.txt" for i in range(12, 34) for j in range(1, 3)],
 83            *[f"GaPt{i:02d}_10.txt" for i in range(13, 34)],
 84            
 85            # Control subjects - Ju prefix
 86            *[f"JuCo{i:02d}_01.txt" for i in range(1, 27)],
 87            
 88            # Parkinson's patients - Ju prefix
 89            *[f"JuPt{i:02d}_{j:02d}.txt" for i in range(1, 30) for j in range(1, 8)],
 90            
 91            # Control subjects - Si prefix
 92            *[f"SiCo{i:02d}_01.txt" for i in range(1, 31)],
 93            
 94            # Parkinson's patients - Si prefix
 95            *[f"SiPt{i:02d}_01.txt" for i in range(2, 41)]
 96        ]
 97        
 98        print(f"Downloading PhysioNet dataset to {dataset_path}")
 99        for filename in tqdm(file_patterns, desc="Downloading files"):
100            file_url = base_url + filename
101            file_path = os.path.join(dataset_path, filename)
102            
103            if os.path.exists(file_path):
104                continue
105                
106            try:
107                response = requests.get(file_url, stream=True)
108                if response.status_code == 200:
109                    with open(file_path, 'wb') as f:
110                        for chunk in response.iter_content(chunk_size=8192):
111                            f.write(chunk)
112                else:
113                    print(f"Could not download {filename} (status: {response.status_code})")
114            except Exception as e:
115                print(f"Error downloading {filename}: {e}")
116        
117        return dataset_path
118    
119    def load_data(self, data_dir: str, **kwargs) -> Tuple[List[pd.DataFrame], List[str]]:
120        """
121        Load PhysioNet VGRF dataset from the specified directory.
122        
123        Args:
124            data_dir: Directory to store/find the dataset
125            **kwargs: Additional arguments (unused for PhysioNet)
126            
127        Returns:
128            Tuple of (data_list, names_list)
129        """
130        # Download dataset if needed
131        dataset_path = self._download_physionet_data(data_dir)
132        
133        physionet_data = []
134        physionet_names = []
135        self.labels = []
136        self.subject_types = []
137        
138        # Load all available files
139        for filepath in sorted(glob(os.path.join(dataset_path, "Ga*.txt"))):
140            filename = os.path.basename(filepath)
141            
142            # Extract subject type from filename
143            if 'Co' in filename:
144                subject_type = 'Control'
145                label = 'Co'
146            elif 'Pt' in filename:
147                subject_type = 'Patient'
148                label = 'Pt'
149            else:
150                continue  # Skip files that don't match expected pattern
151            
152            try:
153                # Read the file - PhysioNet files are tab-delimited with variable columns
154                # Column 0: time, Columns 1-16: VGRF sensors, additional columns may exist
155                df = pd.read_csv(filepath, delimiter='\t', header=None)
156                
157                # Handle variable number of columns
158                n_cols = min(df.shape[1], 19)  # Limit to 19 columns max
159                df = df.iloc[:, :n_cols]
160                
161                # Create column names
162                col_names = ['time']
163                for i in range(1, n_cols):
164                    if i <= 8:
165                        col_names.append(f'VGRF_L{i}')
166                    elif i <= 16:
167                        col_names.append(f'VGRF_R{i-8}')
168                    else:
169                        col_names.append(f'sensor_{i}')
170                
171                df.columns = col_names
172                
173                # Set time as index
174                df = df.set_index('time')
175                
176                # Add subject metadata
177                df['subject_type'] = subject_type
178                df['label'] = label
179                
180                physionet_data.append(df)
181                physionet_names.append(filename)
182                self.labels.append(label)
183                self.subject_types.append(subject_type)
184                
185            except Exception as e:
186                print(f"Error loading {filename}: {e}")
187                continue
188        
189        # Store loaded data
190        self.data = physionet_data
191        self.names = physionet_names
192        
193        print(f"Loaded {len(physionet_data)} PhysioNet files")
194        print(f"Subject distribution: {dict(zip(*np.unique(self.subject_types, return_counts=True)))}")
195        
196        return physionet_data, physionet_names
197    
198    def create_sliding_windows(self, data: List[pd.DataFrame], names: List[str], 
199                             window_size: int = 600, step_size: int = 100) -> List[Dict]:
200        """
201        Create sliding windows from the PhysioNet dataset.
202        
203        Args:
204            data: List of DataFrames containing PhysioNet data
205            names: List of names corresponding to the data
206            window_size: Size of the sliding window (default: 600 for 6 seconds at 100Hz)
207            step_size: Step size for the sliding window (default: 100)
208            
209        Returns:
210            List of dictionaries containing sliding windows for each DataFrame
211        """
212        windows_data = []
213        
214        for idx, df in enumerate(data):
215            # Remove metadata columns for windowing
216            sensor_columns = [col for col in df.columns if col.startswith('VGRF_') or col.startswith('sensor_')]
217            df_sensors = df[sensor_columns]
218            
219            if df_sensors.empty or len(df_sensors) < window_size:
220                continue
221                
222            windows = []
223            
224            # Create windows for each sensor
225            for col in sensor_columns:
226                try:
227                    window_data = sliding_window(df_sensors[col].values, window_size, step_size)
228                    windows.append({"name": col, "data": window_data})
229                except Exception as e:
230                    print(f"Error creating windows for {col} in {names[idx]}: {e}")
231                    continue
232            
233            if windows:
234                windows_data.append({
235                    "name": names[idx],
236                    "windows": windows,
237                    "metadata": {
238                        "subject_type": df['subject_type'].iloc[0] if 'subject_type' in df.columns else 'Unknown',
239                        "label": df['label'].iloc[0] if 'label' in df.columns else 'Unknown',
240                        "window_size": window_size,
241                        "step_size": step_size,
242                        "num_windows": len(windows[0]["data"]) if windows else 0
243                    }
244                })
245        
246        return windows_data
247    
248    def get_supported_formats(self) -> List[str]:
249        """
250        Get list of supported file formats for PhysioNet dataset.
251        
252        Returns:
253            List of supported file extensions
254        """
255        return ['.txt']
256    
257    def get_sensor_info(self) -> Dict[str, List[str]]:
258        """
259        Get information about sensors in the dataset.
260        
261        Returns:
262            Dictionary containing sensor information
263        """
264        return {
265            'sensors': self.metadata['sensors'],
266            'sampling_frequency': self.metadata['sampling_frequency'],
267            'window_size': self.metadata['window_size']
268        }
269    
270    def get_subject_info(self) -> Dict[str, str]:
271        """
272        Get information about subjects in the dataset.
273        
274        Returns:
275            Dictionary containing subject information
276        """
277        return self.metadata['subjects']
278    
279    def get_labels(self) -> List[str]:
280        """
281        Get labels for loaded data.
282        
283        Returns:
284            List of labels corresponding to loaded data
285        """
286        return self.labels
287    
288    def filter_by_subject_type(self, subject_type: str) -> Tuple[List[pd.DataFrame], List[str]]:
289        """
290        Filter loaded data by subject type.
291        
292        Args:
293            subject_type: 'Control' or 'Patient'
294            
295        Returns:
296            Tuple of (filtered_data, filtered_names)
297        """
298        if not self.data:
299            raise ValueError("No data loaded. Call load_data() first.")
300        
301        filtered_data = []
302        filtered_names = []
303        
304        for i, df in enumerate(self.data):
305            if df['subject_type'].iloc[0] == subject_type:
306                filtered_data.append(df)
307                filtered_names.append(self.names[i])
308        
309        return filtered_data, filtered_names