0
# Logging and Utilities
1

2
Cocotb's logging and utilities system provides simulation-aware logging, time manipulation functions, queue implementations, and debugging tools. The system integrates simulation time context with Python's logging framework and offers utilities for testbench development.
3

4
## Capabilities
5

6
### Simulation-Aware Logging
7

8
Logging system with simulation time context and hierarchical naming for testbench debugging.
9

10
```python { .api }
11
class SimLog:
12
    """
13
    Simulator-aware logger with hierarchical naming.
14
    
15
    Automatically adds simulation time and hierarchical context to log messages.
16
    Inherits from Python's logging.Logger with simulation-specific enhancements.
17
    
18
    Methods:
19
    - debug(msg, *args, **kwargs): Debug level logging
20
    - info(msg, *args, **kwargs): Info level logging  
21
    - warning(msg, *args, **kwargs): Warning level logging
22
    - error(msg, *args, **kwargs): Error level logging
23
    - critical(msg, *args, **kwargs): Critical level logging
24
    """
25

26
class SimLogFormatter:
27
    """Custom log formatter for simulator logs."""
28

29
class SimColourLogFormatter(SimLogFormatter):
30
    """Colored log formatter for enhanced readability."""
31

32
class SimTimeContextFilter:
33
    """Filter to add simulation time context to log records."""
34

35
def default_config():
36
    """
37
    Apply default logging configuration for cocotb.
38
    
39
    Sets up simulation-aware logging with appropriate formatters
40
    and filters for testbench development.
41
    """
42
```
43

44
**Usage Examples:**
45

46
```python
47
@cocotb.test()
48
async def logging_test(dut):
49
    """Test demonstrating simulation-aware logging."""
50
    
51
    # Use cocotb's global logger
52
    cocotb.log.info("Test started")
53
    
54
    # Create hierarchical loggers
55
    test_log = SimLog("test.main")
56
    dut_log = SimLog("test.dut")  
57
    monitor_log = SimLog("test.monitor.data_bus")
58
    
59
    # Log at different levels with time stamps
60
    test_log.debug("Debug message - detailed info")
61
    test_log.info("Test phase 1 starting")
62
    test_log.warning("Potential timing issue detected")
63
    
64
    # Simulate some time passing
65
    await Timer(100, units="ns")
66
    
67
    # Logs automatically include simulation time
68
    dut_log.info(f"DUT state: {dut.state.value}")
69
    monitor_log.info(f"Data bus activity: {dut.data_bus.value}")
70
    
71
    await Timer(50, units="ns")
72
    
73
    # Error logging
74
    if dut.error_flag.value:
75
        test_log.error("Error condition detected!")
76
    
77
    test_log.info("Test completed successfully")
78

79
@cocotb.test()
80
async def hierarchical_logging_test(dut):
81
    """Test demonstrating hierarchical logging structure."""
82
    
83
    # Create logger hierarchy for complex testbench
84
    main_log = SimLog("testbench")
85
    cpu_log = SimLog("testbench.cpu")
86
    memory_log = SimLog("testbench.memory")
87
    bus_log = SimLog("testbench.interconnect.bus")
88
    
89
    # Coordinated logging from different components
90
    main_log.info("Starting system-level test")
91
    
92
    cpu_log.info("CPU initialization")
93
    await Timer(25, units="ns")
94
    
95
    memory_log.info("Memory subsystem ready")
96
    await Timer(25, units="ns")
97
    
98
    bus_log.info("Bus protocol check")
99
    await Timer(25, units="ns")
100
    
101
    main_log.info("All subsystems initialized")
102
    
103
    # Component-specific detailed logging
104
    for i in range(3):
105
        cpu_log.debug(f"Instruction fetch cycle {i}")
106
        bus_log.debug(f"Bus transaction {i}: read request")
107
        memory_log.debug(f"Memory read from address 0x{i*4:08X}")
108
        await Timer(20, units="ns")
109
    
110
    main_log.info("Test sequence completed")
111
```
112

113
### Time Utilities
114

115
Functions for simulation time manipulation and conversion between different time units.
116

