|
|
"""
|
|
|
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
|
|
|
|
