Source code for overture_song.entities

import json
from typing import Any, Type

from dataclasses import dataclass

from overture_song.validation import Validatable
from overture_song.utils import Builder, default_value
from typing import List
from dataclasses import is_dataclass, asdict
from overture_song.utils import check_type, check_state


[docs]class Entity(object):
[docs] def to_json(self): return json.dumps(self.to_dict(), indent=4)
[docs] def to_dict(self): if is_dataclass(self): return asdict(self) else: raise NotImplemented("not implemented for non-dataclass object")
def __str__(self): return self.to_json()
[docs]@dataclass(frozen=False) class Metadata(Entity): info: dict = None def __post_init__(self): self.info = {}
[docs] def set_info(self, key: str, value: Any): self.info[key] = value
[docs] def add_info(self, data: dict): if data is None: return self.info.update(data)
[docs]@dataclass(frozen=False) class Study(Metadata, Validatable): studyId: str = None name: str = None organization: str = None description: str = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, studyId, name=None, description=None, organization=None): s = Study() s.studyId = studyId s.name = name s.description = description s.organization = organization return s
[docs] @classmethod def create_from_raw(cls, study_obj): return Study.create( study_obj.studyId, name=study_obj.name, description=study_obj.description, organization=study_obj.organization)
[docs]@dataclass(frozen=False) class File(Metadata, Validatable): objectId: str = None analysisId: str = None fileName: str = None studyId: str = None fileSize: int = -1 fileType: str = None fileMd5sum: str = None fileAccess: str = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, fileName, fileSize, fileType, fileMd5sum, fileAccess, studyId=None, analysisId=None, objectId=None, info=None): f = File() f.objectId = objectId f.analysisId = analysisId f.studyId = studyId f.fileType = fileType f.fileSize = fileSize f.info = default_value(info, {}) f.fileMd5sum = fileMd5sum f.fileAccess = fileAccess f.fileName = fileName return f
[docs]@dataclass(frozen=False) class Sample(Metadata, Validatable): sampleId: str = None specimenId: str = None sampleSubmitterId: str = None sampleType: str = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, specimenId, sampleSubmitterId, sampleType, sampleId=None, info=None): s = Sample() s.info = default_value(info, {}) s.specimenId = specimenId s.sampleType = sampleType s.sampleSubmitterId = sampleSubmitterId s.sampleId = sampleId return s
[docs]@dataclass(frozen=False) class Specimen(Metadata, Validatable): specimenId: str = None donorId: str = None specimenSubmitterId: str = None specimenClass: str = None specimenType: str = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, donorId, specimenSubmitterId, specimenClass, specimenType, specimenId=None, info=None): s = Specimen() s.info = default_value(info, {}) s.specimenId = specimenId s.donorId = donorId s.specimenType = specimenType s.specimenClass = specimenClass s.specimenSubmitterId = specimenSubmitterId return s
[docs]@dataclass(frozen=False) class Donor(Metadata, Validatable): donorId: str = None donorSubmitterId: str = None studyId: str = None donorGender: str = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, donorSubmitterId, studyId, donorGender, donorId=None, info=None): d = Donor() d.donorId = donorId d.info = default_value(info, {}) d.studyId = studyId d.donorSubmitterId = donorSubmitterId d.donorGender = donorGender return d
[docs]@dataclass(frozen=False) class CompositeEntity(Sample): specimen: Type[Specimen] = None donor: Type[Donor] = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def base_on_sample(cls, sample): s = CompositeEntity() s.sampleId = sample.sampleId s.sampleSubmitterId = sample.sampleSubmitterId s.sampleType = sample.sampleType s.info = sample.info s.specimenId = sample.specimenId return s
[docs] @classmethod def create(cls, donor, specimen, sample): c = CompositeEntity.base_on_sample(sample) check_type(donor, Donor) check_type(specimen, Specimen) c.donor = donor c.specimen = specimen return c
[docs]@dataclass(frozen=False) class Experiment(Metadata): pass
[docs]@dataclass(frozen=False) class VariantCall(Experiment, Validatable): analysisId: str = None variantCallingTool: str = None matchedNormalSampleSubmitterId: str = None
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, variantCallingTool, matchedNormalSampleSubmitterId, analysisId=None): s = VariantCall() s.analysisId = analysisId s.variantCallingTool = variantCallingTool s.matchedNormalSampleSubmitterId = matchedNormalSampleSubmitterId return s
[docs]@dataclass(frozen=False) class SequencingRead(Experiment, Validatable): analysisId: str = None aligned: bool = None alignmentTool: str = None insertSize: int = None libraryStrategy: str = None pairedEnd: bool = None referenceGenome: str = None
[docs] @classmethod def builder(cls): return Builder(Analysis)
[docs] def validate(self): raise NotImplemented("not implemented")
[docs] @classmethod def create(cls, aligned, alignmentTool, insertSize, libraryStrategy, pairedEnd, referenceGenome, analysisId=None): s = SequencingRead() s.alignmentTool = alignmentTool s.aligned = aligned s.analysisId = analysisId s.libraryStrategy = libraryStrategy s.insertSize = insertSize s.pairedEnd = pairedEnd s.referenceGenome = referenceGenome return s
[docs]@dataclass(frozen=False) class Analysis(Entity): analysisId: str = None study: str = None analysisState: str = "UNPUBLISHED" # TODO: add typing to this. should be a list of type Sample sample: List[CompositeEntity] = None # TODO: add typing to this. should be a list of type File file: List[File] = None def __post_init__(self): self.sample = [] self.file = []
[docs] @classmethod def builder(cls): return Builder(Analysis)
[docs] @classmethod def from_json(cls, json_string): pass
[docs]@dataclass(frozen=False) class SequencingReadAnalysis(Analysis, Validatable): analysisType: str = "sequencingRead" # TODO: add typing to this. should be a list of type File experiment: Type[SequencingRead] = None
[docs] @classmethod def create(cls, experiment, sample, file, analysisId=None, study=None, analysisState="UNPUBLISHED", info=None): check_type(experiment, SequencingRead) check_state(sample is not None and isinstance(sample, list) and len(sample) > 0, "Atleast one sample must be defined") check_state(file is not None and isinstance(file, list) and len(file) > 0, "Atleast one file must be defined") for s in sample: check_type(s, CompositeEntity) for f in file: check_type(f, File) s = SequencingReadAnalysis() s.sample = sample s.file = file s.info = default_value(info, {}) s.experiment = experiment s.analysisId = analysisId s.study = study s.analysisState = analysisState return s
[docs] def validate(self): raise NotImplemented("not implemented")
[docs]@dataclass(frozen=False) class VariantCallAnalysis(Analysis, Validatable): analysisType: str = 'variantCall' # TODO: add typing to this. should be a list of type File experiment: Type[VariantCall] = None
[docs] @classmethod def create(cls, experiment, sample, file, analysisId=None, study=None, analysisState="UNPUBLISHED", info=None): check_type(experiment, VariantCall) check_state(sample is not None and isinstance(sample, list) and len(sample) > 0, "Atleast one sample must be defined") check_state(file is not None and isinstance(file, list) and len(file) > 0, "Atleast one file must be defined") for s in sample: check_type(s, CompositeEntity) for f in file: check_type(f, File) s = VariantCallAnalysis() s.experiment = experiment s.analysisId = analysisId s.study = study s.analysisState = analysisState s.sample = sample s.file = file s.info = default_value(info, {}) return s
[docs] def validate(self): raise NotImplemented("not implemented")