0
# Bitmap Operations
1

2
Redis bitmap operations for efficient bit-level manipulation of string values. Bitmap operations provide memory-efficient storage and manipulation of binary data, supporting counting, positioning, and bitwise operations across large datasets.
3

4
## Capabilities
5

6
### Individual Bit Operations
7

8
Get and set individual bits within string values treated as bitmaps.
9

10
```python { .api }
11
def getbit(self, key: KeyT, offset: int) -> int: ...
12

13
def setbit(self, key: KeyT, offset: int, value: int) -> int: ...
14
```
15

16
### Bit Position Operations
17

18
Find the position of bits with specific values within bitmap data.
19

20
```python { .api }
21
def bitpos(self, key: KeyT, bit: int, *args: bytes) -> int: ...
22
```
23

24
### Bit Counting
25

26
Count the number of set bits in bitmap data with optional range specification.
27

28
```python { .api }
29
def bitcount(self, key: KeyT, *args: bytes) -> int: ...
30
```
31

32
### Bitwise Operations
33

34
Perform bitwise operations (AND, OR, XOR, NOT) between multiple bitmaps.
35

36
```python { .api }
37
def bitop(self, op_name: bytes, dst: KeyT, *keys: KeyT) -> int: ...
38
```
39

40
### Bitfield Operations
41

42
Advanced bit manipulation supporting multiple data types and operations in a single command.
43

44
```python { .api }
45
def bitfield(self, key: KeyT, *args: bytes) -> List[Optional[int]]: ...
46
```
47

48
## Usage Examples
49

50
### Basic Bitmap Operations
51

52
```python
53
import fakeredis
54

55
client = fakeredis.FakeRedis()
56

57
print("=== Basic Bitmap Operations ===")
58

59
# Set individual bits using SETBIT
60
client.setbit('bitmap:user_activity', 0, 1)  # User 0 active
61
client.setbit('bitmap:user_activity', 5, 1)  # User 5 active
62
client.setbit('bitmap:user_activity', 11, 1) # User 11 active
63
client.setbit('bitmap:user_activity', 15, 1) # User 15 active
64

65
# Get individual bits
66
print(f"User 0 active: {client.getbit('bitmap:user_activity', 0)}")
67
print(f"User 3 active: {client.getbit('bitmap:user_activity', 3)}")
68
print(f"User 5 active: {client.getbit('bitmap:user_activity', 5)}")
69

70
# Count total active users
71
active_count = client.bitcount('bitmap:user_activity')
72
print(f"Total active users: {active_count}")
73

74
# Find first active user
75
first_active = client.bitpos('bitmap:user_activity', 1)
76
print(f"First active user ID: {first_active}")
77

78
# Find first inactive user
79
first_inactive = client.bitpos('bitmap:user_activity', 0)
80
print(f"First inactive user ID: {first_inactive}")
81
```
82

83
### User Activity Tracking
84

85
```python
86
import fakeredis
87
import datetime
88

89
client = fakeredis.FakeRedis()
90

91
print("=== User Activity Tracking ===")
92

93
# Simulate daily user activity for a week
94
days = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']
95
user_activities = {
96
    'monday': [0, 1, 2, 5, 8, 10, 15, 20, 25],
97
    'tuesday': [1, 2, 3, 6, 9, 11, 16, 21, 26],
98
    'wednesday': [0, 2, 4, 7, 10, 12, 17, 22, 27],
99
    'thursday': [1, 3, 5, 8, 11, 13, 18, 23, 28],
100
    'friday': [0, 1, 4, 6, 9, 14, 19, 24, 29],
101
    'saturday': [2, 5, 7, 10, 15, 20, 25, 30],
102
    'sunday': [1, 3, 6, 8, 12, 16, 21, 26, 31]
103
}
104

105
# Record activity for each day
106
for day, active_users in user_activities.items():
107
    bitmap_key = f'activity:{day}'
108
    for user_id in active_users:
109
        client.setbit(bitmap_key, user_id, 1)
110
    
111
    daily_count = client.bitcount(bitmap_key)
112
    print(f"{day.capitalize()}: {daily_count} active users")
113

114
# Find users active every day (intersection)
115
print(f"\n=== Users Active Every Day ===")
116
client.bitop('AND', 'activity:all_days', 
117
             'activity:monday', 'activity:tuesday', 'activity:wednesday',
118
             'activity:thursday', 'activity:friday', 'activity:saturday', 'activity:sunday')
119

120
consistent_users = client.bitcount('activity:all_days')
121
print(f"Users active all 7 days: {consistent_users}")
122

123
# Find users active on any day (union)
124
print(f"\n=== Users Active Any Day ===")
125
client.bitop('OR', 'activity:any_day',
126
             'activity:monday', 'activity:tuesday', 'activity:wednesday', 
127
             'activity:thursday', 'activity:friday', 'activity:saturday', 'activity:sunday')
128

129
any_day_users = client.bitcount('activity:any_day')
130
print(f"Users active on any day: {any_day_users}")
131

132
# Find weekend-only users
133
print(f"\n=== Weekend vs Weekday Analysis ===")
134
client.bitop('OR', 'activity:weekend', 'activity:saturday', 'activity:sunday')
135
client.bitop('OR', 'activity:weekday', 
136
             'activity:monday', 'activity:tuesday', 'activity:wednesday',
137
             'activity:thursday', 'activity:friday')
138

139
weekend_count = client.bitcount('activity:weekend')
140
weekday_count = client.bitcount('activity:weekday')
141

142
print(f"Weekend active users: {weekend_count}")
143
print(f"Weekday active users: {weekday_count}")
144

145
# Users active on weekends but not weekdays
146
client.bitop('NOT', 'activity:not_weekday', 'activity:weekday')
147
client.bitop('AND', 'activity:weekend_only', 'activity:weekend', 'activity:not_weekday')
148
weekend_only = client.bitcount('activity:weekend_only')
149
print(f"Weekend-only users: {weekend_only}")
150
```
151

