| | """
|
| | BOINC Client Integration
|
| | Handles distributed computing task submission and monitoring
|
| | """
|
| |
|
| | import os
|
| | import json
|
| | import time
|
| | import requests
|
| | from typing import Dict, List, Optional
|
| | from pathlib import Path
|
| | from dataclasses import dataclass, asdict
|
| | from datetime import datetime
|
| | import yaml
|
| |
|
| | @dataclass
|
| | class WorkUnit:
|
| | """Represents a BOINC work unit"""
|
| | id: str
|
| | name: str
|
| | workunit_type: str
|
| | input_file: str
|
| | status: str
|
| | created_at: str
|
| | completed_at: Optional[str] = None
|
| | result_file: Optional[str] = None
|
| | error: Optional[str] = None
|
| |
|
| | class BOINCClient:
|
| | """BOINC client for distributed computing integration"""
|
| |
|
| | def __init__(self, config_path: str = "config.yml"):
|
| | with open(config_path, 'r') as f:
|
| | self.config = yaml.safe_load(f)['boinc']
|
| |
|
| | self.project_url = self.config['project_url']
|
| | self.work_dir = Path(self.config['work_dir'])
|
| | self.work_dir.mkdir(parents=True, exist_ok=True)
|
| |
|
| | self.tasks_file = self.work_dir / "tasks.json"
|
| | self.tasks = self._load_tasks()
|
| |
|
| | def _load_tasks(self) -> Dict[str, WorkUnit]:
|
| | """Load existing tasks from disk"""
|
| | if self.tasks_file.exists():
|
| | with open(self.tasks_file, 'r') as f:
|
| | data = json.load(f)
|
| | return {k: WorkUnit(**v) for k, v in data.items()}
|
| | return {}
|
| |
|
| | def _save_tasks(self):
|
| | """Save tasks to disk"""
|
| | with open(self.tasks_file, 'w') as f:
|
| | data = {k: asdict(v) for k, v in self.tasks.items()}
|
| | json.dump(data, f, indent=2)
|
| |
|
| | def submit_task(
|
| | self,
|
| | workunit_type: str,
|
| | input_file: str,
|
| | name: Optional[str] = None
|
| | ) -> str:
|
| | """
|
| | Submit a new work unit to BOINC
|
| |
|
| | Args:
|
| | workunit_type: Type of analysis (variant_calling, blast_search, etc.)
|
| | input_file: Path to input data file
|
| | name: Optional custom name for the work unit
|
| |
|
| | Returns:
|
| | Work unit ID
|
| | """
|
| | task_id = f"wu_{int(time.time() * 1000)}"
|
| |
|
| | if name is None:
|
| | name = f"{workunit_type}_{task_id}"
|
| |
|
| |
|
| | work_unit = WorkUnit(
|
| | id=task_id,
|
| | name=name,
|
| | workunit_type=workunit_type,
|
| | input_file=input_file,
|
| | status="pending",
|
| | created_at=datetime.now().isoformat()
|
| | )
|
| |
|
| |
|
| |
|
| | self._simulate_submission(work_unit)
|
| |
|
| | self.tasks[task_id] = work_unit
|
| | self._save_tasks()
|
| |
|
| | return task_id
|
| |
|
| | def _simulate_submission(self, work_unit: WorkUnit):
|
| | """
|
| | Simulate BOINC submission (for development/demo purposes)
|
| | In production, replace with actual BOINC API calls
|
| | """
|
| |
|
| | task_dir = self.work_dir / work_unit.id
|
| | task_dir.mkdir(exist_ok=True)
|
| |
|
| |
|
| | input_path = Path(work_unit.input_file)
|
| | if input_path.exists():
|
| | import shutil
|
| | shutil.copy(input_path, task_dir / input_path.name)
|
| |
|
| |
|
| | metadata = {
|
| | "task_id": work_unit.id,
|
| | "type": work_unit.workunit_type,
|
| | "input": work_unit.input_file,
|
| | "submitted": work_unit.created_at
|
| | }
|
| |
|
| | with open(task_dir / "metadata.json", 'w') as f:
|
| | json.dump(metadata, f, indent=2)
|
| |
|
| | def get_task_status(self, task_id: str) -> Optional[WorkUnit]:
|
| | """Get status of a specific task"""
|
| | return self.tasks.get(task_id)
|
| |
|
| | def list_tasks(
|
| | self,
|
| | status: Optional[str] = None,
|
| | workunit_type: Optional[str] = None
|
| | ) -> List[WorkUnit]:
|
| | """
|
| | List all tasks with optional filtering
|
| |
|
| | Args:
|
| | status: Filter by status (pending, running, completed, failed)
|
| | workunit_type: Filter by work unit type
|
| | """
|
| | tasks = list(self.tasks.values())
|
| |
|
| | if status:
|
| | tasks = [t for t in tasks if t.status == status]
|
| |
|
| | if workunit_type:
|
| | tasks = [t for t in tasks if t.workunit_type == workunit_type]
|
| |
|
| | return sorted(tasks, key=lambda t: t.created_at, reverse=True)
|
| |
|
| | def update_task_status(self, task_id: str, status: str, **kwargs):
|
| | """Update task status and additional fields"""
|
| | if task_id in self.tasks:
|
| | self.tasks[task_id].status = status
|
| |
|
| | for key, value in kwargs.items():
|
| | if hasattr(self.tasks[task_id], key):
|
| | setattr(self.tasks[task_id], key, value)
|
| |
|
| | if status == "completed":
|
| | self.tasks[task_id].completed_at = datetime.now().isoformat()
|
| |
|
| | self._save_tasks()
|
| |
|
| | def cancel_task(self, task_id: str) -> bool:
|
| | """Cancel a pending or running task"""
|
| | if task_id in self.tasks:
|
| | task = self.tasks[task_id]
|
| | if task.status in ["pending", "running"]:
|
| | task.status = "cancelled"
|
| | self._save_tasks()
|
| | return True
|
| | return False
|
| |
|
| | def get_results(self, task_id: str) -> Optional[Path]:
|
| | """Get results file for a completed task"""
|
| | if task_id in self.tasks:
|
| | task = self.tasks[task_id]
|
| | if task.status == "completed" and task.result_file:
|
| | result_path = Path(task.result_file)
|
| | if result_path.exists():
|
| | return result_path
|
| | return None
|
| |
|
| | def get_statistics(self) -> Dict:
|
| | """Get overall statistics about BOINC tasks"""
|
| | total = len(self.tasks)
|
| | by_status = {}
|
| | by_type = {}
|
| |
|
| | for task in self.tasks.values():
|
| | by_status[task.status] = by_status.get(task.status, 0) + 1
|
| | by_type[task.workunit_type] = by_type.get(task.workunit_type, 0) + 1
|
| |
|
| | completed = [t for t in self.tasks.values() if t.completed_at]
|
| |
|
| | if completed:
|
| | avg_time = sum([
|
| | (datetime.fromisoformat(t.completed_at) -
|
| | datetime.fromisoformat(t.created_at)).total_seconds()
|
| | for t in completed
|
| | ]) / len(completed)
|
| | else:
|
| | avg_time = 0
|
| |
|
| | return {
|
| | "total_tasks": total,
|
| | "by_status": by_status,
|
| | "by_type": by_type,
|
| | "completed_tasks": len(completed),
|
| | "average_completion_time_seconds": avg_time
|
| | }
|
| |
|
| |
|
| | class BOINCTaskManager:
|
| | """High-level task manager for common workflows"""
|
| |
|
| | def __init__(self):
|
| | self.client = BOINCClient()
|
| |
|
| | def submit_variant_calling(self, fastq_file: str) -> str:
|
| | """Submit variant calling task"""
|
| | return self.client.submit_task(
|
| | workunit_type="variant_calling",
|
| | input_file=fastq_file,
|
| | name=f"variant_calling_{Path(fastq_file).stem}"
|
| | )
|
| |
|
| | def submit_blast_search(self, sequence_file: str) -> str:
|
| | """Submit BLAST search task"""
|
| | return self.client.submit_task(
|
| | workunit_type="blast_search",
|
| | input_file=sequence_file,
|
| | name=f"blast_{Path(sequence_file).stem}"
|
| | )
|
| |
|
| | def submit_alignment(self, fastq_file: str) -> str:
|
| | """Submit sequence alignment task"""
|
| | return self.client.submit_task(
|
| | workunit_type="alignment",
|
| | input_file=fastq_file,
|
| | name=f"alignment_{Path(fastq_file).stem}"
|
| | )
|
| |
|
| | def submit_annotation(self, vcf_file: str) -> str:
|
| | """Submit variant annotation task"""
|
| | return self.client.submit_task(
|
| | workunit_type="annotation",
|
| | input_file=vcf_file,
|
| | name=f"annotation_{Path(vcf_file).stem}"
|
| | )
|
| |
|
| | def batch_submit(
|
| | self,
|
| | workunit_type: str,
|
| | input_files: List[str]
|
| | ) -> List[str]:
|
| | """Submit multiple tasks at once"""
|
| | task_ids = []
|
| | for input_file in input_files:
|
| | task_id = self.client.submit_task(workunit_type, input_file)
|
| | task_ids.append(task_id)
|
| | return task_ids
|
| |
|
