gaitsetpy

 21class BaseDatasetLoader(ABC):
 22    """
 23    Base class for all dataset loaders.
 24    
 25    All dataset loaders should inherit from this class and implement the required methods.
 26    This class provides thread-safe concurrent downloading capabilities for efficient data retrieval.
 27    """
 28    
 29    def __init__(self, name: str, description: str = "", max_workers: int = 8):
 30        """
 31        Initialize the dataset loader.
 32        
 33        Args:
 34            name: Name of the dataset
 35            description: Description of the dataset
 36            max_workers: Maximum number of concurrent download threads (default: 8)
 37        """
 38        self.name = name
 39        self.description = description
 40        self.data = None
 41        self.metadata = {}
 42        self.max_workers = max_workers
 43        self._download_stats = {'success': 0, 'failed': 0, 'skipped': 0}
 44    
 45    @abstractmethod
 46    def load_data(self, data_dir: str, **kwargs) -> Tuple[List[pd.DataFrame], List[str]]:
 47        """
 48        Load dataset from the specified directory.
 49        
 50        Args:
 51            data_dir: Directory containing the dataset
 52            **kwargs: Additional arguments specific to the dataset
 53            
 54        Returns:
 55            Tuple of (data_list, names_list)
 56        """
 57        pass
 58    
 59    @abstractmethod
 60    def create_sliding_windows(self, data: List[pd.DataFrame], names: List[str], 
 61                             window_size: int = 192, step_size: int = 32) -> List[Dict]:
 62        """
 63        Create sliding windows from the loaded data.
 64        
 65        Args:
 66            data: List of DataFrames
 67            names: List of names corresponding to the data
 68            window_size: Size of each sliding window
 69            step_size: Step size for sliding windows
 70            
 71        Returns:
 72            List of dictionaries containing sliding windows
 73        """
 74        pass
 75    
 76    @abstractmethod
 77    def get_supported_formats(self) -> List[str]:
 78        """
 79        Get list of supported file formats.
 80        
 81        Returns:
 82            List of supported file extensions
 83        """
 84        pass
 85    
 86    def _download_file(self, url: str, dest_path: str, 
 87                      chunk_size: int = 8192, timeout: int = 30) -> Tuple[bool, str]:
 88        """
 89        Download a single file from URL to destination path.
 90        
 91        This method is thread-safe and can be called concurrently.
 92        
 93        Args:
 94            url: URL to download from
 95            dest_path: Destination file path
 96            chunk_size: Size of chunks to download (default: 8192 bytes)
 97            timeout: Request timeout in seconds (default: 30)
 98            
 99        Returns:
100            Tuple of (success: bool, message: str)
101        """
102        try:
103            # Check if file already exists
104            if os.path.exists(dest_path):
105                self._download_stats['skipped'] += 1
106                return True, f"File already exists: {dest_path}"
107            
108            # Make the request
109            response = requests.get(url, stream=True, timeout=timeout)
110            
111            if response.status_code == 200:
112                # Ensure parent directory exists
113                os.makedirs(os.path.dirname(dest_path) if os.path.dirname(dest_path) else '.', exist_ok=True)
114                
115                # Write file in chunks
116                with open(dest_path, 'wb') as f:
117                    for chunk in response.iter_content(chunk_size=chunk_size):
118                        if chunk:
119                            f.write(chunk)
120                
121                self._download_stats['success'] += 1
122                return True, f"Successfully downloaded: {dest_path}"
123            else:
124                self._download_stats['failed'] += 1
125                return False, f"HTTP {response.status_code}: {url}"
126                
127        except requests.exceptions.Timeout:
128            self._download_stats['failed'] += 1
129            return False, f"Timeout downloading: {url}"
130        except requests.exceptions.RequestException as e:
131            self._download_stats['failed'] += 1
132            return False, f"Request error for {url}: {str(e)}"
133        except IOError as e:
134            self._download_stats['failed'] += 1
135            return False, f"IO error for {dest_path}: {str(e)}"
136        except Exception as e:
137            self._download_stats['failed'] += 1
138            return False, f"Unexpected error for {url}: {str(e)}"
139    
140    def download_files_concurrent(self, 
141                                  download_tasks: List[Dict[str, str]], 
142                                  show_progress: bool = True,
143                                  desc: str = "Downloading files") -> Dict[str, Any]:
144        """
145        Download multiple files concurrently using a thread pool.
146        
147        Args:
148            download_tasks: List of dicts with 'url' and 'dest_path' keys
149            show_progress: Whether to show progress bar (default: True)
150            desc: Description for progress bar
151            
152        Returns:
153            Dictionary with download statistics and results
154            
155        Example:
156            tasks = [
157                {'url': 'http://example.com/file1.txt', 'dest_path': '/path/to/file1.txt'},
158                {'url': 'http://example.com/file2.txt', 'dest_path': '/path/to/file2.txt'}
159            ]
160            results = loader.download_files_concurrent(tasks)
161        """
162        # Reset stats
163        self._download_stats = {'success': 0, 'failed': 0, 'skipped': 0}
164        
165        results = []
166        failed_downloads = []
167        
168        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
169            # Submit all download tasks
170            future_to_task = {
171                executor.submit(self._download_file, task['url'], task['dest_path']): task
172                for task in download_tasks
173            }
174            
175            # Process completed tasks with optional progress bar
176            if show_progress:
177                futures = tqdm(as_completed(future_to_task), 
178                             total=len(download_tasks), 
179                             desc=desc)
180            else:
181                futures = as_completed(future_to_task)
182            
183            for future in futures:
184                task = future_to_task[future]
185                try:
186                    success, message = future.result()
187                    results.append({
188                        'url': task['url'],
189                        'dest_path': task['dest_path'],
190                        'success': success,
191                        'message': message
192                    })
193                    
194                    if not success:
195                        failed_downloads.append({
196                            'url': task['url'],
197                            'dest_path': task['dest_path'],
198                            'error': message
199                        })
200                        
201                except Exception as e:
202                    error_msg = f"Exception during download: {str(e)}"
203                    results.append({
204                        'url': task['url'],
205                        'dest_path': task['dest_path'],
206                        'success': False,
207                        'message': error_msg
208                    })
209                    failed_downloads.append({
210                        'url': task['url'],
211                        'dest_path': task['dest_path'],
212                        'error': error_msg
213                    })
214        
215        # Return comprehensive results
216        return {
217            'total': len(download_tasks),
218            'success': self._download_stats['success'],
219            'failed': self._download_stats['failed'],
220            'skipped': self._download_stats['skipped'],
221            'failed_downloads': failed_downloads,
222            'all_results': results
223        }
224    
225    def set_max_workers(self, max_workers: int):
226        """
227        Set the maximum number of concurrent download threads.
228        
229        Args:
230            max_workers: Maximum number of threads (must be positive)
231        """
232        if max_workers < 1:
233            raise ValueError("max_workers must be at least 1")
234        self.max_workers = max_workers
235    
236    def get_download_stats(self) -> Dict[str, int]:
237        """
238        Get statistics from the last download operation.
239        
240        Returns:
241            Dictionary with success, failed, and skipped counts
242        """
243        return self._download_stats.copy()
244    
245    def get_info(self) -> Dict[str, Any]:
246        """
247        Get information about the dataset.
248        
249        Returns:
250            Dictionary containing dataset information
251        """
252        return {
253            'name': self.name,
254            'description': self.description,
255            'metadata': self.metadata,
256            'supported_formats': self.get_supported_formats(),
257            'max_workers': self.max_workers
258        }

