Push commit

Author: CodingBash

crispr-ambiguous-mapping/crispr_ambiguous_mapping/mapping/main_mapping.py

@@ -25,7 +25,7 @@
 from ..processing import crispr_guide_counting
 from ..processing import crispr_sequence_encoding
 from ..parsing import reporter_umitools_fastq_parsing, reporter_standard_fastq_parsing
-from ..models.mapping_models import WhitelistReporterCountsResult, SampleWhitelistReporterCountsResult, GeneralGuideCountType
+from ..models.mapping_models import WhitelistReporterCountsResult, GeneralGuideCountType
 
 # 
 # Deprecated 
@@ -159,7 +159,7 @@ def get_whitelist_reporter_counts_from_fastq(whitelist_guide_reporter_df: Option
                                                        protospacer_hamming_threshold_strict: Optional[int] = None, 
 
                                                        store_intermediates: bool = False,
-                                                       cores: int=1) -> Union[WhitelistReporterCountsResult, SampleWhitelistReporterCountsResult]:
+                                                       cores: int=1) -> WhitelistReporterCountsResult:
     # Input parameter validation checks
 
     protospacer_pattern_regex = None if ((protospacer_pattern_regex is not None) and  (protospacer_pattern_regex.strip() == "")) else protospacer_pattern_regex

crispr-ambiguous-mapping/crispr_ambiguous_mapping/models/mapping_models.py

@@ -255,19 +255,7 @@ class CountInput:
 @dataclass
 class WhitelistReporterCountsResult:
     all_match_set_whitelist_reporter_counter_series_results: AllMatchSetWhitelistReporterCounterSeriesResults
-    observed_guide_reporter_umi_counts_inferred: GeneralMappingInferenceDict
+    observed_guide_reporter_umi_counts_inferred: Union[GeneralMappingInferenceDict, DefaultDict[str, GeneralMappingInferenceDict]]
     quality_control_result: QualityControlResult
     count_input: CountInput
 
-@dataclass
-class SampleWhitelistReporterCountsResult:
-    all_match_set_whitelist_reporter_counter_series_results_all_samples: DefaultDict[str, AllMatchSetWhitelistReporterCounterSeriesResults]
-    observed_guide_reporter_umi_counts_inferred_all_samples: DefaultDict[str, GeneralMappingInferenceDict]
-    quality_control_result_all_samples: DefaultDict[str, GeneralMappingInferenceDict]
-    count_input: CountInput
-
-#
-# Types
-#
-
-

crispr-ambiguous-mapping/crispr_ambiguous_mapping/processing/crispr_count_processing.py

@@ -15,8 +15,25 @@ def check_match_result_non_error(match_result):
     return False if match_result is None else match_result.error is None # If match_result is None, treat as error. If match_result is not None, but error is None, then non_error
 
 # Filter dict with observed sequence inference results for only those that do not contain any mapping errors
-def get_non_error_dict(observed_guide_reporter_umi_counts_inferred: GeneralMappingInferenceDict, attribute_name: str) -> MatchSetWhitelistReporterObservedSequenceCounterSeriesResults:
-    return {observed_guide_reporter_key: observed_guide_reporter_umi_counts_inferred_value for observed_guide_reporter_key, observed_guide_reporter_umi_counts_inferred_value in observed_guide_reporter_umi_counts_inferred.items() if check_match_result_non_error(getattr(observed_guide_reporter_umi_counts_inferred_value.inferred_value, attribute_name))}
+def get_non_error_dict(observed_guide_reporter_umi_counts_inferred: Union[GeneralMappingInferenceDict, DefaultDict[str, GeneralMappingInferenceDict]], attribute_name: str) -> MatchSetWhitelistReporterObservedSequenceCounterSeriesResults:
+    """
+    Return a dict of all entries that are non-error.
+    Works with both sample-barcode (nested dict) and no-sample-barcode (flat dict) cases.
+    """
+    non_error_dict = {}
+
+    for key, val in observed_guide_reporter_umi_counts_inferred.items():
+        if hasattr(val, "inferred_value"):  # flat dict
+            attr = getattr(val.inferred_value, attribute_name, None)
+            if check_match_result_non_error(attr):
+                non_error_dict[key] = val
+        else:  # nested dict (sample-barcode case)
+            for inner_key, inner_val in val.items():
+                attr = getattr(inner_val.inferred_value, attribute_name, None)
+                if check_match_result_non_error(attr):
+                    non_error_dict[(key, inner_key)] = inner_val  # tuple key: (sample, guide)
+
+    return non_error_dict
 
 #
 # Given the datastructure containing the inference results "observed_guide_reporter_umi_counts_inferred", iterate through the entire datastructure to generate