117
```python { .api }
118
def get_sim_time(units="step"):
119
    """
120
    Get current simulation time.
121
    
122
    Parameters:
123
    - units: Time units to return ("fs", "ps", "ns", "us", "ms", "sec", "step")
124
    
125
    Returns:
126
    Current simulation time in specified units
127
    """
128

129
def get_time_from_sim_steps(steps, units):
130
    """
131
    Convert simulation steps to time units.
132
    
133
    Parameters:
134
    - steps: Number of simulation steps
135
    - units: Target time units
136
    
137
    Returns:
138
    Time value in specified units
139
    """
140

141
def get_sim_steps(time, units, round_mode=None):
142
    """
143
    Convert time to simulation steps.
144
    
145
    Parameters:
146
    - time: Time value
147
    - units: Time units of input
148
    - round_mode: Rounding mode for conversion
149
    
150
    Returns:
151
    Number of simulation steps
152
    """
153
```
154

155
**Usage Examples:**
156

157
```python
158
@cocotb.test()
159
async def time_utilities_test(dut):
160
    """Test demonstrating time utility functions."""
161
    
162
    # Record start time
163
    start_time_ns = get_sim_time("ns")
164
    start_time_ps = get_sim_time("ps")
165
    start_time_steps = get_sim_time("step")
166
    
167
    cocotb.log.info(f"Test start: {start_time_ns} ns, {start_time_ps} ps, {start_time_steps} steps")
168
    
169
    # Simulate some time
170
    await Timer(150, units="ns")
171
    
172
    # Check elapsed time
173
    current_time_ns = get_sim_time("ns")
174
    elapsed_ns = current_time_ns - start_time_ns
175
    
176
    cocotb.log.info(f"Elapsed time: {elapsed_ns} ns")
177
    
178
    # Convert between units
179
    steps_in_100ns = get_sim_steps(100, "ns")
180
    time_from_steps = get_time_from_sim_steps(steps_in_100ns, "ps")
181
    
182
    cocotb.log.info(f"100ns = {steps_in_100ns} steps = {time_from_steps} ps")
183
    
184
    # Performance measurement
185
    operation_start = get_sim_time("ps")
186
    
187
    # Simulate some operation
188
    for i in range(10):
189
        dut.data.value = i
190
        await Timer(5, units="ns")
191
    
192
    operation_end = get_sim_time("ps")
193
    operation_duration = operation_end - operation_start
194
    
195
    cocotb.log.info(f"Operation took {operation_duration} ps ({operation_duration/1000} ns)")
196

197
@cocotb.test()
198
async def timing_analysis_test(dut):
199
    """Test demonstrating timing analysis with utilities."""
200
    
201
    # Measure setup and hold times
202
    timestamps = []
203
    
204
    # Clock edge timing
205
    for cycle in range(5):
206
        await RisingEdge(dut.clk)
207
        edge_time = get_sim_time("ps")
208
        timestamps.append(edge_time)
209
        
210
        # Setup time check
211
        dut.data_in.value = cycle
212
        setup_time = get_sim_time("ps")
213
        
214
        cocotb.log.info(f"Cycle {cycle}: Clock edge at {edge_time} ps, data setup at {setup_time} ps")
215
        
216
        # Check setup timing
217
        setup_margin = edge_time - setup_time  # Should be negative (data before clock)
218
        if setup_margin > -1000:  # Less than 1ns setup
219
            cocotb.log.warning(f"Setup time violation: {setup_margin} ps")
220
    
221
    # Calculate clock period statistics
222
    periods = []
223
    for i in range(1, len(timestamps)):
224
        period = timestamps[i] - timestamps[i-1]
225
        periods.append(period)
226
    
227
    avg_period = sum(periods) / len(periods)
228
    frequency_mhz = 1e6 / avg_period  # Convert ps to MHz
229
    
230
    cocotb.log.info(f"Average clock period: {avg_period} ps ({frequency_mhz:.2f} MHz)")
231
```
232

233
### Queue System
234

235
Async queue implementations for producer-consumer patterns and inter-coroutine communication.
236

