Fast unpack of SAM flag bitfield to named symbols

Question

def sam_flag_explainer(flag:int):
    """
    Intepret sam flag and make a flag-explained-dictionary. 

    """
    try:
        # Convert sam flags to binary format, like 0101001
        # flag is a integer.
        flag_binary = "{0:b}".format(flag)
    except Exception as err:
        if int(flag) == flag:
            flag = int(flag)
            flag_binary = "{0:b}".format(flag)
        else:
            print(f"{flag=}")
            raise ValueError from err

    flag_template = {
        "read_paired":False,
        "read_mapped_in_proper_pair":False,
        "read_unmapped":False,
        "mate_unmapped":False,
        "read_reverse_strand":False,
        "mate_reverse_strand":False,
        "first_in_pair":False,
        "second_in_pair":False,
        "not_primary_alignment":False,
        "read_fails_platform/vendor_quality_checks":False,
        "read_is_PCR_or_optical_duplicate":False,
        "supplementary_alignment":False,
    }

    for k, f in zip(flag_template.keys(), flag_binary[::-1]): # should intepret the flag from back.
        if int(f) == 1:
            flag_template[k] = True
    
    return flag_template

The code is the above.

Schematically, this function does:

Convert integer into binary number -> Read it from the back -> Modify the value of the key based on binary number element.

This function is called a lot of times in my program.

And this function is slow and it accounts for a large percentage of the overall execution time.

How could I make this faster?

Answer 1

I would prefer an implementation that is simple, built-in, type-safe, and explicit (the original implementation is arguably explicit, but not simple or type-safe; and whereas it does use built-ins it doesn't necessarily take advantage of the right ones).

This is faster, but if it isn't fast enough, you'd be in diminishing returns and need to re-evaluate your choice of using Python. Justin Chang's brute-force cache method does technically work and is quite fast, but is also higher-complexity, has less structure, and takes up 32 kB of memory; it's up to you to choose which trade-offs are more important.

__slots__ has at least three advantages - it improves class performance; it rejects arbitrary member assignment; and it gets the class halfway to type safety (defining which members are there, but not their types). Whereas it does require duplication of member names, as Matthieu M has suggested this can be reused to form _fields_ in runtime.

A note about type safety: sadly, the ctypes machinery doesn't really produce type-safe structs in a strict sense unless you go out of your way and add a hint for every member as e.g. read_paired: int. It's unclear if this is of value to you.

Also, having a / in a member name is not a great idea (even though it does technically work with ctypes), so I have replaced this with another underscore.

import ctypes
import struct


class SAMFlags(ctypes.LittleEndianStructure):
    __slots__ = (
        'read_paired',
        'read_mapped_in_proper_pair',
        'read_unmapped',
        'mate_unmapped',
        'read_reverse_strand',
        'mate_reverse_strand',
        'first_in_pair',
        'second_in_pair',
        'not_primary_alignment',
        'read_fails_platform_vendor_quality_checks',
        'read_is_PCR_or_optical_duplicate',
        'supplementary_alignment',
    )

    _fields_ = [(name, ctypes.c_uint8, 1) for name in __slots__]

    def __init__(self, flags: int) -> None:
        struct.pack_into('<H', self, 0, flags)

With this quick hack to do timing:

measures = []
for method in (SAMFlags, sam_flag_explainer_op):
    for bits_set in range(13):
        flags = (1 << bits_set) - 1
        def run():
            return method(flags)
        for _ in range(1_000):
            measures.append([
                method.__name__,
                bits_set,
                timeit.timeit(stmt=run, number=100)/100,
            ])
df = pd.DataFrame(
    data=measures,
    columns=['method', 'bits', 'time']
)
seaborn.lineplot(data=df, x='bits', y='time', hue='method')
plt.show()

we get:

Unsurprisingly, the original method exhibits a clear O(n) in the number of bits set in the flag.

Answer 2

There are 2^F possible flag configurations, where F is the number of fields. You can precompute each configuration and store the cached values in a list. To retrieve the relevant configuration, index into the list based on the least significant bits of flag (the remainder mod 2^F):

flag_template = {
    "read_paired": False,
    "read_mapped_in_proper_pair": False,
    "read_unmapped": False,
    "mate_unmapped": False,
    "read_reverse_strand": False,
    "mate_reverse_strand": False,
    "first_in_pair": False,
    "second_in_pair": False,
    "not_primary_alignment": False,
    "read_fails_platform/vendor_quality_checks": False,
    "read_is_PCR_or_optical_duplicate": False,
    "supplementary_alignment": False
}
flag_template_keys = flag_template.keys()
flag_field_count = len(flag_template_keys)
flag_field_combos = (2 ** flag_field_count)


def compute_explainer(flag: int):
    flag_binary = "{0:b}".format(flag)
    copy_template = dict(flag_template)  # make a copy
    for k, f in zip(flag_template_keys, flag_binary[::-1]):
        copy_template[k] = (int(f) == 1)  # set true / false
    return copy_template


cached_explainers = [0] * flag_field_combos  # precompute
for i in range(flag_field_combos):
    cached_explainers[i] = compute_explainer(i)


def sam_flag_explainer(flag: int):
    return cached_explainers[int(flag) % flag_field_combos]

This ought to catch most errors because int(flag) will fail in similar cases. The only edge case I can find is that the revised version gladly accepts negative numbers. You can write a small test suite to compare speeds and verify the output is identical for nonnegative flag values:

import time
start_time = time.time()
for i in range(10000000):
    sam_flag_explainer(i)
print("time elapsed: {:.2f}s".format(time.time() - start_time))
for i in [3, 17, 234343434]:
    print(sam_flag_explainer(i))
print(sam_flag_explainer("dfdfdf"))

Here's the time difference for the testing suites on my computer (Python 3.10):

original time elapsed: 57.75s
revised time elapsed: 2.60s

Hope that helps. Good work on making the question clear.

EDIT: Reinderien makes good points regarding the broader context in which this function may be used. Based on the use case, you may need to weigh tradeoffs. For example, one use case might be converting from 1 million flags to a table with 1 million rows and the fields as columns. In that case, I would suggest finding a way to vectorize the operation, taking in all the flags at once, so that you can speed up the heavy lifting. Another use case might be that some downstream function / API call / utility requires arguments with field names. In that case, you might be able to adjust the downstream process to use the flag directly. Finally, you might be able to inspect the upstream process (whatever is creating these flags) and configure it to output the explained form directly. Good luck.