cellmap_flow.finetune.job_manager.manager ========================================= .. py:module:: cellmap_flow.finetune.job_manager.manager .. autoapi-nested-parse:: FinetuneJobManager: the finetune jobs a dashboard follows. The finetune routes call it, as the session's ``finetune_job_manager``. It holds the jobs by job id, and their listeners, and puts the package's modules together: - ``submit_finetuning_job``: check the request and work out what to run (submit), make the run's directory and its record (persistence), launch the trainer, and follow the job; - ``rehydrate_session``: follow again the jobs an earlier dashboard left running (persistence.rehydrate); - ``monitor_job``: what each job's monitor thread runs (monitor); - ``restart_finetuning_job`` (restart) and ``cancel_job``; - ``get_job_status``, ``list_jobs``, ``get_job_logs``, ``get_job``; - ``add_listener``, ``remove_listener`` (listener). Stopping a job early is between the finetune route and the trainer: the route writes the run's ``stop_signal.json``. Attributes ---------- .. autoapisummary:: cellmap_flow.finetune.job_manager.manager.logger Classes ------- .. autoapisummary:: cellmap_flow.finetune.job_manager.manager.FinetuneJobManager Module Contents --------------- .. py:data:: logger .. py:class:: FinetuneJobManager Orchestrate finetuning jobs from submission to completion. Manages the full lifecycle: 1. Validation and job submission to LSF 2. Background monitoring of training progress, told to the listeners 3. Post-training checks, and the job's record in its metadata.json 4. Restarts and cancellation .. py:attribute:: jobs :type: Dict[str, cellmap_flow.finetune.job_manager.state.FinetuneJob] .. py:method:: add_listener(listener) -> None Tell ``listener`` about job events; see listener.FinetuneJobListener. Adding one already added does nothing. .. py:method:: remove_listener(listener) -> None Stop telling ``listener``. .. py:method:: submit_finetuning_job(model_config, corrections_path: pathlib.Path, lora_r: int = 8, num_epochs: int = 10, batch_size: int = 8, learning_rate: float = 0.0001, output_base: Optional[pathlib.Path] = None, queue: Optional[str] = None, charge_group: Optional[str] = None, walltime: Optional[str] = None, checkpoint_path_override: Optional[pathlib.Path] = None, auto_serve: bool = True, mask_unannotated: bool = False, loss_type: str = 'combined', label_smoothing: float = 0.0, distillation_lambda: Optional[float] = None, distillation_scope: str = 'unlabeled', margin: float = 0.3, balance_classes: bool = False, augment: bool = False, output_type: str = 'binary', select_channel: Optional[int] = None, offsets: Optional[str] = None) -> cellmap_flow.finetune.job_manager.state.FinetuneJob Submit finetuning job to LSF cluster. :param model_config: Model configuration object (FlyModelConfig, etc.) :param corrections_path: Path to corrections.zarr directory :param lora_r: LoRA rank (default: 8) :param num_epochs: Number of training epochs (default: 10) :param batch_size: Training batch size (default: 8) :param learning_rate: Learning rate (default: 1e-4) :param output_base: Base directory for outputs (default: output/finetuning) :param queue: LSF queue name (default: the site's, gpu_h100 at Janelia) :param charge_group: LSF charge group (default: the site's, cellmap at Janelia) :param walltime: LSF run limit, "HH:MM" or minutes (default: the site's, 08:00 at Janelia) :param checkpoint_path_override: Optional path to override checkpoint detection (default: None) :param auto_serve: Automatically start inference server after training (default: True) :returns: FinetuneJob object tracking the submitted job :raises ValueError: If validation fails :raises RuntimeError: If job submission fails .. py:method:: rehydrate_session(session_path) -> int Pick up the jobs of a session that are still alive on the cluster. Jobs live only in this process's memory, so after a dashboard restart a running job disappeared: it could not be seen or cancelled, and with auto-serve it waited for a restart until walltime. Each job's metadata.json now records its LSF job id and status; a run whose recorded status is not final and that bjobs still reports as pending or running is monitored again, which also tells the listeners again once the log shows its server. persistence.rehydrate says which runs those are, with one bjobs call per session, and records the others that have ended as final. Returns how many jobs were picked up. .. py:method:: monitor_job(finetune_job: cellmap_flow.finetune.job_manager.state.FinetuneJob) Follow ``finetune_job`` until it ends (monitor.monitor_job); its monitor thread runs this. .. py:method:: cancel_job(job_id: str) -> bool Cancel a running job. :param job_id: Job ID to cancel :returns: True if successfully cancelled, False otherwise .. py:method:: get_job_status(job_id: str) -> Optional[Dict[str, Any]] Get detailed status of a specific job. :param job_id: Job ID to query :returns: Dictionary with job status details, or None if not found .. py:method:: list_jobs() -> List[Dict[str, Any]] Get list of all jobs with their status. :returns: List of job status dictionaries .. py:method:: get_job_logs(job_id: str) -> Optional[str] Get full log content for a job. :param job_id: Job ID :returns: Log file content as string, or None if not found. A byte that is not UTF-8 is read as U+FFFD (see tailer.LogTailer). .. py:method:: get_job(job_id: str) -> Optional[cellmap_flow.finetune.job_manager.state.FinetuneJob] Get a FinetuneJob object by ID. :param job_id: Job ID to retrieve :returns: FinetuneJob object, or None if not found .. py:method:: restart_finetuning_job(job_id: str, updated_params: Optional[Dict[str, Any]] = None) -> cellmap_flow.finetune.job_manager.state.FinetuneJob Restart training on the same GPU via control endpoint. Primary path sends an HTTP restart request to the running inference server in the same process as the training loop. Falls back to file signal if control endpoint is unavailable. :param job_id: ID of job to restart :param updated_params: Dict of updated training parameters :returns: Same FinetuneJob object (updated in-place) :raises ValueError: If job not found or not in a restartable state