152
### Feature Flag Management
153

154
```python
155
import fakeredis
156

157
client = fakeredis.FakeRedis()
158

159
print("=== Feature Flag Management ===")
160

161
# Define feature flags
162
features = {
163
    'beta_ui': [1, 5, 10, 15, 20, 25, 30],      # Beta UI users
164
    'premium': [2, 7, 12, 17, 22, 27],          # Premium users  
165
    'analytics': [3, 8, 13, 18, 23, 28],        # Analytics access
166
    'api_v2': [4, 9, 14, 19, 24, 29],           # API v2 access
167
    'mobile_app': [0, 5, 10, 15, 20, 25, 30]    # Mobile app users
168
}
169

170
# Set feature flags
171
for feature, user_ids in features.items():
172
    for user_id in user_ids:
173
        client.setbit(f'feature:{feature}', user_id, 1)
174
    
175
    enabled_count = client.bitcount(f'feature:{feature}')
176
    print(f"{feature}: {enabled_count} users enabled")
177

178
# Find users with multiple features
179
print(f"\n=== Feature Combinations ===")
180

181
# Users with both beta UI and mobile app
182
client.bitop('AND', 'combo:beta_mobile', 'feature:beta_ui', 'feature:mobile_app')
183
beta_mobile_count = client.bitcount('combo:beta_mobile')
184
print(f"Beta UI + Mobile app: {beta_mobile_count} users")
185

186
# Premium users with analytics
187
client.bitop('AND', 'combo:premium_analytics', 'feature:premium', 'feature:analytics')
188
premium_analytics_count = client.bitcount('combo:premium_analytics')
189
print(f"Premium + Analytics: {premium_analytics_count} users")
190

191
# Users with any premium feature (premium OR analytics OR api_v2)
192
client.bitop('OR', 'temp:premium_or_analytics', 'feature:premium', 'feature:analytics')
193
client.bitop('OR', 'combo:any_premium', 'temp:premium_or_analytics', 'feature:api_v2')
194
any_premium_count = client.bitcount('combo:any_premium')
195
print(f"Any premium feature: {any_premium_count} users")
196

197
# Check specific user's features
198
def check_user_features(user_id: int):
199
    user_features = []
200
    for feature in features.keys():
201
        if client.getbit(f'feature:{feature}', user_id):
202
            user_features.append(feature)
203
    return user_features
204

205
print(f"\n=== Individual User Features ===")
206
test_users = [5, 12, 20]
207
for user_id in test_users:
208
    user_features = check_user_features(user_id)
209
    print(f"User {user_id}: {', '.join(user_features)}")
210
```
211

212
### Performance Monitoring with Bitmaps
213

