Suppose we have a sequence of UTF-8 bytes—for example, the contents of a binary file opened for reading—and want to determine how many characters (Unicode code points) it represents.
This task is straightforward if the entire byte sequence fits in memory as a bytes object. In that case, we can simply call the sequence’s decode() method to convert the bytes into a string and then use the len() function to determine the string’s length.
However, the situation is different when the UTF-8 byte sequence is too large to fit in memory all at once. In this case, we need to process the sequence in fixed-size chunks and use an algorithm to count the characters it contains.
There are two fundamental approaches to solving this problem.
The first approach relies on decoding and re-encoding each chunk using the bytes.decode() and str.encode() methods. The main steps are as follows:
- Decode the chunk, ignoring any incomplete character at the end—that is, a character that cannot yet be fully decoded.
- Increase the character counter by the length of the decoded string.
- Determine how many bytes were consumed when decoding the successfully decoded text.
- Prepend the remaining bytes from the current chunk to the next chunk so that the incomplete character can be decoded when more bytes become available.
Decoding and re-encoding each chunk may not be particularly efficient, so we should also consider an alternative approach.
Without decoding, we can work directly with the bytes. In this case, we need to understand certain details of UTF-8 encoding and use the properties of individual bytes to determine which bytes mark the beginning of a character and how many characters the sequence contains. In the following image, you can see a summary of the fundamental properties of UTF-8 encoding, which we will use.
Based on these properties, we first define two helper functions. One determines whether a valid UTF-8 byte marks the beginning of a character; such a byte is known as a leading byte. The other determines how many continuation bytes must follow a given leading byte. The definitions of these functions are shown below. We will use the Protocol class and the imported modules later.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
from __future__ import annotations from typing import Protocol from itertools import repeat import operator as op class ReadableBinaryStream(Protocol): def read(self, size: int = -1, /) -> bytes: ... def readable(self) -> bool: ... class Byte(int): def __new__(cls, value: int | Byte) -> Byte: if isinstance(value, bool) or not isinstance(value, int): raise TypeError("The value of Byte must be an integer.") if not 0 <= value <= 255: raise ValueError("The value of Byte must be between 0 and 255") return super().__new__(cls, value) def is_leading_byte(utf8_byte: Byte | int) -> bool: """Return True if the byte passed as an argument could be the leading byte of a valid UTF-8 encoded character. Sources of the value ranges: - Unicode Standard, Chapter 3, Table 3.7, “Well-Formed UTF-8 Byte Sequences” - RFC 3629, Chapter 4, “Syntax of UTF-8 Byte Sequences” """ utf8_byte = Byte(utf8_byte) return 0x00 <= utf8_byte <= 0x7F or 0xC2 <= utf8_byte <= 0xF4 def num_continuation_bytes(utf8_byte: Byte | int) -> int: """Return the number of continuation bytes following the given UTF-8 leading byte. Return values: 0–3: the number of continuation bytes for a valid UTF-8 leading byte. -1: if the argument is not a valid UTF-8 leading byte. Sources that specify how many continuation bytes are associated with each leading byte: - Unicode Standard, Chapter 3, Table 3.7, “Well-Formed UTF-8 Byte Sequences” - RFC 3629, Chapter 4, “Syntax of UTF-8 Byte Sequences” """ utf8_byte = Byte(utf8_byte) if not is_leading_byte(utf8_byte): return -1 if utf8_byte <= 0x7F: return 0 elif utf8_byte <= 0xDF: return 1 elif utf8_byte <= 0xEF: return 2 else: return 3 |
With these preliminaries in place, we can now implement several solutions to the problem. They use different approaches and, consequently, different Python language features.
The individual function definitions are shown below, and the comments explain the underlying principle behind each implementation. The first function implements the solution that uses the bytes.decode() and str.encode() methods. All other implementations rely on the properties of UTF-8 encoding.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 |
# Each of the following functions counts the number of valid characters (code points) in a # binary file opened for reading that contains UTF-8 bytes. Processing is performed in chunks of ‘chunk_size’ bytes. def count_chars_in_file1(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # Decode the bytes of the chunk. If there is a truncated, non-decodable character at the end, ignore it. # Increase the character counter by the length of the decoded text. # Next, determine how many bytes the successfully decoded text occupied within the chunk. # Prepend the remaining bytes of the chunk to the next chunk so they can be properly decoded there. total_chars = 0 puffer: bytes = b"" # Stores the bytes of the truncated characters remaining at the chunk boundary. while chunk := binfileobj.read(chunk_size): # If there are any remaining bytes from the previous iteration, place them at the beginning of the new chunk. chunk: bytes = puffer + chunk # Decode the chunk, ignoring any truncated characters at the end. decoded_text = chunk.decode('utf-8', errors='ignore') total_chars += len(decoded_text) # Calculate how many bytes the successfully decoded text occupies. bytecount_processed: int = len(decoded_text.encode('utf-8')) # Store the remainder of the chunk for the next iteration. puffer = chunk[bytecount_processed:] # Once there are no more bytes to read, decode the bytes remaining in the buffer. # Increase the character counter by the number of valid characters in the result. if puffer: decoded_text = puffer.decode('utf-8', errors='ignore') total_chars += len(decoded_text) return total_chars def count_chars_in_file2(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # Count the identified leading bytes of characters in the chunk sequentially, # then update the character counter accordingly. total_chars = 0 while chunk := binfileobj.read(chunk_size): leading_byte_count = sum(is_leading_byte(utf8_byte) for utf8_byte in chunk) total_chars += leading_byte_count return total_chars def count_chars_in_file3(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # Count the number of continuation bytes in each chunk. # The bit pattern of any UTF-8 continuation byte is 10xxxxxx. Based on this, we mask out # the top two bits of each byte in the chunk to filter and count the continuation bytes. # Subtracting this from the total byte count of the chunk gives the number of leading bytes, # which is then used to update the character counter. total_chars = 0 while chunk := binfileobj.read(chunk_size): # The number of continuation bytes. continuation_byte_count = sum(1 for b in chunk if (b & 0b11000000) == 0b10000000) # Update the character counter with the number of leading bytes within the chunk. total_chars += (len(chunk) - continuation_byte_count) return total_chars def count_chars_in_file4(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # Count the number of continuation bytes in each chunk. # The bit pattern of any UTF-8 continuation byte is 10xxxxxx. Based on this, we mask out # the top two bits of each byte in the chunk to filter and count the continuation bytes. # Subtracting this from the total byte count of the chunk gives the number of leading bytes, # which is then used to update the character counter. # To ensure that the intermediate loop iterating over the bytes executes within the fast C # implementation rather than at the Python level, the masking of the top two bits is # implemented using tools from the itertools and operator modules, while the counting is # handled by the count() method. This is called on a bytes() container to remain at the byte # level. This is preferable to using a tuple() or list(), as converting the bytes into int # objects would otherwise introduce unnecessary overhead. total_chars = 0 while chunk := binfileobj.read(chunk_size): # The number of continuation bytes. continuation_byte_count = bytes(map(op.and_, chunk, repeat(0b11000000))).count(0b10000000) # Other possible implementations: # continuation_byte_count = op.countOf(map(op.and_, chunk, repeat(0b11000000)), 0b10000000) # continuation_byte_count = op.countOf(map(op.rshift, chunk, repeat(6)), 0b10) # Update the character counter with the number of leading bytes within the chunk. total_chars += (len(chunk) - continuation_byte_count) return total_chars def count_chars_in_file5(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # We obtain the leading bytes within the current chunk by removing all continuation # bytes from the chunk's bytes. This is performed using the translate() method. # To achieve this, we determine all possible continuation byte values. # The bit pattern of UTF-8 continuation bytes is 10xxxxxx. This means the range of # continuation bytes is 10000000 to 10111111. # Therefore, the value of the continuation bytes falls between 128 (0x80) and 191 (0xBF). # All possible continuation bytes. possible_utf8_continuation_bytes = bytes(range(0x80, 0xC0)) total_chars = 0 while chunk := binfileobj.read(chunk_size): # Delete the continuation bytes from the chunk's bytes so that only the leading bytes remain. leading_byte_count = len(chunk.translate(None, possible_utf8_continuation_bytes)) total_chars += leading_byte_count return total_chars def count_chars_in_file6(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # In each current chunk, the leading bytes are counted by extracting the number of # continuation bytes that follow them. Based on this, we determine the position of the next leading byte. # The number of continuation bytes is determined by calling a dedicated function. total_chars = 0 while chunk := binfileobj.read(chunk_size): # Find the first leading byte in the chunk. for idx, utf8_byte in enumerate(chunk): if is_leading_byte(utf8_byte): break else: continue leading_byte_pos, leading_byte = idx, utf8_byte total_chars += 1 while True: new_leading_byte_pos = leading_byte_pos + num_continuation_bytes(leading_byte) + 1 if new_leading_byte_pos > len(chunk) - 1: break leading_byte_pos = new_leading_byte_pos leading_byte = chunk[leading_byte_pos] total_chars += 1 return total_chars def count_chars_in_file7(binfileobj: ReadableBinaryStream, chunk_size: int = 1024 * 1024) -> int: if not binfileobj.readable(): raise ValueError("The file object is not readable.") # How it works: # In each current chunk, the leading bytes are counted by extracting the number of # continuation bytes that follow them. Based on this, we determine the position of the next leading byte. # The number of continuation bytes is retrieved from a dedicated, precomputed container. # This is implemented using a 256-element tuple container, where the indices represent the possible # values of a byte, and the elements represent the number of continuation bytes for that given value. # If the byte value is not a valid leading byte, the corresponding element in the tuple will be -1. utf8_continuation_byte_counts = tuple( 0 if b <= 0x7F else 1 if 0xC2 <= b <= 0xDF else 2 if 0xE0 <= b <= 0xEF else 3 if 0xF0 <= b <= 0xF4 else -1 for b in range(256) ) total_chars = 0 while chunk := binfileobj.read(chunk_size): # Find the first leading byte in the chunk. for idx, utf8_byte in enumerate(chunk): if is_leading_byte(utf8_byte): break else: continue leading_byte_pos = idx leading_byte = utf8_byte total_chars += 1 while True: new_leading_byte_pos = leading_byte_pos + utf8_continuation_byte_counts[leading_byte] + 1 if new_leading_byte_pos > len(chunk) - 1: break leading_byte_pos = new_leading_byte_pos leading_byte = chunk[leading_byte_pos] total_chars += 1 return total_chars |
We measured the execution times using the test code shown below. We performed the measurements using two chunk sizes and files of 1, 10, 100, and 1000 MB. The number of characters in these files increases in direct proportion to their size.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
from timeit import timeit file_folder = r"D:\" for filesize in (1, 10, 100, 1000): file_path = rf"{file_folder}\test_text_file_{filesize}MB.txt" chunk_size = 128 * 1024 * 1024 # bytes print(f'File size: {filesize} MB. Chunk size: {chunk_size / 1024 ** 2} MiB') with open(file_path, 'r', encoding='UTF-8', newline='') as ftxt: while ftxt.read(chunk_size): pass with open(file_path, 'r', encoding='UTF-8', newline='') as f: total_chars = sum(len(chunk) for chunk in iter(lambda: f.read(chunk_size), '')) print('\tTotal number of characters: {}'.format(total_chars)) # The functions to be tested. functions_to_test = [count_chars_in_file1, count_chars_in_file2, count_chars_in_file3, count_chars_in_file4, count_chars_in_file5, count_chars_in_file6, count_chars_in_file7] for count_chars_in_file in functions_to_test: with open(file_path, 'rb') as file: t = timeit(lambda: count_chars_in_file(file, chunk_size), number=1) print('\t{}: {:.3f}'.format(count_chars_in_file.__name__, t)) |
The execution-time measurements are summarized in the table in the following image.
As you can see, there are substantial differences in the execution times of the various implementations, but the results also clearly demonstrate their linear scaling behavior. Increasing the file size by a factor of ten results in an approximately tenfold increase in execution time, indicating linear time complexity, or O(n). Thus, the significant differences in execution time are due to the differences in the implementations.
Among the implementations tested, Solution 5, which uses the bytes.translate() method, is clearly the fastest. Surprisingly, Solution 1, which decodes and re-encodes each chunk, ranks second, contrary to what we might expect. The remaining implementations take one or two orders of magnitude longer to run.
The reason for this significant difference is that bytes.translate(), bytes.decode() and str.encode() rely on optimized native code in CPython. These methods process the data without incurring the overhead associated with Python-level loops and the operations they execute.
However, implementations that process individual bytes in Python-level loops incur considerable overhead, as reflected in the measurement results. You can also observe that these implementations run faster in newer Python versions, possibly reflecting improvements in Python’s instruction execution and runtime optimizations.
Comparing Solutions 6 and 7 reveals another interesting difference. Replacing a function call that determines the number of continuation bytes with an indexed lookup in a precomputed container significantly reduces execution time.
The reason is that retrieving an item from a tuple by index can be considerably faster than calling a separate Python function. During a function call, the interpreter must perform several tasks, including handling the call, preparing the arguments, setting up the function’s execution context, executing the function body, and returning the result to the caller. Depending on the Python implementation and version, these tasks can introduce varying amounts of overhead.
By contrast, retrieving an item from a tuple by index is much simpler: the interpreter checks that the index is valid, locates the corresponding item in the tuple’s internal data structure, and returns a reference to it.
The difference between these two approaches becomes particularly significant when the operation must be performed on every input byte or at least on a large proportion of the input bytes. In such cases, function-call overhead can occur on every iteration, allowing even a small difference per call to accumulate into a substantial difference in total execution time.
Writing the functions presented in this article requires knowledge of several Python language features. All these features are explained and illustrated with examples in the e-book Python Knowledge Building Step by Step: From the Basics to the First Desktop Application.