261class BaseFeatureExtractor(ABC):
262    """
263    Base class for all feature extractors.
264    
265    All feature extractors should inherit from this class and implement the required methods.
266    """
267    
268    def __init__(self, name: str, description: str = ""):
269        """
270        Initialize the feature extractor.
271        
272        Args:
273            name: Name of the feature extractor
274            description: Description of the feature extractor
275        """
276        self.name = name
277        self.description = description
278        self.config = {}
279    
280    @abstractmethod
281    def extract_features(self, windows: List[Dict], fs: int, **kwargs) -> List[Dict]:
282        """
283        Extract features from sliding windows.
284        
285        Args:
286            windows: List of sliding window dictionaries
287            fs: Sampling frequency
288            **kwargs: Additional arguments for feature extraction
289            
290        Returns:
291            List of feature dictionaries
292        """
293        pass
294    
295    @abstractmethod
296    def get_feature_names(self) -> List[str]:
297        """
298        Get names of features extracted by this extractor.
299        
300        Returns:
301            List of feature names
302        """
303        pass
304    
305    def configure(self, config: Dict[str, Any]):
306        """
307        Configure the feature extractor.
308        
309        Args:
310            config: Configuration dictionary
311        """
312        self.config.update(config)
313    
314    def get_info(self) -> Dict[str, Any]:
315        """
316        Get information about the feature extractor.
317        
318        Returns:
319            Dictionary containing feature extractor information
320        """
321        return {
322            'name': self.name,
323            'description': self.description,
324            'config': self.config,
325            'feature_names': self.get_feature_names()
326        }

