Source code for openprotein.app.services.assaydata
import io
import pandas as pd
from openprotein.api import assaydata
from openprotein.app.models import AssayDataset, AssayMetadata
from openprotein.base import APISession
[docs]
class DataAPI:
"""API interface for calling AssayData endpoints"""
[docs]
def __init__(self, session: APISession):
self.session = session
[docs]
def list(self) -> list[AssayDataset]:
"""
List all assay datasets.
Returns
-------
List[AssayDataset]
List of all assay datasets.
"""
metadata = assaydata.assaydata_list(self.session)
return [AssayDataset(self.session, x) for x in metadata]
[docs]
def create(
self, table: pd.DataFrame, name: str, description: str | None = None
) -> AssayDataset:
"""
Create a new assay dataset.
Parameters
----------
table : pd.DataFrame
DataFrame containing the assay data.
name : str
Name of the assay dataset.
description : str, optional
Description of the assay dataset, by default None.
Returns
-------
AssayDataset
Created assay dataset.
"""
stream = io.BytesIO()
table.to_csv(stream, index=False)
stream.seek(0)
metadata = assaydata.assaydata_post(
self.session, stream, name, assay_description=description
)
metadata.sequence_length = len(table["sequence"].values[0])
return AssayDataset(self.session, metadata)
[docs]
def get(self, assay_id: str, verbose: bool = False) -> AssayDataset:
"""
Get an assay dataset by its ID.
Parameters
----------
assay_id : str
ID of the assay dataset.
Returns
-------
AssayDataset
Assay dataset with the specified ID.
Raises
------
KeyError
If no assay dataset with the given ID is found.
"""
return AssayDataset(
self.session, assaydata.get_assay_metadata(self.session, assay_id)
)
[docs]
def load_assay(self, assay_id: str) -> AssayDataset:
"""
Reload a Submitted job to resume from where you left off!
Parameters
----------
assay_id : str
The identifier of the job whose details are to be loaded.
Returns
-------
Job
Job
Raises
------
HTTPError
If the request to the server fails.
InvalidJob
If the Job is of the wrong type
"""
metadata = self.get(assay_id)
# if job_details.job_type != JobType.train:
# raise InvalidJob(f"Job {job_id} is not of type {JobType.train}")
return AssayDataset(
self.session,
metadata,
)
def __len__(self) -> int:
"""
Get the number of assay datasets.
Returns
-------
int
Number of assay datasets.
"""
return len(self.list())