214
```python
215
import fakeredis
216
import random
217
import time
218

219
client = fakeredis.FakeRedis()
220

221
print("=== Performance Monitoring with Bitmaps ===")
222

223
# Simulate server response times over 24 hours (1440 minutes)
224
# Bit = 1 means response time was acceptable (< 500ms)
225
# Bit = 0 means response time was slow (>= 500ms)
226

227
def simulate_server_performance():
228
    """Simulate 24 hours of server performance data"""
229
    performance_key = 'performance:today'
230
    
231
    # Simulate performance for each minute of the day
232
    for minute in range(1440):  # 24 * 60 minutes
233
        # 95% chance of good performance, 5% chance of slow response
234
        is_good = random.random() < 0.95
235
        client.setbit(performance_key, minute, 1 if is_good else 0)
236
    
237
    return performance_key
238

239
# Generate performance data
240
performance_key = simulate_server_performance()
241
print("Generated 24 hours of performance data...")
242

243
# Analyze performance
244
print(f"\n=== Performance Analysis ===")
245
total_minutes = 1440
246
good_minutes = client.bitcount(performance_key)
247
bad_minutes = total_minutes - good_minutes
248

249
print(f"Total minutes: {total_minutes}")
250
print(f"Good performance minutes: {good_minutes}")
251
print(f"Poor performance minutes: {bad_minutes}")
252
print(f"Uptime percentage: {(good_minutes / total_minutes) * 100:.2f}%")
253

254
# Find first incident
255
first_incident = client.bitpos(performance_key, 0)
256
if first_incident != -1:
257
    hours = first_incident // 60
258
    minutes = first_incident % 60
259
    print(f"First incident at: {hours:02d}:{minutes:02d}")
260
else:
261
    print("No incidents detected!")
262

263
# Analyze performance by hour
264
print(f"\n=== Hourly Performance Breakdown ===")
265
for hour in range(24):
266
    start_minute = hour * 60
267
    end_minute = start_minute + 59
268
    
269
    # Count good minutes in this hour using bitcount with range
270
    # Note: Redis bitcount range is in bytes, so we need to be careful
271
    hour_key = f'performance:hour_{hour}'
272
    
273
    # Copy the hour's data to analyze it
274
    for minute in range(60):
275
        global_minute = start_minute + minute
276
        bit_value = client.getbit(performance_key, global_minute)
277
        client.setbit(hour_key, minute, bit_value)
278
    
279
    good_in_hour = client.bitcount(hour_key)
280
    percentage = (good_in_hour / 60) * 100
281
    
282
    print(f"Hour {hour:02d}:00-{hour:02d}:59: {good_in_hour}/60 good minutes ({percentage:.1f}%)")
283

284
# Find longest streak of good performance
285
print(f"\n=== Performance Streaks ===")
286

287
def find_longest_streak(bitmap_key, bit_value, total_bits):
288
    """Find the longest consecutive streak of a specific bit value"""
289
    max_streak = 0
290
    current_streak = 0
291
    max_start = 0
292
    current_start = 0
293
    
294
    for i in range(total_bits):
295
        if client.getbit(bitmap_key, i) == bit_value:
296
            if current_streak == 0:
297
                current_start = i
298
            current_streak += 1
299
            if current_streak > max_streak:
300
                max_streak = current_streak
301
                max_start = current_start
302
        else:
303
            current_streak = 0
304
    
305
    return max_streak, max_start
306

307
# Find longest good streak
308
good_streak_length, good_streak_start = find_longest_streak(performance_key, 1, 1440)
309
if good_streak_length > 0:
310
    start_hour = good_streak_start // 60
311
    start_min = good_streak_start % 60
312
    end_minute = good_streak_start + good_streak_length - 1
313
    end_hour = end_minute // 60
314
    end_min = end_minute % 60
315
    
316
    print(f"Longest good streak: {good_streak_length} minutes")
317
    print(f"From {start_hour:02d}:{start_min:02d} to {end_hour:02d}:{end_min:02d}")
318

319
# Find longest outage streak  
320
bad_streak_length, bad_streak_start = find_longest_streak(performance_key, 0, 1440)
321
if bad_streak_length > 0:
322
    start_hour = bad_streak_start // 60
323
    start_min = bad_streak_start % 60
324
    end_minute = bad_streak_start + bad_streak_length - 1
325
    end_hour = end_minute // 60
326
    end_min = end_minute % 60
327
    
328
    print(f"Longest outage: {bad_streak_length} minutes")
329
    print(f"From {start_hour:02d}:{start_min:02d} to {end_hour:02d}:{end_min:02d}")
330
```
331

332
### Advanced Bitfield Operations
333