crispr-ambiguous-mapping/crispr_ambiguous_mapping/processing/crispr_editing_processing.py

@@ -26,7 +26,7 @@
 from .crispr_count_processing import get_counterseries_all_results
 from ..quality_control.crispr_mapping_quality_control import perform_counts_quality_control
 from ..models.mapping_models import GeneralGuideCountType, GeneralMappingInferenceDict
-from ..models.mapping_models import AllMatchSetWhitelistReporterCounterSeriesResults, WhitelistReporterCountsResult, SampleWhitelistReporterCountsResult, InferenceResult, CountInput, QualityControlResult
+from ..models.mapping_models import AllMatchSetWhitelistReporterCounterSeriesResults, WhitelistReporterCountsResult, InferenceResult, CountInput, QualityControlResult
 
 
 # TODO: There will probably be some type errors with the DefaultDict when testing on non UMI (since it requires CounterType), so make sure to test with different variations of inputs
@@ -41,7 +41,7 @@ def get_whitelist_reporter_counts_with_umi(observed_guide_reporter_umi_counts: G
                                            surrogate_hamming_threshold_strict: Optional[int] = 2, 
                                            guide_barcode_hamming_threshold_strict: Optional[int] = 2, 
                                            store_intermediates: bool = False,
-                                           cores: int=1) -> Union[WhitelistReporterCountsResult, SampleWhitelistReporterCountsResult]:
+                                           cores: int=1) -> WhitelistReporterCountsResult:
 
     # Generate whitelist dataframe based on all observed sequences if none provided
     if whitelist_guide_reporter_df is None:
@@ -163,8 +163,12 @@ def pad_series(series):
     print(f"Mapping inference results of length {len(inferred_true_reporter_sequences)} to the result object")
     # Some organization: Map the inferred result of each observed sequence to a dict with the inferred result and correspoding count
 
+
+
+    # NOTE 20251031: This may be able to be removed
     if contains_sample_barcode:
         observed_guide_reporter_umi_counts_inferred_all_samples: DefaultDict[str, GeneralMappingInferenceDict] = defaultdict(lambda: defaultdict(dict))
+        
         # Add all cell_barcodes
         for observed_guide_reporter_key_index, observed_guide_reporter_key in enumerate(observed_guide_reporter_list): # Iterate through each observed guide key
             observed_guide_reporter_cell_counts = observed_guide_reporter_umi_counts[observed_guide_reporter_key]
@@ -185,19 +189,12 @@ def pad_series(series):
         # GET THE MAPPED COUNT SERIES BASED ON THE INFERENCE RESULTS
         print("Prepare the processed count series ")
         all_cell_barcodes: List[str] = list(observed_guide_reporter_umi_counts_inferred_all_samples.keys())
