aegis_sim.recording.fastarecorder
1import logging 2import pathlib 3 4from .recorder import Recorder 5from aegis_sim import variables 6from aegis_sim.parameterization import parametermanager 7from aegis_sim.utilities.fasta import encode_population_to_fasta 8from aegis_sim.utilities.funcs import skip 9 10 11class FastaRecorder(Recorder): 12 """Write the living population as a FASTA reference (one record per individual) 13 plus a sidecar JSON mapping for the round-trip decoder. 14 15 Output: /fasta/step{step}.genome.fasta + /fasta/step{step}.mapping.json 16 Rate parameter: FASTA_RATE (0 disables; still writes at the final step if a 17 non-zero rate is set or the rate is 0). 18 """ 19 20 def __init__(self, odir: pathlib.Path): 21 self.odir = odir / "fasta" 22 self.init_odir() 23 24 def write(self, population): 25 """ 26 # OUTPUT SPECIFICATION 27 path: /fasta/step{step}.genome.fasta and /fasta/step{step}.mapping.json 28 filetype: fasta + json 29 category: log 30 description: Per-individual genome FASTA (4-letter XOR-masked packing) and sidecar mapping for lossless decode back to the original bit array and architect-derived phenotypes. 31 trait granularity: individual 32 time granularity: snapshot 33 frequency parameter: FASTA_RATE 34 structure: FASTA (one record per individual) + JSON sidecar (encoding metadata + XOR mask). 35 """ 36 37 if parametermanager.parameters.FASTA_RATE <= 0: 38 return 39 40 step = variables.steps 41 should_skip = skip("FASTA_RATE") 42 is_first_step = step == 1 43 is_last_step = step == parametermanager.parameters.STEPS_PER_SIMULATION 44 45 if not (is_first_step or not should_skip or is_last_step): 46 return 47 48 if len(population) == 0: 49 return 50 51 mask_seed = parametermanager.parameters.FASTA_MASK_SEED 52 53 logging.debug(f"fasta recorded at step {step}.") 54 encode_population_to_fasta( 55 population=population, 56 output_dir=self.odir, 57 name=f"step{step}", 58 mask_seed=mask_seed, 59 )
12class FastaRecorder(Recorder): 13 """Write the living population as a FASTA reference (one record per individual) 14 plus a sidecar JSON mapping for the round-trip decoder. 15 16 Output: /fasta/step{step}.genome.fasta + /fasta/step{step}.mapping.json 17 Rate parameter: FASTA_RATE (0 disables; still writes at the final step if a 18 non-zero rate is set or the rate is 0). 19 """ 20 21 def __init__(self, odir: pathlib.Path): 22 self.odir = odir / "fasta" 23 self.init_odir() 24 25 def write(self, population): 26 """ 27 # OUTPUT SPECIFICATION 28 path: /fasta/step{step}.genome.fasta and /fasta/step{step}.mapping.json 29 filetype: fasta + json 30 category: log 31 description: Per-individual genome FASTA (4-letter XOR-masked packing) and sidecar mapping for lossless decode back to the original bit array and architect-derived phenotypes. 32 trait granularity: individual 33 time granularity: snapshot 34 frequency parameter: FASTA_RATE 35 structure: FASTA (one record per individual) + JSON sidecar (encoding metadata + XOR mask). 36 """ 37 38 if parametermanager.parameters.FASTA_RATE <= 0: 39 return 40 41 step = variables.steps 42 should_skip = skip("FASTA_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 mask_seed = parametermanager.parameters.FASTA_MASK_SEED 53 54 logging.debug(f"fasta recorded at step {step}.") 55 encode_population_to_fasta( 56 population=population, 57 output_dir=self.odir, 58 name=f"step{step}", 59 mask_seed=mask_seed, 60 )
Write the living population as a FASTA reference (one record per individual) plus a sidecar JSON mapping for the round-trip decoder.
Output: /fasta/step{step}.genome.fasta + /fasta/step{step}.mapping.json Rate parameter: FASTA_RATE (0 disables; still writes at the final step if a non-zero rate is set or the rate is 0).
def
write(self, population):
25 def write(self, population): 26 """ 27 # OUTPUT SPECIFICATION 28 path: /fasta/step{step}.genome.fasta and /fasta/step{step}.mapping.json 29 filetype: fasta + json 30 category: log 31 description: Per-individual genome FASTA (4-letter XOR-masked packing) and sidecar mapping for lossless decode back to the original bit array and architect-derived phenotypes. 32 trait granularity: individual 33 time granularity: snapshot 34 frequency parameter: FASTA_RATE 35 structure: FASTA (one record per individual) + JSON sidecar (encoding metadata + XOR mask). 36 """ 37 38 if parametermanager.parameters.FASTA_RATE <= 0: 39 return 40 41 step = variables.steps 42 should_skip = skip("FASTA_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 mask_seed = parametermanager.parameters.FASTA_MASK_SEED 53 54 logging.debug(f"fasta recorded at step {step}.") 55 encode_population_to_fasta( 56 population=population, 57 output_dir=self.odir, 58 name=f"step{step}", 59 mask_seed=mask_seed, 60 )
OUTPUT SPECIFICATION
path: /fasta/step{step}.genome.fasta and /fasta/step{step}.mapping.json filetype: fasta + json category: log description: Per-individual genome FASTA (4-letter XOR-masked packing) and sidecar mapping for lossless decode back to the original bit array and architect-derived phenotypes. trait granularity: individual time granularity: snapshot frequency parameter: FASTA_RATE structure: FASTA (one record per individual) + JSON sidecar (encoding metadata + XOR mask).