Skip to content

overlay_task

Abstract base classes for overlay tasks (OverlayTask, DaemonTask) and a reference BlinkLedTask implementation.

rfsoc_rfdc.overlay_task

Attributes

TASK_STATE = {'IDLE': 0, 'RUNNING': 1, 'PAUSE': 2, 'STOP': 3} module-attribute

Classes

OverlayTask(overlay, name='OverlayTask')

Bases: ABC

An abstract base class for creating tasks to be run on an RFSoCOverlay.

Attributes:

Name Type Description
ol RFSoCOverlay

An instance of RFSoCOverlay to operate on.
task_name str

The name of the task.
thread Thread

The thread on which the task runs.

Methods:

Name Description
run

An abstract method to define the task's behavior.
start

Starts the task's thread.
join

Waits for the task's thread to complete.

Initializes the OverlayTask with a given RFSoCOverlay instance and task name.

Parameters:

Name Type Description Default
overlay RFSoCOverlay

The RFSoCOverlay instance to operate on.

 required
name str

The name of the task. Defaults to "OverlayTask".

 'OverlayTask'

Raises:

Type Description
TypeError

If the overlay is not an instance of RFSoCOverlay.
Source code in rfsoc_rfdc/overlay_task.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self, overlay, name="OverlayTask"):
    """
    Initializes the OverlayTask with a given RFSoCOverlay instance and task name.

    Args:
        overlay (RFSoCOverlay): The RFSoCOverlay instance to operate on.
        name (str): The name of the task. Defaults to "OverlayTask".

    Raises:
        TypeError: If the overlay is not an instance of RFSoCOverlay.
    """
    if not isinstance(overlay, RFSoCOverlay):
        raise TypeError("This task is not an RFSoCOverlay instance.")
    self.ol = overlay
    self.task_name = name
    self.task_state = TASK_STATE['IDLE']
    self.thread = threading.Thread(target=self.run)
Attributes
ol = overlay instance-attribute
task_name = name instance-attribute
task_state = TASK_STATE['IDLE'] instance-attribute
thread = threading.Thread(target=(self.run)) instance-attribute
Functions
run() abstractmethod

Abstract method that defines the task's behavior. Must be implemented by subclasses.

Source code in rfsoc_rfdc/overlay_task.py 
51
52
53
54
55
56
@abstractmethod
def run(self):
    """
    Abstract method that defines the task's behavior. Must be implemented by subclasses.
    """
    pass
start()

Starts the task's thread, causing it to run concurrently

Source code in rfsoc_rfdc/overlay_task.py 
58
59
60
61
def start(self):
    """Starts the task's thread, causing it to run concurrently"""
    self.thread.start()
    self.task_state = TASK_STATE['RUNNING']
pause()

Pause task execution by changing task state

Source code in rfsoc_rfdc/overlay_task.py 
63
64
65
66
67
def pause(self):
    """Pause task execution by changing task state"""
    if self.task_state != TASK_STATE['RUNNING']:
        raise Exception(f"Task is not running")
    self.task_state = TASK_STATE['PAUSE']
resume()

Resume task execution by changing task state

Source code in rfsoc_rfdc/overlay_task.py 
69
70
71
72
73
def resume(self):
    """Resume task execution by changing task state"""
    if self.task_state != TASK_STATE['PAUSE']:
        raise Exception(f"Task is not paused")
    self.task_state = TASK_STATE['RUNNING']
stop()

Stop task execution

Source code in rfsoc_rfdc/overlay_task.py 
75
76
77
78
def stop(self):
    """Stop task execution"""
    self.task_state = TASK_STATE['STOP']
    self.thread.join()
join()

Blocks until the task's thread terminates

Source code in rfsoc_rfdc/overlay_task.py 
80
81
82
def join(self):
    """Blocks until the task's thread terminates"""
    self.thread.join()

DaemonTask

Bases: ABC

Functions
generate_memory() abstractmethod
Source code in rfsoc_rfdc/overlay_task.py 
87
88
89
@abstractmethod
def generate_memory(self):
    pass

BlinkLedTask(overlay)

Bases: OverlayTask

A task that blinks LEDs on an RFSoCOverlay.

Inherits from OverlayTask.

Attributes:

Name Type Description
green_leds AxiGPIO

AxiGPIO instance for controlling green LEDs.
red_leds AxiGPIO

AxiGPIO instance for controlling red LEDs.

Methods:

Name Description
run

Implements the LED blinking behavior.

Initializes the BlinkLedTask with a given RFSoCOverlay instance.

Parameters:

Name Type Description Default
overlay RFSoCOverlay

The RFSoCOverlay instance to operate on.

 required
Source code in rfsoc_rfdc/overlay_task.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
def __init__(self, overlay):
    """
    Initializes the BlinkLedTask with a given RFSoCOverlay instance.

    Args:
        overlay (RFSoCOverlay): The RFSoCOverlay instance to operate on.
    """
    super().__init__(overlay, name="BlinkLedTask")
    self.green_leds = AxiGPIO(self.ol.ip_dict['axi_gpio_led']).channel1
    self.red_leds = AxiGPIO(self.ol.ip_dict['axi_gpio_led']).channel2
    for leds in [self.red_leds, self.green_leds]:
        leds.setdirection("out")
        leds.setlength(8)
Attributes
green_leds = AxiGPIO(self.ol.ip_dict['axi_gpio_led']).channel1 instance-attribute
red_leds = AxiGPIO(self.ol.ip_dict['axi_gpio_led']).channel2 instance-attribute
Functions
run()

Runs the LED blinking task. Alternates the LEDs between on and off states at a fixed interval.

Source code in rfsoc_rfdc/overlay_task.py 
120
121
122
123
124
125
126
127
128
129
130
def run(self):
    """
    Runs the LED blinking task. Alternates the LEDs between on and off states at a fixed interval.
    """
    interval = 0.3
    while self.task_state != TASK_STATE["STOP"]:
        if self.task_state == TASK_STATE["RUNNING"]:
            self.green_leds.write(0xff, 0xff)  # Turn on all green LEDs
            time.sleep(interval)
            self.green_leds.write(0x00, 0xff)  # Turn off all green LEDs
            time.sleep(interval)