-        all_match_set_whitelist_reporter_counter_series_results_all_samples: DefaultDict[str, AllMatchSetWhitelistReporterCounterSeriesResults] = defaultdict(AllMatchSetWhitelistReporterCounterSeriesResults)
-        quality_control_result_all_samples: DefaultDict[str, QualityControlResult] = defaultdict(QualityControlResult)
-        for cell_barcode_i, cell_barcode in enumerate(all_cell_barcodes):
-            observed_guide_reporter_umi_counts_inferred_per_sample = observed_guide_reporter_umi_counts_inferred_all_samples[cell_barcode]
-            all_match_set_whitelist_reporter_counter_series_results_per_sample = get_counterseries_all_results(observed_guide_reporter_umi_counts_inferred_per_sample, whitelist_guide_reporter_df, contains_guide_barcode, contains_guide_surrogate, contains_guide_umi)
-            quality_control_result_per_sample: QualityControlResult = perform_counts_quality_control(observed_guide_reporter_umi_counts_inferred_per_sample, contains_guide_umi, contains_guide_surrogate, contains_guide_barcode)
-
-            all_match_set_whitelist_reporter_counter_series_results_all_samples[cell_barcode] = all_match_set_whitelist_reporter_counter_series_results_per_sample
-            quality_control_result_all_samples[cell_barcode] = quality_control_result_per_sample
-
-            if cell_barcode_i % 2500 == 0:
-                print(f"- Processed cell_barcode {cell_barcode_i} out of {len(all_cell_barcodes)}")
-
+        
+        all_match_set_whitelist_reporter_counter_series_results: AllMatchSetWhitelistReporterCounterSeriesResults
+        quality_control_result: QualityControlResult
+        
+        all_match_set_whitelist_reporter_counter_series_results = get_counterseries_all_results(observed_guide_reporter_umi_counts_inferred_all_samples, whitelist_guide_reporter_df, contains_guide_barcode, contains_guide_surrogate, contains_guide_umi, contains_sample_barcode)
+        quality_control_result: QualityControlResult = perform_counts_quality_control(observed_guide_reporter_umi_counts_inferred_all_samples, contains_guide_umi, contains_guide_surrogate, contains_guide_barcode, contains_sample_barcode)
 
         count_input= CountInput(whitelist_guide_reporter_df=whitelist_guide_reporter_df,
             contains_surrogate=contains_guide_surrogate,
@@ -208,10 +205,10 @@ def pad_series(series):
             surrogate_hamming_threshold_strict=surrogate_hamming_threshold,
             guide_barcode_hamming_threshold_strict=guide_barcode_hamming_threshold)
 
-        return SampleWhitelistReporterCountsResult(all_match_set_whitelist_reporter_counter_series_results_all_samples=all_match_set_whitelist_reporter_counter_series_results_all_samples,
-                                                    observed_guide_reporter_umi_counts_inferred_all_samples=observed_guide_reporter_umi_counts_inferred_all_samples, 
-                                                    quality_control_result_all_samples=quality_control_result_all_samples, 
-                                                    count_input=count_input)
+        return WhitelistReporterCountsResult(all_match_set_whitelist_reporter_counter_series_results=all_match_set_whitelist_reporter_counter_series_results, 
+                                             observed_guide_reporter_umi_counts_inferred=observed_guide_reporter_umi_counts_inferred_all_samples, 
+                                             quality_control_result=quality_control_result, 
+                                             count_input=count_input)
     else:    
 
         observed_guide_reporter_umi_counts_inferred: GeneralMappingInferenceDict = defaultdict(dict)
@@ -229,13 +226,13 @@ def pad_series(series):
         # GET THE MAPPED COUNT SERIES BASED ON THE INFERENCE RESULTS
         print("Prepare the processed count series ")
         # Count
-        all_match_set_whitelist_reporter_counter_series_results = get_counterseries_all_results(observed_guide_reporter_umi_counts_inferred, whitelist_guide_reporter_df, contains_guide_barcode, contains_guide_surrogate, contains_guide_umi)
+        all_match_set_whitelist_reporter_counter_series_results = get_counterseries_all_results(observed_guide_reporter_umi_counts_inferred, whitelist_guide_reporter_df, contains_guide_barcode, contains_guide_surrogate, contains_guide_umi, contains_sample_barcode)
 
         after_counterseries_time = datetime.now()
         print(f"{(after_counterseries_time-after_inference_processing_time).seconds} seconds for counter series generation")
 
         print("Preparing quality control")
-        quality_control_result: QualityControlResult = perform_counts_quality_control(observed_guide_reporter_umi_counts_inferred, contains_guide_umi, contains_guide_surrogate, contains_guide_barcode)
+        quality_control_result: QualityControlResult = perform_counts_quality_control(observed_guide_reporter_umi_counts_inferred, contains_guide_umi, contains_guide_surrogate, contains_guide_barcode, contains_sample_barcode)
 
         after_qualitycontrol_time = datetime.now()
         print(f"{(after_qualitycontrol_time-after_counterseries_time).seconds} seconds for quality control")