1# Creating a bytes object from a list of integers
2byte_list = [72, 101, 108, 108, 111]
3byte_stream = bytes(byte_list)
4print(byte_stream)  # Output: b'Hello'
5

1# Creating a mutable bytearray
2mutable_bytes = bytearray([72, 101, 108, 108, 111])
3mutable_bytes[0] = 87  # Change 'H' to 'W'
4print(mutable_bytes)  # Output: bytearray(b'Wello')
5

1# Reading bytes from a file using `open()` in binary mode
2with open('my_file.bin', 'rb') as f:
3    file_bytes = f.read()
4print(file_bytes)
5

1import socket
2
3# Receiving bytes from a socket
4# This is a simplified example and needs proper socket setup
5
6s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
7s.bind(('localhost', 12345))
8s.listen(1)
9conn, addr = s.accept()
10
11data = conn.recv(1024)
12print(data)
13
14conn.close()
15s.close()
16

1# Reading a specific number of bytes
2with open('my_file.bin', 'rb') as f:
3    first_10_bytes = f.read(10)
4    print(first_10_bytes)
5

1# Iterating over a byte stream
2with open('my_file.bin', 'rb') as f:
3    for byte in f.read():
4        print(byte)  # Prints each byte as an integer
5

1# Writing bytes to a file
2data = b'Hello, world!'
3with open('output.bin', 'wb') as f:
4    f.write(data)
5

1import socket
2
3# Sending bytes over a network socket
4# This is a simplified example and needs proper socket setup
5
6s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
7s.connect(('localhost', 12345))
8
9data = b'Hello from client!'
10s.sendall(data)
11
12s.close()
13

1# Using a generator to efficiently process a large byte stream
2def byte_stream_generator(filename, chunk_size=4096):
3    with open(filename, 'rb') as f:
4        while True:
5            chunk = f.read(chunk_size)
6            if not chunk:
7                break
8            yield chunk
9
10# Example usage
11for chunk in byte_stream_generator('large_file.bin'):
12    # Process the chunk
13    print(f'Processed chunk of size: {len(chunk)}')
14

1import mmap
2
3# Memory mapping a file for efficient byte stream access
4with open('large_file.bin', 'rb') as f:
5    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
6        # Access bytes using mm[offset]
7        print(mm[0:10])  # Print the first 10 bytes
8

1import io
2
3data = b"This is some data to be treated as a stream"
4
5# Create an in-memory byte stream
6byte_stream = io.BytesIO(data)
7
8# Read from the stream
9print(byte_stream.read(5))  # Output: b'This '
10
11# Write to the stream
12byte_stream.write(b" more data")
13
14# Get the current position
15print(byte_stream.tell())
16

1import asyncio
2
3async def read_bytes_async(stream, n):
4    loop = asyncio.get_event_loop()
5    return await loop.run_in_executor(None, stream.read, n)
6
7async def main():
8    with open('my_file.bin', 'rb') as f:
9        # Wrap the file object to make it non-blocking
10        loop = asyncio.get_event_loop()
11        f_reader = asyncio.StreamReader(loop=loop)
12        await f_reader.set_transport(asyncio.StreamReader.StreamReaderTransport(f, loop=loop))
13
14        chunk = await read_bytes_async(f, 1024) # Read 1024 bytes asynchronously
15        print(chunk)
16
17if __name__ == "__main__":
18    asyncio.run(main())
19

1import zlib
2
3# Compressing and decompressing a byte stream using zlib
4data = b'This is a long string that can be compressed.'
5compressed_data = zlib.compress(data)
6print(f'Original size: {len(data)}, Compressed size: {len(compressed_data)}')
7
8decompressed_data = zlib.decompress(compressed_data)
9print(f'Decompressed data: {decompressed_data}')
10

1# Encoding and decoding using UTF-8
2text = '你好，世界！'
3encoded_text = text.encode('utf-8')
4print(encoded_text)
5
6decoded_text = encoded_text.decode('utf-8')
7print(decoded_text)
8