|
| 1 | +"""BoltzGen model for protein structure and sequence design.""" |
| 2 | + |
| 3 | +from typing import Any, BinaryIO, Literal |
| 4 | + |
| 5 | +from pydantic import BaseModel, Field |
| 6 | + |
| 7 | +from openprotein.base import APISession |
| 8 | +from openprotein.common import ModelMetadata |
| 9 | +from openprotein.common.model_metadata import ModelDescription |
| 10 | +from openprotein.jobs import Future, Job |
| 11 | +from openprotein.models.base import ProteinModel |
| 12 | +from openprotein.protein import Protein |
| 13 | + |
| 14 | + |
| 15 | +class BoltzGenRequest(BaseModel): |
| 16 | + "Specification for an BoltzGen request." |
| 17 | + |
| 18 | + n: int = 1 |
| 19 | + # protein: Protein |
| 20 | + structure_text: str | None = None |
| 21 | + design_spec: dict[str, Any] |
| 22 | + diffusion_batch_size: int | None = None |
| 23 | + step_scale: float | None = None |
| 24 | + noise_scale: float | None = None |
| 25 | + |
| 26 | + |
| 27 | +class BoltzGenJob(Job): |
| 28 | + """Job schema for an BoltzGen request.""" |
| 29 | + |
| 30 | + job_type: Literal["/models/boltzgen"] |
| 31 | + |
| 32 | + |
| 33 | +class BoltzGenFuture(Future): |
| 34 | + """Future for handling the results of an BoltzGen job.""" |
| 35 | + |
| 36 | + job: BoltzGenJob |
| 37 | + |
| 38 | + def get_pdb(self, replicate: int = 0) -> str: |
| 39 | + """ |
| 40 | + Retrieve the PDB file for a specific design. |
| 41 | +
|
| 42 | + Args: |
| 43 | + design_index (int): The 0-based index of the design to retrieve. |
| 44 | +
|
| 45 | + Returns: |
| 46 | + str: The content of the PDB file as a string. |
| 47 | + """ |
| 48 | + return _boltzgen_api_result_get( |
| 49 | + session=self.session, job_id=self.id, replicate=replicate |
| 50 | + ) |
| 51 | + |
| 52 | + def get(self, replicate: int = 0): |
| 53 | + """Default result accessor, returns the first PDB.""" |
| 54 | + # TODO handle different design index |
| 55 | + return self.get_pdb(replicate=replicate) |
| 56 | + |
| 57 | + |
| 58 | +def _boltzgen_api_post( |
| 59 | + session: APISession, request: BoltzGenRequest, **kwargs |
| 60 | +) -> BoltzGenJob: |
| 61 | + """ |
| 62 | + POST a request for BoltzGen design. |
| 63 | +
|
| 64 | + Returns a Job object that can be used to retrieve results later. |
| 65 | + """ |
| 66 | + endpoint = "v1/design/models/boltzgen" |
| 67 | + body = request.model_dump(exclude_none=True) |
| 68 | + body.update(kwargs) |
| 69 | + response = session.post(endpoint, json=body) |
| 70 | + return BoltzGenJob.model_validate(response.json()) |
| 71 | + |
| 72 | + |
| 73 | +def _boltzgen_api_get_metadata(session: APISession) -> ModelMetadata: |
| 74 | + """ |
| 75 | + POST a request for BoltzGen design. |
| 76 | +
|
| 77 | + Returns a Job object that can be used to retrieve results later. |
| 78 | + """ |
| 79 | + endpoint = f"v1/design/models/boltzgen" |
| 80 | + response = session.get(endpoint) |
| 81 | + return ModelMetadata.model_validate(response.json()) |
| 82 | + |
| 83 | + |
| 84 | +def _boltzgen_api_result_get( |
| 85 | + session: APISession, job_id: str, replicate: int = 0 |
| 86 | +) -> str: |
| 87 | + """ |
| 88 | + POST a request for BoltzGen design. |
| 89 | +
|
| 90 | + # Returns a Job object that can be used to retrieve results later. |
| 91 | + """ |
| 92 | + endpoint = f"v1/design/{job_id}/results" |
| 93 | + response = session.get(endpoint, params={"replicate": replicate}) |
| 94 | + return response.text |
| 95 | + |
| 96 | + |
| 97 | +class BoltzGenModel(ProteinModel): |
| 98 | + """ |
| 99 | + BoltzGen model for generating de novo protein structures. |
| 100 | +
|
| 101 | + This model supports functionalities like unconditional design, scaffolding, |
| 102 | + and binder design. |
| 103 | + """ |
| 104 | + |
| 105 | + model_id: str = "boltzgen" |
| 106 | + |
| 107 | + def __init__(self, session: APISession, model_id: str = "boltzgen"): |
| 108 | + # The model_id from the API might be more specific, e.g., "boltzgen-v1.1" |
| 109 | + super().__init__(session, model_id) |
| 110 | + |
| 111 | + def get_metadata(self) -> ModelMetadata: |
| 112 | + return ModelMetadata( |
| 113 | + model_id="boltzgen", |
| 114 | + description=ModelDescription(summary="BoltzGen"), |
| 115 | + dimension=0, |
| 116 | + output_types=["pdb"], |
| 117 | + input_tokens=[], |
| 118 | + token_descriptions=[[]], |
| 119 | + ) |
| 120 | + |
| 121 | + def generate( |
| 122 | + self, |
| 123 | + design_spec: dict[str, Any], |
| 124 | + structure_file: str | bytes | BinaryIO | None = None, |
| 125 | + n: int = 1, |
| 126 | + diffusion_batch_size: int | None = None, |
| 127 | + step_scale: float | None = None, |
| 128 | + noise_scale: float | None = None, |
| 129 | + **kwargs, |
| 130 | + ) -> BoltzGenFuture: |
| 131 | + """ |
| 132 | + Run a protein structure generate job using BoltzGen. |
| 133 | +
|
| 134 | + Parameters |
| 135 | + ---------- |
| 136 | + design_spec : dict[str, Any] |
| 137 | + The BoltzGen design specification to run. This is the Python representation |
| 138 | + of the BoltzGen yaml request specification. |
| 139 | + structure_file : BinaryIO, optional |
| 140 | + An input PDB file (as a file-like object) used for inpainting or other |
| 141 | + guided design tasks where parts of an existing structure are provided. |
| 142 | + n : int, optional |
| 143 | + The number of unique design trajectories to run (default is 1). |
| 144 | + diffusion_batch_size : int, optional |
| 145 | + The batch size for diffusion sampling. Controls how many samples are |
| 146 | + processed in parallel during the diffusion process. |
| 147 | + step_scale : float, optional |
| 148 | + Scaling factor for the number of diffusion steps. Higher values may |
| 149 | + improve quality at the cost of longer generation time. |
| 150 | + noise_scale : float, optional |
| 151 | + Scaling factor for the noise schedule during diffusion. Controls the |
| 152 | + amount of noise added at each step of the reverse diffusion process. |
| 153 | +
|
| 154 | + Other Parameters |
| 155 | + ---------------- |
| 156 | + **kwargs : dict |
| 157 | + Additional keyword args that are passed directly to the boltzgen |
| 158 | + inference script. Overwrites any preceding options. |
| 159 | +
|
| 160 | + Returns |
| 161 | + ------- |
| 162 | + BoltzGenFuture |
| 163 | + A future object that can be used to retrieve the results of the design |
| 164 | + job upon completion. |
| 165 | + """ |
| 166 | + request = BoltzGenRequest( |
| 167 | + n=n, |
| 168 | + design_spec=design_spec, |
| 169 | + diffusion_batch_size=diffusion_batch_size, |
| 170 | + step_scale=step_scale, |
| 171 | + noise_scale=noise_scale, |
| 172 | + ) |
| 173 | + if structure_file is not None: |
| 174 | + if isinstance(structure_file, bytes): |
| 175 | + structure_text = structure_file.decode() |
| 176 | + elif isinstance(structure_file, str): |
| 177 | + structure_text = structure_file |
| 178 | + else: |
| 179 | + structure_text = structure_file.read().decode() |
| 180 | + request.structure_text = structure_text |
| 181 | + |
| 182 | + # Submit the job via the private API function |
| 183 | + job = _boltzgen_api_post( |
| 184 | + session=self.session, |
| 185 | + request=request, |
| 186 | + **kwargs, |
| 187 | + ) |
| 188 | + |
| 189 | + # Return the future object |
| 190 | + return BoltzGenFuture(session=self.session, job=job) |
| 191 | + |
| 192 | + predict = generate |
0 commit comments