237
```python { .api }
238
class Queue[T](maxsize=0):
239
    """
240
    Generic async queue implementation.
241
    
242
    Parameters:
243
    - maxsize: Maximum queue size (0 for unlimited)
244
    
245
    Methods:
246
    - put(item): Put item into queue (async)
247
    - get(): Get item from queue (async)  
248
    - put_nowait(item): Put item without waiting
249
    - get_nowait(): Get item without waiting
250
    - qsize(): Get current queue size
251
    - empty(): Check if queue is empty
252
    - full(): Check if queue is full
253
    """
254
    
255
    async def put(self, item):
256
        """
257
        Put item into queue.
258
        
259
        Parameters:
260
        - item: Item to add to queue
261
        
262
        Blocks if queue is full until space available.
263
        """
264
    
265
    async def get(self):
266
        """
267
        Get item from queue.
268
        
269
        Returns:
270
        Item from queue
271
        
272
        Blocks if queue is empty until item available.
273
        """
274
    
275
    def put_nowait(self, item):
276
        """
277
        Put item without waiting.
278
        
279
        Parameters:
280
        - item: Item to add
281
        
282
        Raises:
283
        QueueFull: If queue is at capacity
284
        """
285
    
286
    def get_nowait(self):
287
        """
288
        Get item without waiting.
289
        
290
        Returns:
291
        Item from queue
292
        
293
        Raises:
294
        QueueEmpty: If queue is empty
295
        """
296
    
297
    def qsize(self) -> int:
298
        """Get current number of items in queue."""
299
    
300
    @property
301
    def maxsize(self) -> int:
302
        """Get maximum number of items allowed in queue."""
303
    
304
    def empty(self) -> bool:
305
        """Check if queue is empty."""
306
    
307
    def full(self) -> bool:
308
        """Check if queue is full."""
309

310
class PriorityQueue(maxsize=0):
311
    """
312
    Priority queue implementation (smallest item first).
313
    
314
    Items should be comparable or tuples of (priority, item).
315
    """
316

317
class LifoQueue(maxsize=0):
318
    """
319
    LIFO queue (stack) implementation.
320
    
321
    Last item put is first item retrieved.
322
    """
323

324
class QueueFull(Exception):
325
    """Raised when queue is full and put_nowait() is called."""
326

327
class QueueEmpty(Exception):
328
    """Raised when queue is empty and get_nowait() is called."""
329
```
330

331
**Usage Examples:**
332

333
```python
334
@cocotb.test()
335
async def queue_basic_test(dut):
336
    """Test demonstrating basic queue operations."""
337
    
338
    # Create bounded queue
339
    data_queue = Queue(maxsize=5)
340
    
341
    # Producer coroutine
342
    async def producer():
343
        for i in range(10):
344
            await data_queue.put(f"data_{i}")
345
            cocotb.log.info(f"Produced: data_{i} (queue size: {data_queue.qsize()})")
346
            await Timer(20, units="ns")
347
    
348
    # Consumer coroutine
349
    async def consumer():
350
        consumed = []
351
        while len(consumed) < 10:
352
            item = await data_queue.get()
353
            consumed.append(item)
354
            cocotb.log.info(f"Consumed: {item} (queue size: {data_queue.qsize()})")
355
            await Timer(30, units="ns")  # Slower than producer
356
        return consumed
357
    
358
    # Run producer and consumer concurrently
359
    producer_task = cocotb.start_soon(producer())
360
    consumer_task = cocotb.start_soon(consumer())
361
    
362
    # Wait for completion
363
    await producer_task.join()
364
    consumed_items = await consumer_task.join()
365
    
366
    cocotb.log.info(f"Test completed. Consumed {len(consumed_items)} items")
367

368
@cocotb.test()
369
async def queue_types_test(dut):
370
    """Test demonstrating different queue types."""
371
    
372
    # FIFO queue (default)
373
    fifo_queue = Queue()
374
    
375
    # LIFO queue (stack)
376
    lifo_queue = LifoQueue()
377
    
378
    # Priority queue
379
    priority_queue = PriorityQueue()
380
    
381
    # Fill queues with test data
382
    test_items = ["first", "second", "third"]
383
    priorities = [(3, "low"), (1, "high"), (2, "medium")]
384
    
385
    # FIFO operations
386
    for item in test_items:
387
        await fifo_queue.put(item)
388
    
389
    fifo_result = []
390
    while not fifo_queue.empty():
391
        fifo_result.append(await fifo_queue.get())
392
    
393
    # LIFO operations  
394
    for item in test_items:
395
        await lifo_queue.put(item)
396
    
397
    lifo_result = []
398
    while not lifo_queue.empty():
399
        lifo_result.append(await lifo_queue.get())
400
    
401
    # Priority operations
402
    for priority_item in priorities:
403
        await priority_queue.put(priority_item)
404
    
405
    priority_result = []
406
    while not priority_queue.empty():
407
        priority_result.append(await priority_queue.get())
408
    
409
    # Log results
410
    cocotb.log.info(f"FIFO order: {fifo_result}")
411
    cocotb.log.info(f"LIFO order: {lifo_result}")
412
    cocotb.log.info(f"Priority order: {priority_result}")
413

414
@cocotb.test()
415
async def queue_flow_control_test(dut):
416
    """Test demonstrating queue flow control."""
417
    
418
    # Small queue to test flow control
419
    flow_queue = Queue(maxsize=2)
420
    
421
    # Fast producer that will block
422
    async def fast_producer():
423
        for i in range(6):
424
            cocotb.log.info(f"Attempting to put item_{i}")
425
            await flow_queue.put(f"item_{i}")
426
            cocotb.log.info(f"Successfully put item_{i}")
427
            await Timer(10, units="ns")
428
    
429
    # Slow consumer
430
    async def slow_consumer():
431
        consumed = []
432
        for _ in range(6):
433
            await Timer(50, units="ns")  # Slow processing
434
            item = await flow_queue.get()
435
            consumed.append(item)
436
            cocotb.log.info(f"Consumed: {item}")
437
        return consumed
438
    
439
    # Run with flow control
440
    producer_task = cocotb.start_soon(fast_producer())
441
    consumer_task = cocotb.start_soon(slow_consumer())
442
    
443
    await consumer_task.join()
444
    await producer_task.join()
445
    
446
    cocotb.log.info("Flow control test completed")
447
```
448