329class BasePreprocessor(ABC):
330    """
331    Base class for all preprocessors.
332    
333    All preprocessors should inherit from this class and implement the required methods.
334    """
335    
336    def __init__(self, name: str, description: str = ""):
337        """
338        Initialize the preprocessor.
339        
340        Args:
341            name: Name of the preprocessor
342            description: Description of the preprocessor
343        """
344        self.name = name
345        self.description = description
346        self.config = {}
347        self.fitted = False
348    
349    @abstractmethod
350    def fit(self, data: Union[pd.DataFrame, np.ndarray], **kwargs):
351        """
352        Fit the preprocessor to the data.
353        
354        Args:
355            data: Input data to fit on
356            **kwargs: Additional arguments for fitting
357        """
358        pass
359    
360    @abstractmethod
361    def transform(self, data: Union[pd.DataFrame, np.ndarray], **kwargs) -> Union[pd.DataFrame, np.ndarray]:
362        """
363        Transform the data using the fitted preprocessor.
364        
365        Args:
366            data: Input data to transform
367            **kwargs: Additional arguments for transformation
368            
369        Returns:
370            Transformed data
371        """
372        pass
373    
374    def fit_transform(self, data: Union[pd.DataFrame, np.ndarray], **kwargs) -> Union[pd.DataFrame, np.ndarray]:
375        """
376        Fit the preprocessor and transform the data.
377        
378        Args:
379            data: Input data to fit and transform
380            **kwargs: Additional arguments
381            
382        Returns:
383            Transformed data
384        """
385        self.fit(data, **kwargs)
386        return self.transform(data, **kwargs)
387    
388    def configure(self, config: Dict[str, Any]):
389        """
390        Configure the preprocessor.
391        
392        Args:
393            config: Configuration dictionary
394        """
395        self.config.update(config)
396    
397    def get_info(self) -> Dict[str, Any]:
398        """
399        Get information about the preprocessor.
400        
401        Returns:
402            Dictionary containing preprocessor information
403        """
404        return {
405            'name': self.name,
406            'description': self.description,
407            'config': self.config,
408            'fitted': self.fitted
409        }

412class BaseEDAAnalyzer(ABC):
413    """
414    Base class for all EDA analyzers.
415    
416    All EDA analyzers should inherit from this class and implement the required methods.
417    """
418    
419    def __init__(self, name: str, description: str = ""):
420        """
421        Initialize the EDA analyzer.
422        
423        Args:
424            name: Name of the EDA analyzer
425            description: Description of the EDA analyzer
426        """
427        self.name = name
428        self.description = description
429        self.config = {}
430    
431    @abstractmethod
432    def analyze(self, data: Union[pd.DataFrame, List[pd.DataFrame]], **kwargs) -> Dict[str, Any]:
433        """
434        Perform analysis on the data.
435        
436        Args:
437            data: Input data to analyze
438            **kwargs: Additional arguments for analysis
439            
440        Returns:
441            Dictionary containing analysis results
442        """
443        pass
444    
445    @abstractmethod
446    def visualize(self, data: Union[pd.DataFrame, List[pd.DataFrame]], **kwargs):
447        """
448        Create visualizations of the data.
449        
450        Args:
451            data: Input data to visualize
452            **kwargs: Additional arguments for visualization
453        """
454        pass
455    
456    def configure(self, config: Dict[str, Any]):
457        """
458        Configure the EDA analyzer.
459        
460        Args:
461            config: Configuration dictionary
462        """
463        self.config.update(config)
464    
465    def get_info(self) -> Dict[str, Any]:
466        """
467        Get information about the EDA analyzer.
468        
469        Returns:
470            Dictionary containing EDA analyzer information
471        """
472        return {
473            'name': self.name,
474            'description': self.description,
475            'config': self.config
476        }