Skip to content

Commit

Permalink
fix merge in design
Browse files Browse the repository at this point in the history
  • Loading branch information
@CChahrour
CChahrour committed May 28, 2024
1 parent 9338a77 commit 7397902
Showing 1 changed file with 19 additions and 174 deletions.
193 changes: 19 additions & 174 deletions seqnado/design.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -484,10 +484,7 @@ def controls_performed(self) -> List[str]:
control.add(f.control_performed)
return list(control)


def query(
self, sample_name: str, full_experiment: bool = False
) -> Union[FastqSetIP, Dict[str, FastqSetIP]]:
def query(self, sample_name: str, full_experiment: bool = False) -> Union[FastqSetIP, IPExperiment]:
"""
Extracts a pair of fastq files from the design.
"""
Expand All @@ -497,50 +494,27 @@ def query(
)
is_control = False

experiment_files = dict()
is_control = False

experiment_files = dict()

if sample_name in ip_names or sample_name in control_names:
for experiment in self.experiments:
if experiment.ip_set_fullname == sample_name:
experiment_files["ip"] = experiment.ip
experiment_files["control"] = experiment.control

experiment_files["ip"] = experiment.ip
experiment_files["control"] = experiment.control

exp = experiment
break
elif (
experiment.has_control
and experiment.control_fullname == sample_name
):
is_control = True
experiment_files["ip"] = experiment.ip
experiment_files["control"] = experiment.control
is_control = True
experiment_files["ip"] = experiment.ip
experiment_files["control"] = experiment.control
else:
raise ValueError(f"Could not find sample with name {sample_name}")
exp = experiment
break

if full_experiment:
return experiment_files
else:
return (
experiment_files["ip"]
if not is_control
else experiment_files["control"]
)
raise ValueError(f"Could not find sample with name {sample_name}")


if full_experiment:
return experiment_files
return exp
else:
return (
experiment_files["ip"]
if not is_control
else experiment_files["control"]
)
return exp.ip if not is_control else exp.control

@classmethod
def from_fastq_files(cls, fq: List[Union[str, pathlib.Path]], **kwargs):
Expand Down Expand Up @@ -655,9 +629,10 @@ def to_dataframe(self) -> pd.DataFrame:
experiment.control.r1.path if experiment.control else None
),
"control_r2": (
experiment.control.r2.path if experiment.control and experiment.control.r2 else None
experiment.control.r2.path if experiment.control else None
),
}

for k, v in metadata.model_dump(exclude_none=True).items():
row[k] = v

Expand Down Expand Up @@ -758,7 +733,7 @@ def from_design(
subset_value: Optional[List[str]] = None,
include_controls: bool = False,
):