334
```python
335
import fakeredis
336

337
client = fakeredis.FakeRedis()
338

339
print("=== Advanced Bitfield Operations ===")
340

341
# Store multiple counters in a single key using bitfields
342
metrics_key = 'metrics:counters'
343

344
# Initialize counters: 4 x 16-bit unsigned integers
345
# Counter 0: Page views (offset 0)
346
# Counter 1: Unique visitors (offset 16) 
347
# Counter 2: API calls (offset 32)
348
# Counter 3: Errors (offset 48)
349

350
print("=== Initializing Counters ===")
351
result = client.bitfield(
352
    metrics_key,
353
    'SET', 'u16', '0', '1500',    # Page views: 1500
354
    'SET', 'u16', '16', '245',    # Unique visitors: 245  
355
    'SET', 'u16', '32', '892',    # API calls: 892
356
    'SET', 'u16', '48', '7'       # Errors: 7
357
)
358
print(f"Initial values set: {result}")
359

360
# Read all counters
361
print(f"\n=== Reading All Counters ===")
362
values = client.bitfield(
363
    metrics_key,
364
    'GET', 'u16', '0',    # Page views
365
    'GET', 'u16', '16',   # Unique visitors
366
    'GET', 'u16', '32',   # API calls  
367
    'GET', 'u16', '48'    # Errors
368
)
369
print(f"Page views: {values[0]}")
370
print(f"Unique visitors: {values[1]}")  
371
print(f"API calls: {values[2]}")
372
print(f"Errors: {values[3]}")
373

374
# Increment counters atomically
375
print(f"\n=== Incrementing Counters ===")
376
increments = client.bitfield(
377
    metrics_key,
378
    'INCRBY', 'u16', '0', '25',   # +25 page views
379
    'INCRBY', 'u16', '16', '3',   # +3 unique visitors
380
    'INCRBY', 'u16', '32', '47',  # +47 API calls
381
    'INCRBY', 'u16', '48', '2'    # +2 errors
382
)
383
print(f"New values after increment: {increments}")
384

385
# Demonstrate overflow handling
386
print(f"\n=== Overflow Handling ===")
387

388
# Set a counter near its maximum value (65535 for u16)
389
client.bitfield(metrics_key, 'SET', 'u16', '64', '65530')
390

391
# Try different overflow behaviors
392
print("WRAP overflow (default):")
393
wrap_result = client.bitfield(
394
    metrics_key,
395
    'INCRBY', 'u16', '64', '10'  # This will wrap around
396
)
397
print(f"65530 + 10 with WRAP = {wrap_result[0]}")
398

399
# Reset and try FAIL overflow
400
client.bitfield(metrics_key, 'SET', 'u16', '64', '65530')
401
print("\nFAIL overflow:")
402
fail_result = client.bitfield(
403
    metrics_key,
404
    'OVERFLOW', 'FAIL',
405
    'INCRBY', 'u16', '64', '10'  # This will return None
406
)
407
print(f"65530 + 10 with FAIL = {fail_result[0]}")
408

409
# Reset and try SAT overflow  
410
client.bitfield(metrics_key, 'SET', 'u16', '64', '65530')
411
print("\nSAT overflow:")
412
sat_result = client.bitfield(
413
    metrics_key,
414
    'OVERFLOW', 'SAT',
415
    'INCRBY', 'u16', '64', '10'  # This will saturate at 65535
416
)
417
print(f"65530 + 10 with SAT = {sat_result[0]}")
418

419
# Working with signed integers
420
print(f"\n=== Signed Integer Operations ===")
421
signed_key = 'signed:temperatures'
422

423
# Store temperature readings as signed 8-bit integers (-128 to 127)
424
temperatures = [22, -5, 15, -12, 35, -20, 8]
425
print("Storing temperature readings:")
426

427
for i, temp in enumerate(temperatures):
428
    offset = i * 8  # 8 bits per temperature
429
    client.bitfield(signed_key, 'SET', 'i8', str(offset), str(temp))
430
    print(f"Sensor {i}: {temp}°C")
431

432
# Read all temperatures
433
print(f"\nReading all temperatures:")
434
temp_commands = []
435
for i in range(len(temperatures)):
436
    temp_commands.extend(['GET', 'i8', str(i * 8)])
437

438
all_temps = client.bitfield(signed_key, *temp_commands)
439
for i, temp in enumerate(all_temps):
440
    print(f"Sensor {i}: {temp}°C")
441

442
# Calculate average temperature increase
443
print(f"\nSimulating temperature changes:")
444
changes = [2, -3, 1, 5, -2, 4, -1]  # Temperature changes
445

446
change_commands = []
447
for i, change in enumerate(changes):
448
    change_commands.extend(['INCRBY', 'i8', str(i * 8), str(change)])
449

450
new_temps = client.bitfield(signed_key, *change_commands)
451
print("New temperatures after changes:")
452
for i, temp in enumerate(new_temps):
453
    original = temperatures[i] 
454
    change = changes[i]
455
    print(f"Sensor {i}: {original}°C + {change}°C = {temp}°C")
456
```
457

458
## Types
459

460
```python { .api }
461
# Bitmap operation types
462
BitmapOperation = Literal["AND", "OR", "XOR", "NOT"]
463

464
# Bitfield encoding types
465
BitfieldType = Union[
466
    str,  # e.g., "u8", "i16", "u32", "i64"
467
    bytes  # e.g., b"u8", b"i16"
468
]
469

470
# Bitfield overflow policies
471
OverflowPolicy = Literal["WRAP", "SAT", "FAIL"]
472

473
# Bitfield operation result
474
BitfieldResult = List[Optional[int]]
475
```