449
### Debugging Utilities
450

451
Tools for debugging coroutines, handling tracebacks, and development support.
452

453
```python { .api }
454
def want_color_output() -> bool:
455
    """
456
    Check if colored terminal output is wanted.
457
    
458
    Returns:
459
    True if colored output should be used
460
    """
461

462
def remove_traceback_frames(tb_or_exc, frame_names):
463
    """
464
    Strip specific frames from traceback for cleaner error reporting.
465
    
466
    Parameters:
467
    - tb_or_exc: Traceback or exception object
468
    - frame_names: List of frame names to remove
469
    
470
    Returns:
471
    Cleaned traceback
472
    """
473

474
def walk_coro_stack(coro):
475
    """
476
    Walk coroutine call stack for debugging.
477
    
478
    Parameters:
479
    - coro: Coroutine to examine
480
    
481
    Returns:
482
    Iterator over stack frames
483
    """
484

485
def extract_coro_stack(coro, limit=None):
486
    """
487
    Extract coroutine call stack as list.
488
    
489
    Parameters:
490
    - coro: Coroutine to examine
491
    - limit: Maximum number of frames
492
    
493
    Returns:
494
    List of stack frames
495
    """
496
```
497

498
**Usage Examples:**
499

500
```python
501
@cocotb.test()
502
async def debugging_test(dut):
503
    """Test demonstrating debugging utilities."""
504
    
505
    # Check terminal capabilities
506
    if want_color_output():
507
        cocotb.log.info("Color output is available")
508
    else:
509
        cocotb.log.info("Color output not available")
510
    
511
    # Start a coroutine for stack inspection
512
    debug_task = cocotb.start_soon(nested_coroutine_example(dut))
513
    
514
    # Allow some execution
515
    await Timer(50, units="ns")
516
    
517
    # Extract coroutine stack
518
    if not debug_task.done():
519
        stack = extract_coro_stack(debug_task._handle, limit=5)
520
        cocotb.log.info(f"Coroutine stack has {len(stack)} frames")
521
        
522
        for i, frame in enumerate(stack):
523
            cocotb.log.info(f"  Frame {i}: {frame}")
524
    
525
    # Wait for completion
526
    await debug_task.join()
527

528
async def nested_coroutine_example(dut):
529
    """Example nested coroutine for stack inspection."""
530
    
531
    async def level_1():
532
        await level_2()
533
    
534
    async def level_2():
535
        await level_3()
536
    
537
    async def level_3():
538
        # This creates a deep call stack
539
        await Timer(100, units="ns")
540
        return "completed"
541
    
542
    return await level_1()
543

544
@cocotb.test()
545
async def error_handling_test(dut):
546
    """Test demonstrating error handling utilities."""
547
    
548
    async def failing_coroutine():
549
        await Timer(25, units="ns")
550
        raise ValueError("Intentional test error")
551
    
552
    # Catch and analyze error
553
    try:
554
        await failing_coroutine()
555
    except Exception as e:
556
        cocotb.log.info(f"Caught exception: {e}")
557
        
558
        # Get clean traceback
559
        import traceback
560
        tb = traceback.format_exc()
561
        cocotb.log.info("Full traceback:")
562
        for line in tb.split('\n'):
563
            if line.strip():
564
                cocotb.log.info(f"  {line}")
565
```
566

