snakemake-interface-scheduler-plugins

from typing import Dict, Mapping, Optional, Union, Sequence from dataclasses import dataclass, field from snakemake_interface_scheduler_plugins.settings import SchedulerSettingsBase from snakemake_interface_scheduler_plugins.base import SchedulerBase from snakemake_interface_scheduler_plugins.interfaces.dag import SchedulerDAGInterface from snakemake_interface_scheduler_plugins.interfaces.jobs import SchedulerJobInterface # Define settings for your scheduler plugin. These will appear in the Snakemake CLI # as --scheduler-<plugin-name>-<param-name>. All fields should be Optional. @dataclass class SchedulerSettings(SchedulerSettingsBase): my_custom_param: Optional[int] = field( default=None, metadata={ "help": "A custom parameter for my scheduler plugin.", "env_var": False # Use True for sensitive info like passwords } ) # Implement the core scheduler logic class Scheduler(SchedulerBase): def __init__( self, workflow: 'snakemake.workflow.Workflow', dag: SchedulerDAGInterface, dryrun: bool, scheduler_settings: SchedulerSettings ): super().__init__(workflow, dag, dryrun, scheduler_settings) self.scheduler_settings = scheduler_settings print(f"Initializing MyCustomScheduler with param: {self.scheduler_settings.my_custom_param}") def _choose_single_job(self, jobs: Sequence[SchedulerJobInterface]) -> Optional[SchedulerJobInterface]: # Example: always pick the first job available if jobs: print(f"MyCustomScheduler choosing job: {jobs[0].jobid}") return jobs[0] return None def _get_selected_jobs( self, jobs: Sequence[SchedulerJobInterface] ) -> Optional[Sequence[SchedulerJobInterface]]: # Example: return all available jobs, or None to fall back to Snakemake's greedy scheduler if self.scheduler_settings.my_custom_param is not None and self.scheduler_settings.my_custom_param < 0: print("MyCustomScheduler falling back to greedy scheduler.") return None # Indicate fallback to Snakemake's internal greedy scheduler chosen_job = self._choose_single_job(jobs) if chosen_job: return [chosen_job] return [] # Note: This is a skeleton for plugin development. To use, a Snakemake plugin package # named `snakemake-scheduler-plugin-<name>` must be created and published to PyPI, # which then exposes this `Scheduler` class and `SchedulerSettings` via entry points.