if isinstance(design, Design):
df = (
design.to_dataframe()
Expand All @@ -785,11 +760,13 @@ def from_design(
)
df = pd.concat([df_ip, df_control])


if subset_value:
df = df.query(f"{subset_column} in {subset_value}")

samples = df.index.tolist()


reference_sample = reference_sample or df.index[0]

return cls(
Expand Down Expand Up @@ -901,20 +878,7 @@ class BigWigFiles(BaseModel):
assay: Literal["ChIP", "ATAC", "RNA", "SNP"]
names: List[str]
pileup_method: Union[
Literal["deeptools", "homer", False],
List[
Literal[
"deeptools",
"homer",
]
],
Literal["deeptools", "homer", False],
List[
Literal[
"deeptools",
"homer",
]
],
Literal["deeptools", "homer"], List[Literal["deeptools", "homer"]]
] = None
make_bigwigs: bool = False
scale_method: Optional[Literal["cpm", "rpkm", "spikein", "csaw", "merged"]] = None
Expand All @@ -926,9 +890,7 @@ def model_post_init(self, __context: Any) -> None:
self.pileup_method = [self.pileup_method]

if self.include_unscaled and not self.scale_method:
self.scale_method = [
"unscaled",
]
self.scale_method = ["unscaled",]
elif self.include_unscaled and self.scale_method:
self.scale_method = ["unscaled", self.scale_method]
else:
Expand Down Expand Up @@ -995,7 +957,6 @@ def files(self) -> List[str]:
class HeatmapFiles(BaseModel):
assay: Literal["ChIP", "ATAC", "RNA", "SNP"]
make_heatmaps: bool = False
make_heatmaps: bool = False

@property
def heatmap_files(self) -> List[str]:
Expand All @@ -1011,10 +972,6 @@ def files(self) -> List[str]:
return self.heatmap_files
else:
return []
if self.make_heatmaps:
return self.heatmap_files
else:
return []


class HubFiles(BaseModel):
Expand Down Expand Up @@ -1068,15 +1025,9 @@ class Output(BaseModel):
sample_names: List[str]

make_bigwigs: bool = False
pileup_method: Union[
Literal["deeptools", "homer", False],
List[Literal["deeptools", "homer"]],
pileup_method: Union[
Literal["deeptools", "homer", False],
List[Literal["deeptools", "homer"]],
pileup_method: Optional[
Union[Literal["deeptools", "homer"], List[Literal["deeptools", "homer"]]]
] = None


scale_method: Optional[Literal["cpm", "rpkm", "spikein", "csaw"]] = None

make_heatmaps: bool = False
Expand Down Expand Up @@ -1289,10 +1240,8 @@ def peaks(self):
s
for s in self.sample_names
if not any([c in s for c in self.control_names])
if not any([c in s for c in self.control_names])
]


pcf_samples = PeakCallingFiles(
assay=self.assay,
names=ip_sample_names,
Expand Down Expand Up @@ -1338,108 +1287,4 @@ def files(self) -> List[str]:
if file_list:
files.extend(file_list)

return files


class SNPOutput(Output):
assay: Literal["SNP"]
call_snps: bool = False
sample_names: List[str]
make_ucsc_hub: bool = False
snp_calling_method: Optional[
Union[
Literal["bcftools", "deepvariant", False],
List[Literal["bcftools", "deepvariant"]],
]
] = None

@property
def design(self):
return ["seqnado_output/design.csv"]

@property
def snp_files(self) -> List[str]:
if self.call_snps:
return expand(
"seqnado_output/variant/{method}/{sample}.vcf.gz",
sample=self.sample_names,
method=self.snp_calling_method,
)
else:
return []

@computed_field
@property
def files(self) -> List[str]:
files = []
files.extend(
QCFiles(
assay=self.assay,
fastq_screen=self.fastq_screen,
library_complexity=self.library_complexity,
).files
)

for file_list in (
self.snp_files,
self.design,
):
if file_list:
files.extend(file_list)

if self.call_snps:
files.append(self.snp_files)

return files


class SNPOutput(Output):
assay: Literal["SNP"]
call_snps: bool = False
sample_names: List[str]
make_ucsc_hub: bool = False
snp_calling_method: Optional[
Union[
Literal["bcftools", "deepvariant", False],
List[Literal["bcftools", "deepvariant"]],
]
] = None

@property
def design(self):
return ["seqnado_output/design.csv"]

@property
def snp_files(self) -> List[str]:
if self.call_snps:
return expand(
"seqnado_output/variant/{method}/{sample}.vcf.gz",
sample=self.sample_names,
method=self.snp_calling_method,
)
else:
return []

@computed_field
@property
def files(self) -> List[str]:
files = []
files.extend(
QCFiles(
assay=self.assay,
fastq_screen=self.fastq_screen,
library_complexity=self.library_complexity,
).files
)

for file_list in (
self.snp_files,
self.design,
):
if file_list:
files.extend(file_list)

if self.call_snps:
files.append(self.snp_files)

return files
return files

0 comments on commit 7397902

Please sign in to comment.