567
### Deprecated Utilities
568

569
Legacy utility functions maintained for backwards compatibility.
570

571
```python { .api }
572
def hexdump(x):
573
    """
574
    DEPRECATED: Hexdump of buffer.
575
    
576
    Parameters:
577
    - x: Buffer to dump
578
    
579
    Returns:
580
    Hexadecimal representation string
581
    """
582

583
def hexdiffs(x, y):
584
    """
585
    DEPRECATED: Show hex differences between buffers.
586
    
587
    Parameters:
588
    - x: First buffer
589
    - y: Second buffer
590
    
591
    Returns:
592
    String showing differences
593
    """
594

595
def pack(ctypes_obj):
596
    """
597
    DEPRECATED: Convert ctypes object to string.
598
    
599
    Parameters:
600
    - ctypes_obj: Ctypes object to convert
601
    
602
    Returns:
603
    String representation
604
    """
605

606
def unpack(ctypes_obj, string, bytes=None):
607
    """
608
    DEPRECATED: Unpack string to ctypes object.
609
    
610
    Parameters:
611
    - ctypes_obj: Target ctypes object
612
    - string: String to unpack
613
    - bytes: Optional byte specification
614
    
615
    Returns:
616
    Unpacked ctypes object
617
    """
618
```
619

620
### Utility Classes
621

622
Helper classes for advanced patterns and singleton implementations.
623

624
```python { .api }
625
class ParametrizedSingleton(type):
626
    """
627
    Metaclass for parametrized singleton pattern.
628
    
629
    Creates singleton instances based on constructor parameters.
630
    """
631

632
class lazy_property:
633
    """
634
    Property decorator for lazy evaluation.
635
    
636
    Property value is computed only once and cached.
637
    """
638
```
639

640
**Usage Examples:**
641

642
```python
643
class ConfigManager(metaclass=ParametrizedSingleton):
644
    """Example using parametrized singleton."""
645
    
646
    def __init__(self, config_name):
647
        self.config_name = config_name
648
        self.settings = {}
649
        cocotb.log.info(f"Created config manager for: {config_name}")
650

651
class TestBench:
652
    """Example using lazy property."""
653
    
654
    def __init__(self, dut):
655
        self.dut = dut
656
    
657
    @lazy_property
658
    def memory_map(self):
659
        """Expensive computation done only once."""
660
        cocotb.log.info("Computing memory map...")
661
        # Simulate expensive operation
662
        return {"base": 0x1000, "size": 0x1000}
663

664
@cocotb.test()
665
async def utility_classes_test(dut):
666
    """Test demonstrating utility classes."""
667
    
668
    # Singleton pattern test
669
    config1 = ConfigManager("test_config")
670
    config2 = ConfigManager("test_config")  # Same instance
671
    config3 = ConfigManager("other_config")  # Different instance
672
    
673
    assert config1 is config2  # Same singleton
674
    assert config1 is not config3  # Different parameter
675
    
676
    # Lazy property test
677
    tb = TestBench(dut)
678
    
679
    # First access computes value
680
    mem_map1 = tb.memory_map
681
    
682
    # Second access uses cached value
683
    mem_map2 = tb.memory_map
684
    
685
    assert mem_map1 is mem_map2  # Same cached object
686
    cocotb.log.info(f"Memory map: {mem_map1}")
687
```