From a7df73f64165d12f5f8a5255f131fc0bd6f902c9 Mon Sep 17 00:00:00 2001 From: Jeremy Marcus Date: Fri, 31 Jul 2026 14:50:03 -0700 Subject: [PATCH 1/4] stopgap fix to speed up regions_from_list for pileup counting --- dimelo/load_processed.py | 139 +++++++++++++++++++++++++-------------- 1 file changed, 91 insertions(+), 48 deletions(-) diff --git a/dimelo/load_processed.py b/dimelo/load_processed.py index fbdd6cc2..79b69b78 100644 --- a/dimelo/load_processed.py +++ b/dimelo/load_processed.py @@ -165,63 +165,106 @@ def pileup_counts_from_bedmethyl( """ parsed_motif = utils.ParsedMotif(motif) - regions_dict = utils.regions_dict_from_input(regions, window_size) chunks_list = utils.process_chunks_from_regions_dict( regions_dict, chunk_size=chunk_size ) - cores_to_run = utils.cores_to_run(cores) - # Initialize shared memory as length-one numpy arrays to make it easy to map to buffer in subprocesses - shm_valid = shared_memory.SharedMemory( - create=True, size=np.dtype(np.int32).itemsize - ) - shm_modified = shared_memory.SharedMemory( - create=True, size=np.dtype(np.int32).itemsize - ) - - manager = multiprocessing.Manager() - lock = manager.Lock() - - with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: - futures = [ - executor.submit( - pileup_counts_process_chunk, - bedmethyl_file, - parsed_motif, - chunk, - shm_modified.name, - shm_valid.name, - lock, - single_strand, - ) - for chunk in chunks_list - ] - for future in tqdm( - concurrent.futures.as_completed(futures), - total=len(futures), + # If one core requested, do not spawn processes + # This removes unnecessary overhead and avoids nested process generation (i.e. from regions_to_list) + if cores_to_run == 1: + # TODO: This is a quick and dirty version as a proof of concept; should refactor proces_chunks_from_regions_dict instead + source_tabix = pysam.TabixFile(str(bedmethyl_file)) + valid_base_count = 0 + modified_base_count = 0 + for chunk in tqdm( + chunks_list, + total=len(chunks_list), disable=quiet, desc="Loading data", - leave=False, + leave=False ): - try: - future.result() - except Exception as err: - raise RuntimeError("pileup_counts_process_chunk failed.") from err - - # Directly convert shared memory buffers to integers - modified_base_count = int.from_bytes( - shm_modified.buf[:4], byteorder="little", signed=True - ) - valid_base_count = int.from_bytes( - shm_valid.buf[:4], byteorder="little", signed=True - ) - # Close and unlink shared memory - not fully handled by garbage collection otherwise - shm_modified.close() - shm_modified.unlink() - shm_valid.close() - shm_valid.unlink() + chromosome = chunk["chromosome"] + subregion_start = chunk["subregion_start"] + subregion_end = chunk["subregion_end"] + strand = chunk["strand"] + + valid_base_subregion_counts = 0 + modified_base_subregion_counts = 0 + + # tabix throws and error if the contig is not present + # by the current design, this should be silent + if chromosome in source_tabix.contigs: + for row in source_tabix.fetch( + chromosome, max(subregion_start, 0), subregion_end + ): + ( + keep_basemod, + _, + modified_in_row, + valid_in_row, + ) = process_pileup_row( + row=row, + parsed_motif=parsed_motif, + region_strand=strand, + single_strand=single_strand, + ) + if keep_basemod: + valid_base_subregion_counts += valid_in_row + modified_base_subregion_counts += modified_in_row + valid_base_count += valid_base_subregion_counts + modified_base_count += modified_base_subregion_counts + else: + # Initialize shared memory as length-one numpy arrays to make it easy to map to buffer in subprocesses + shm_valid = shared_memory.SharedMemory( + create=True, size=np.dtype(np.int32).itemsize + ) + shm_modified = shared_memory.SharedMemory( + create=True, size=np.dtype(np.int32).itemsize + ) + + manager = multiprocessing.Manager() + lock = manager.Lock() + + with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: + futures = [ + executor.submit( + pileup_counts_process_chunk, + bedmethyl_file, + parsed_motif, + chunk, + shm_modified.name, + shm_valid.name, + lock, + single_strand, + ) + for chunk in chunks_list + ] + for future in tqdm( + concurrent.futures.as_completed(futures), + total=len(futures), + disable=quiet, + desc="Loading data", + leave=False, + ): + try: + future.result() + except Exception as err: + raise RuntimeError("pileup_counts_process_chunk failed.") from err + + # Directly convert shared memory buffers to integers + modified_base_count = int.from_bytes( + shm_modified.buf[:4], byteorder="little", signed=True + ) + valid_base_count = int.from_bytes( + shm_valid.buf[:4], byteorder="little", signed=True + ) + # Close and unlink shared memory - not fully handled by garbage collection otherwise + shm_modified.close() + shm_modified.unlink() + shm_valid.close() + shm_valid.unlink() return modified_base_count, valid_base_count From b069ed89cc5e6801cca0867574e7968a8fd04241 Mon Sep 17 00:00:00 2001 From: Jeremy Marcus Date: Fri, 31 Jul 2026 16:25:28 -0700 Subject: [PATCH 2/4] refactor pileup_counts with 1 core --- dimelo/load_processed.py | 106 +++++++++++++++++++++------------------ 1 file changed, 56 insertions(+), 50 deletions(-) diff --git a/dimelo/load_processed.py b/dimelo/load_processed.py index 79b69b78..e8c6fd1d 100644 --- a/dimelo/load_processed.py +++ b/dimelo/load_processed.py @@ -174,8 +174,6 @@ def pileup_counts_from_bedmethyl( # If one core requested, do not spawn processes # This removes unnecessary overhead and avoids nested process generation (i.e. from regions_to_list) if cores_to_run == 1: - # TODO: This is a quick and dirty version as a proof of concept; should refactor proces_chunks_from_regions_dict instead - source_tabix = pysam.TabixFile(str(bedmethyl_file)) valid_base_count = 0 modified_base_count = 0 for chunk in tqdm( @@ -185,34 +183,12 @@ def pileup_counts_from_bedmethyl( desc="Loading data", leave=False ): - chromosome = chunk["chromosome"] - subregion_start = chunk["subregion_start"] - subregion_end = chunk["subregion_end"] - strand = chunk["strand"] - - valid_base_subregion_counts = 0 - modified_base_subregion_counts = 0 - - # tabix throws and error if the contig is not present - # by the current design, this should be silent - if chromosome in source_tabix.contigs: - for row in source_tabix.fetch( - chromosome, max(subregion_start, 0), subregion_end - ): - ( - keep_basemod, - _, - modified_in_row, - valid_in_row, - ) = process_pileup_row( - row=row, - parsed_motif=parsed_motif, - region_strand=strand, - single_strand=single_strand, - ) - if keep_basemod: - valid_base_subregion_counts += valid_in_row - modified_base_subregion_counts += modified_in_row + modified_base_subregion_counts, valid_base_subregion_counts = pileup_counts_process_chunk( + bedmethyl_file, + parsed_motif, + chunk, + single_strand, + ) valid_base_count += valid_base_subregion_counts modified_base_count += modified_base_subregion_counts else: @@ -230,7 +206,7 @@ def pileup_counts_from_bedmethyl( with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: futures = [ executor.submit( - pileup_counts_process_chunk, + pileup_counts_process_chunk_parallel, bedmethyl_file, parsed_motif, chunk, @@ -251,7 +227,7 @@ def pileup_counts_from_bedmethyl( try: future.result() except Exception as err: - raise RuntimeError("pileup_counts_process_chunk failed.") from err + raise RuntimeError("pileup_counts_process_chunk_parallel failed.") from err # Directly convert shared memory buffers to integers modified_base_count = int.from_bytes( @@ -529,13 +505,10 @@ def pileup_counts_process_chunk( bedmethyl_file, parsed_motif, chunk, - shm_name_modified, - shm_name_valid, - lock, single_strand, -) -> None: +) -> tuple[int, int]: """ - Helper function to allow pileup_counts_from_bedmethyl to operate in a parallized fashion. + Helper function to allow pileup_counts_from_bedmethyl to operate in a modular fashion. Sum up modified and valid counts for a subregion chunk in a bedmethyl file. @@ -543,30 +516,21 @@ def pileup_counts_process_chunk( bedmethyl_file: Path to bedmethyl file parsed_motif: ParsedMotif object chunk: a dict containing subregion chunk information - shm_name_modified: the name string for the shared memory location containing the modified counts sum - shm_name_valid: the name string for the shared memory location containing the valid counts sum - lock: a manager.Lock object to allow synchronization in accessing shared memory single_strand: True if only single-strand mods are desired Returns: - None. Counts are added in-place to shared memory. + tuple containing counts of (modified_bases, total_bases) for the requested chunk """ source_tabix = pysam.TabixFile(str(bedmethyl_file)) - existing_valid = shared_memory.SharedMemory(name=shm_name_valid) - existing_modified = shared_memory.SharedMemory(name=shm_name_modified) - valid_base_counts = np.ndarray((1,), dtype=np.int32, buffer=existing_valid.buf) - modified_base_counts = np.ndarray( - (1,), dtype=np.int32, buffer=existing_modified.buf - ) + + valid_base_subregion_counts = 0 + modified_base_subregion_counts = 0 chromosome = chunk["chromosome"] subregion_start = chunk["subregion_start"] subregion_end = chunk["subregion_end"] strand = chunk["strand"] - valid_base_subregion_counts = 0 - modified_base_subregion_counts = 0 - # tabix throws and error if the contig is not present # by the current design, this should be silent if chromosome in source_tabix.contigs: @@ -588,6 +552,48 @@ def pileup_counts_process_chunk( valid_base_subregion_counts += valid_in_row modified_base_subregion_counts += modified_in_row + return modified_base_subregion_counts, valid_base_subregion_counts + +def pileup_counts_process_chunk_parallel( + bedmethyl_file, + parsed_motif, + chunk, + shm_name_modified, + shm_name_valid, + lock, + single_strand, +) -> None: + """ + Helper function to allow pileup_counts_from_bedmethyl to operate in a parallized fashion. + + Sum up modified and valid counts for a subregion chunk in a bedmethyl file. + + Args: + bedmethyl_file: Path to bedmethyl file + parsed_motif: ParsedMotif object + chunk: a dict containing subregion chunk information + shm_name_modified: the name string for the shared memory location containing the modified counts sum + shm_name_valid: the name string for the shared memory location containing the valid counts sum + lock: a manager.Lock object to allow synchronization in accessing shared memory + single_strand: True if only single-strand mods are desired + + Returns: + None. Counts are added in-place to shared memory. + """ + existing_valid = shared_memory.SharedMemory(name=shm_name_valid) + existing_modified = shared_memory.SharedMemory(name=shm_name_modified) + valid_base_counts = np.ndarray((1,), dtype=np.int32, buffer=existing_valid.buf) + modified_base_counts = np.ndarray( + (1,), dtype=np.int32, buffer=existing_modified.buf + ) + + modified_base_subregion_counts, valid_base_subregion_counts = pileup_counts_process_chunk( + bedmethyl_file, + parsed_motif, + chunk, + single_strand, + ) + with lock: valid_base_counts[0] += valid_base_subregion_counts modified_base_counts[0] += modified_base_subregion_counts From 5e48be460700e15c8a7b4a0f369e21d2f1cf3f15 Mon Sep 17 00:00:00 2001 From: Jeremy Marcus Date: Fri, 31 Jul 2026 16:51:55 -0700 Subject: [PATCH 3/4] refactor pileup_vectors for single-core operation --- dimelo/load_processed.py | 208 ++++++++++++++++++++++++--------------- 1 file changed, 129 insertions(+), 79 deletions(-) diff --git a/dimelo/load_processed.py b/dimelo/load_processed.py index e8c6fd1d..d86f57f9 100644 --- a/dimelo/load_processed.py +++ b/dimelo/load_processed.py @@ -303,66 +303,93 @@ def pileup_vectors_from_bedmethyl( regions_dict, chunk_size=chunk_size ) - cores_to_run = utils.cores_to_run(cores) - # Peek at a region to figure out what size the vectors should be first_key = next(iter(regions_dict)) first_tuple = regions_dict[first_key][0] region_len = first_tuple[1] - first_tuple[0] - # Initialize shared memory as numpy arrays to make it easy to map to buffer in subprocesses - shm_valid = shared_memory.SharedMemory( - create=True, size=(region_len) * np.dtype(np.int32).itemsize - ) - shm_modified = shared_memory.SharedMemory( - create=True, size=(region_len) * np.dtype(np.int32).itemsize - ) - - manager = multiprocessing.Manager() - lock = manager.Lock() + cores_to_run = utils.cores_to_run(cores) - with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: - futures = [ - executor.submit( - pileup_vectors_process_chunk, + # If one core requested, do not spawn processes + # This removes unnecessary overhead and avoids nested process generation (i.e. from regions_to_list) + if cores_to_run == 1: + modified_base_counts = np.zeros(region_len, dtype=int) + valid_base_counts = np.zeros(region_len, dtype=int) + for chunk in tqdm( + chunks_list, + total=len(chunks_list), + disable=quiet, + desc="Loading data", + leave=False, + ): + subregion_start_idx, subregion_end_idx, modified_base_subregion, valid_base_subregion = pileup_vectors_process_chunk( bedmethyl_file, parsed_motif, chunk, region_len, - shm_modified.name, - shm_valid.name, - lock, single_strand, regions_5to3prime, ) - for chunk in chunks_list - ] - for future in tqdm( - concurrent.futures.as_completed(futures), - total=len(futures), - disable=quiet, - desc="Loading data", - leave=False, - ): - try: - future.result() - except Exception as err: - raise RuntimeError("pileup_vectors_process_chunk failed.") from err - - # We need to convert these shared memory buffers to numpy arrays which - # we then copy, so that they no longer reference the shared memory which - # will soon be de-allocated - modified_base_counts = np.copy( - np.ndarray((region_len,), dtype=np.int32, buffer=shm_modified.buf) - ) - valid_base_counts = np.copy( - np.ndarray((region_len,), dtype=np.int32, buffer=shm_valid.buf) - ) - # Close and unlink shared memory - not fully handled by garbage collection otherwise - shm_modified.close() - shm_modified.unlink() - shm_valid.close() - shm_valid.unlink() + valid_base_counts[ + subregion_start_idx : subregion_end_idx + ] += valid_base_subregion + modified_base_counts[ + subregion_start_idx : subregion_end_idx + ] += modified_base_subregion + else: + # Initialize shared memory as numpy arrays to make it easy to map to buffer in subprocesses + shm_valid = shared_memory.SharedMemory( + create=True, size=(region_len) * np.dtype(np.int32).itemsize + ) + shm_modified = shared_memory.SharedMemory( + create=True, size=(region_len) * np.dtype(np.int32).itemsize + ) + + manager = multiprocessing.Manager() + lock = manager.Lock() + + with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: + futures = [ + executor.submit( + pileup_vectors_process_chunk_parallel, + bedmethyl_file, + parsed_motif, + chunk, + region_len, + shm_modified.name, + shm_valid.name, + lock, + single_strand, + regions_5to3prime, + ) + for chunk in chunks_list + ] + for future in tqdm( + concurrent.futures.as_completed(futures), + total=len(futures), + disable=quiet, + desc="Loading data", + leave=False, + ): + try: + future.result() + except Exception as err: + raise RuntimeError("pileup_vectors_process_chunk failed.") from err + + # We need to convert these shared memory buffers to numpy arrays which + # we then copy, so that they no longer reference the shared memory which + # will soon be de-allocated + modified_base_counts = np.copy( + np.ndarray((region_len,), dtype=np.int32, buffer=shm_modified.buf) + ) + valid_base_counts = np.copy( + np.ndarray((region_len,), dtype=np.int32, buffer=shm_valid.buf) + ) + # Close and unlink shared memory - not fully handled by garbage collection otherwise + shm_modified.close() + shm_modified.unlink() + shm_valid.close() + shm_valid.unlink() return modified_base_counts, valid_base_counts @@ -402,39 +429,10 @@ def pileup_vectors_process_chunk( parsed_motif, chunk, region_len, - shm_name_modified, - shm_name_valid, - lock, single_strand, regions_5to3prime, -) -> None: - """ - Helper function to allow pileup_vectors_from_bedmethyl to operate in a parallized fashion. - - Sum up modified and valid counts for a subregion chunk in a bedmethyl file. - - Args: - bedmethyl_file: Path to bedmethyl file - parsed_motif: ParsedMotif object - chunk: a dict containing subregion chunk information - shm_name_modified: the name string for the shared memory location containing the modified counts array - shm_name_valid: the name string for the shared memory location containing the valid counts array - lock: a manager.Lock object to allow synchronization in accessing shared memory - single_strand: True if only single-strand mods are desired - regions_5to3prime: True means negative strand regions get flipped, False means no flipping - - Returns: - None. Counts are added to arrays in-place to shared memory. - """ +) -> tuple[int, int, np.ndarray, np.ndarray]: source_tabix = pysam.TabixFile(str(bedmethyl_file)) - existing_valid = shared_memory.SharedMemory(name=shm_name_valid) - existing_modified = shared_memory.SharedMemory(name=shm_name_modified) - valid_base_counts = np.ndarray( - (region_len,), dtype=np.int32, buffer=existing_valid.buf - ) - modified_base_counts = np.ndarray( - (region_len,), dtype=np.int32, buffer=existing_modified.buf - ) chromosome = chunk["chromosome"] region_start = chunk["region_start"] @@ -488,13 +486,65 @@ def pileup_vectors_process_chunk( modified_base_subregion[pileup_coord_in_subregion] += ( modified_in_row ) + + subregion_start_idx = subregion_offset + subregion_end_idx = subregion_offset + abs(subregion_end - subregion_start) + + return subregion_start_idx, subregion_end_idx, modified_base_subregion, valid_base_subregion + +def pileup_vectors_process_chunk_parallel( + bedmethyl_file, + parsed_motif, + chunk, + region_len, + shm_name_modified, + shm_name_valid, + lock, + single_strand, + regions_5to3prime, +) -> None: + """ + Helper function to allow pileup_vectors_from_bedmethyl to operate in a parallized fashion. + + Sum up modified and valid counts for a subregion chunk in a bedmethyl file. + + Args: + bedmethyl_file: Path to bedmethyl file + parsed_motif: ParsedMotif object + chunk: a dict containing subregion chunk information + shm_name_modified: the name string for the shared memory location containing the modified counts array + shm_name_valid: the name string for the shared memory location containing the valid counts array + lock: a manager.Lock object to allow synchronization in accessing shared memory + single_strand: True if only single-strand mods are desired + regions_5to3prime: True means negative strand regions get flipped, False means no flipping + + Returns: + None. Counts are added to arrays in-place to shared memory. + """ + existing_valid = shared_memory.SharedMemory(name=shm_name_valid) + existing_modified = shared_memory.SharedMemory(name=shm_name_modified) + valid_base_counts = np.ndarray( + (region_len,), dtype=np.int32, buffer=existing_valid.buf + ) + modified_base_counts = np.ndarray( + (region_len,), dtype=np.int32, buffer=existing_modified.buf + ) + + subregion_start_idx, subregion_end_idx, modified_base_subregion, valid_base_subregion = pileup_vectors_process_chunk( + bedmethyl_file, + parsed_motif, + chunk, + region_len, + single_strand, + regions_5to3prime, + ) with lock: valid_base_counts[ - subregion_offset : subregion_offset + abs(subregion_end - subregion_start) + subregion_start_idx : subregion_end_idx ] += valid_base_subregion modified_base_counts[ - subregion_offset : subregion_offset + abs(subregion_end - subregion_start) + subregion_start_idx : subregion_end_idx ] += modified_base_subregion # Close the file descriptor/handle to the shared memory existing_modified.close() From 0dab125c6145df9f0bf4a8ef816ff31eb2a76275 Mon Sep 17 00:00:00 2001 From: Jeremy Marcus Date: Fri, 31 Jul 2026 16:53:52 -0700 Subject: [PATCH 4/4] formatting fixes --- dimelo/load_processed.py | 87 ++++++++++++++++++++++++++-------------- 1 file changed, 56 insertions(+), 31 deletions(-) diff --git a/dimelo/load_processed.py b/dimelo/load_processed.py index d86f57f9..fac7aeb4 100644 --- a/dimelo/load_processed.py +++ b/dimelo/load_processed.py @@ -181,13 +181,15 @@ def pileup_counts_from_bedmethyl( total=len(chunks_list), disable=quiet, desc="Loading data", - leave=False + leave=False, ): - modified_base_subregion_counts, valid_base_subregion_counts = pileup_counts_process_chunk( - bedmethyl_file, - parsed_motif, - chunk, - single_strand, + modified_base_subregion_counts, valid_base_subregion_counts = ( + pileup_counts_process_chunk( + bedmethyl_file, + parsed_motif, + chunk, + single_strand, + ) ) valid_base_count += valid_base_subregion_counts modified_base_count += modified_base_subregion_counts @@ -199,11 +201,13 @@ def pileup_counts_from_bedmethyl( shm_modified = shared_memory.SharedMemory( create=True, size=np.dtype(np.int32).itemsize ) - + manager = multiprocessing.Manager() lock = manager.Lock() - with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: + with concurrent.futures.ProcessPoolExecutor( + max_workers=cores_to_run + ) as executor: futures = [ executor.submit( pileup_counts_process_chunk_parallel, @@ -227,7 +231,9 @@ def pileup_counts_from_bedmethyl( try: future.result() except Exception as err: - raise RuntimeError("pileup_counts_process_chunk_parallel failed.") from err + raise RuntimeError( + "pileup_counts_process_chunk_parallel failed." + ) from err # Directly convert shared memory buffers to integers modified_base_count = int.from_bytes( @@ -322,7 +328,12 @@ def pileup_vectors_from_bedmethyl( desc="Loading data", leave=False, ): - subregion_start_idx, subregion_end_idx, modified_base_subregion, valid_base_subregion = pileup_vectors_process_chunk( + ( + subregion_start_idx, + subregion_end_idx, + modified_base_subregion, + valid_base_subregion, + ) = pileup_vectors_process_chunk( bedmethyl_file, parsed_motif, chunk, @@ -330,12 +341,12 @@ def pileup_vectors_from_bedmethyl( single_strand, regions_5to3prime, ) - valid_base_counts[ - subregion_start_idx : subregion_end_idx - ] += valid_base_subregion - modified_base_counts[ - subregion_start_idx : subregion_end_idx - ] += modified_base_subregion + valid_base_counts[subregion_start_idx:subregion_end_idx] += ( + valid_base_subregion + ) + modified_base_counts[subregion_start_idx:subregion_end_idx] += ( + modified_base_subregion + ) else: # Initialize shared memory as numpy arrays to make it easy to map to buffer in subprocesses shm_valid = shared_memory.SharedMemory( @@ -348,7 +359,9 @@ def pileup_vectors_from_bedmethyl( manager = multiprocessing.Manager() lock = manager.Lock() - with concurrent.futures.ProcessPoolExecutor(max_workers=cores_to_run) as executor: + with concurrent.futures.ProcessPoolExecutor( + max_workers=cores_to_run + ) as executor: futures = [ executor.submit( pileup_vectors_process_chunk_parallel, @@ -486,11 +499,17 @@ def pileup_vectors_process_chunk( modified_base_subregion[pileup_coord_in_subregion] += ( modified_in_row ) - + subregion_start_idx = subregion_offset subregion_end_idx = subregion_offset + abs(subregion_end - subregion_start) - return subregion_start_idx, subregion_end_idx, modified_base_subregion, valid_base_subregion + return ( + subregion_start_idx, + subregion_end_idx, + modified_base_subregion, + valid_base_subregion, + ) + def pileup_vectors_process_chunk_parallel( bedmethyl_file, @@ -530,7 +549,12 @@ def pileup_vectors_process_chunk_parallel( (region_len,), dtype=np.int32, buffer=existing_modified.buf ) - subregion_start_idx, subregion_end_idx, modified_base_subregion, valid_base_subregion = pileup_vectors_process_chunk( + ( + subregion_start_idx, + subregion_end_idx, + modified_base_subregion, + valid_base_subregion, + ) = pileup_vectors_process_chunk( bedmethyl_file, parsed_motif, chunk, @@ -540,12 +564,10 @@ def pileup_vectors_process_chunk_parallel( ) with lock: - valid_base_counts[ - subregion_start_idx : subregion_end_idx - ] += valid_base_subregion - modified_base_counts[ - subregion_start_idx : subregion_end_idx - ] += modified_base_subregion + valid_base_counts[subregion_start_idx:subregion_end_idx] += valid_base_subregion + modified_base_counts[subregion_start_idx:subregion_end_idx] += ( + modified_base_subregion + ) # Close the file descriptor/handle to the shared memory existing_modified.close() existing_valid.close() @@ -604,6 +626,7 @@ def pileup_counts_process_chunk( return modified_base_subregion_counts, valid_base_subregion_counts + def pileup_counts_process_chunk_parallel( bedmethyl_file, parsed_motif, @@ -637,11 +660,13 @@ def pileup_counts_process_chunk_parallel( (1,), dtype=np.int32, buffer=existing_modified.buf ) - modified_base_subregion_counts, valid_base_subregion_counts = pileup_counts_process_chunk( - bedmethyl_file, - parsed_motif, - chunk, - single_strand, + modified_base_subregion_counts, valid_base_subregion_counts = ( + pileup_counts_process_chunk( + bedmethyl_file, + parsed_motif, + chunk, + single_strand, + ) ) with lock: