aegis_sim.recording.vcfrecorder
1import logging 2import pathlib 3 4from .recorder import Recorder 5from aegis_sim import variables 6from aegis_sim.parameterization import parametermanager 7from aegis_sim.utilities.vcf import encode_population_to_vcf 8from aegis_sim.utilities.funcs import skip 9 10 11class VCFRecorder(Recorder): 12 """Write the living population as a VCF (one row per genome bit, one column 13 per individual) at the configured rate. 14 15 Output: /vcf/step{step}.vcf 16 Rate parameter: VCF_RATE (0 disables; still writes at the final step when 17 VCF_RATE > 0). 18 Only the composite architecture is supported. 19 """ 20 21 def __init__(self, odir: pathlib.Path): 22 self.odir = odir / "vcf" 23 self.init_odir() 24 25 def write(self, population): 26 """ 27 # OUTPUT SPECIFICATION 28 path: /vcf/step{step}.vcf 29 filetype: vcf (v4.2) 30 category: log 31 description: Per-individual diploid genotypes at every genome bit, formatted as a self-contained VCF (header records architecture metadata for decoding). Compatible with PLINK, vcftools, ADMIXTOOLS, scikit-allel. 32 trait granularity: bit (one row per logical bit position) 33 time granularity: snapshot 34 frequency parameter: VCF_RATE 35 structure: VCF v4.2. 36 """ 37 38 if parametermanager.parameters.VCF_RATE <= 0: 39 return 40 41 step = variables.steps 42 should_skip = skip("VCF_RATE") 43 is_first_step = step == 1 44 is_last_step = step == parametermanager.parameters.STEPS_PER_SIMULATION 45 46 if not (is_first_step or not should_skip or is_last_step): 47 return 48 49 if len(population) == 0: 50 return 51 52 logging.debug(f"vcf recorded at step {step}.") 53 encode_population_to_vcf( 54 population=population, 55 output_dir=self.odir, 56 name=f"step{step}", 57 )
12class VCFRecorder(Recorder): 13 """Write the living population as a VCF (one row per genome bit, one column 14 per individual) at the configured rate. 15 16 Output: /vcf/step{step}.vcf 17 Rate parameter: VCF_RATE (0 disables; still writes at the final step when 18 VCF_RATE > 0). 19 Only the composite architecture is supported. 20 """ 21 22 def __init__(self, odir: pathlib.Path): 23 self.odir = odir / "vcf" 24 self.init_odir() 25 26 def write(self, population): 27 """ 28 # OUTPUT SPECIFICATION 29 path: /vcf/step{step}.vcf 30 filetype: vcf (v4.2) 31 category: log 32 description: Per-individual diploid genotypes at every genome bit, formatted as a self-contained VCF (header records architecture metadata for decoding). Compatible with PLINK, vcftools, ADMIXTOOLS, scikit-allel. 33 trait granularity: bit (one row per logical bit position) 34 time granularity: snapshot 35 frequency parameter: VCF_RATE 36 structure: VCF v4.2. 37 """ 38 39 if parametermanager.parameters.VCF_RATE <= 0: 40 return 41 42 step = variables.steps 43 should_skip = skip("VCF_RATE") 44 is_first_step = step == 1 45 is_last_step = step == parametermanager.parameters.STEPS_PER_SIMULATION 46 47 if not (is_first_step or not should_skip or is_last_step): 48 return 49 50 if len(population) == 0: 51 return 52 53 logging.debug(f"vcf recorded at step {step}.") 54 encode_population_to_vcf( 55 population=population, 56 output_dir=self.odir, 57 name=f"step{step}", 58 )
Write the living population as a VCF (one row per genome bit, one column per individual) at the configured rate.
Output: /vcf/step{step}.vcf Rate parameter: VCF_RATE (0 disables; still writes at the final step when VCF_RATE > 0). Only the composite architecture is supported.
def
write(self, population):
26 def write(self, population): 27 """ 28 # OUTPUT SPECIFICATION 29 path: /vcf/step{step}.vcf 30 filetype: vcf (v4.2) 31 category: log 32 description: Per-individual diploid genotypes at every genome bit, formatted as a self-contained VCF (header records architecture metadata for decoding). Compatible with PLINK, vcftools, ADMIXTOOLS, scikit-allel. 33 trait granularity: bit (one row per logical bit position) 34 time granularity: snapshot 35 frequency parameter: VCF_RATE 36 structure: VCF v4.2. 37 """ 38 39 if parametermanager.parameters.VCF_RATE <= 0: 40 return 41 42 step = variables.steps 43 should_skip = skip("VCF_RATE") 44 is_first_step = step == 1 45 is_last_step = step == parametermanager.parameters.STEPS_PER_SIMULATION 46 47 if not (is_first_step or not should_skip or is_last_step): 48 return 49 50 if len(population) == 0: 51 return 52 53 logging.debug(f"vcf recorded at step {step}.") 54 encode_population_to_vcf( 55 population=population, 56 output_dir=self.odir, 57 name=f"step{step}", 58 )
OUTPUT SPECIFICATION
path: /vcf/step{step}.vcf filetype: vcf (v4.2) category: log description: Per-individual diploid genotypes at every genome bit, formatted as a self-contained VCF (header records architecture metadata for decoding). Compatible with PLINK, vcftools, ADMIXTOOLS, scikit-allel. trait granularity: bit (one row per logical bit position) time granularity: snapshot frequency parameter: VCF_RATE structure: VCF v4.2.