Class ThreadAttribute

Namespace: Belay.Attributes
Assembly: Belay.Attributes.dll

Marks a method to be executed as a background thread on the MicroPython device. Methods decorated with this attribute run asynchronously on the device without blocking the host application or other device operations.

csharp

[AttributeUsage(AttributeTargets.Method)]
public sealed class ThreadAttribute : Attribute

Inheritance

object ← Attribute ← ThreadAttribute

Inherited Members

Examples

Continuous Sensor Monitoring

public class EnvironmentMonitor : Device
{
    [Thread]
    public async Task StartContinuousMonitoringAsync(int intervalMs = 1000)
    {
        await ExecuteAsync($@"
            import _thread
            import time
            import machine

            def monitor_environment():
                adc = machine.ADC(machine.Pin(26))
                while globals().get('monitoring_active', True):
                    try:
                        # Read sensor values
                        temp = read_temperature(adc)
                        humidity = read_humidity()

                        # Log or transmit data
                        print(f'Temp: C, Humidity: %')

                        time.sleep_ms({intervalMs})
                    except Exception as e:
                        print(f'Monitoring error: ')
                        time.sleep_ms(5000)  # Back off on error

            # Start monitoring thread
            _thread.start_new_thread(monitor_environment, ())
        ");
    }

    [Task]
    public async Task StopMonitoringAsync()
    {
        await ExecuteAsync("monitoring_active = False");
    }
}

Event-Driven Responses

public class ButtonHandler : Device
{
    [Thread]
    public async Task StartButtonWatcherAsync()
    {
        await ExecuteAsync(@"
            import _thread
            import machine
            import time

            def watch_buttons():
                button1 = machine.Pin(2, machine.Pin.IN, machine.Pin.PULL_UP)
                button2 = machine.Pin(3, machine.Pin.IN, machine.Pin.PULL_UP)

                last_state = [True, True]  # Pulled up initially

                while globals().get('button_watching', True):
                    current_state = [button1.value(), button2.value()]

                    # Check for button presses (high to low transition)
                    for i, (last, current) in enumerate(zip(last_state, current_state)):
                        if last and not current:  # Button pressed
                            print(f'Button {i+1} pressed!')
                            handle_button_press(i+1)

                    last_state = current_state
                    time.sleep_ms(50)  # 50ms polling

            _thread.start_new_thread(watch_buttons, ())
        ");
    }
}

Watchdog and Health Monitoring

public class SystemMonitor : Device
{
    [Thread(Name = "system_watchdog")]
    public async Task StartSystemWatchdogAsync()
    {
        await ExecuteAsync(@"
            import _thread
            import machine
            import gc
            import time

            def system_watchdog():
                last_heartbeat = time.ticks_ms()

                while globals().get('watchdog_active', True):
                    try:
                        current_time = time.ticks_ms()

                        # Check system health
                        free_mem = gc.mem_free()
                        if free_mem < 1000:  # Low memory warning
                            print(f'WARNING: Low memory:  bytes')
                            gc.collect()

                        # Check for system heartbeat
                        if time.ticks_diff(current_time, last_heartbeat) > 30000:
                            print('WARNING: No heartbeat for 30 seconds')

                        # Update heartbeat if main loop is responsive
                        if globals().get('system_heartbeat', 0) > last_heartbeat:
                            last_heartbeat = globals()['system_heartbeat']

                        time.sleep_ms(5000)  # Check every 5 seconds

                    except Exception as e:
                        print(f'Watchdog error: {e}')
                        time.sleep_ms(10000)  # Back off on error

            _thread.start_new_thread(system_watchdog, ())
        ");
    }
}

Remarks

The enables long-running or continuous operations to execute on MicroPython devices using the _thread module. This is essential for background monitoring, data collection, or reactive behavior that should run independently of host application calls.

Thread methods are non-blocking from the host perspective - they start the background operation and return immediately. The background code continues to run on the device until explicitly stopped or the device disconnects.

Thread management includes:

Automatic thread lifecycle tracking
Graceful shutdown on device disconnection
Thread monitoring and health checks
Inter-thread communication support

Constructors

ThreadAttribute()

Initializes a new instance of the class.

Properties

AutoRestart

Gets or sets a value indicating whether gets or sets whether the thread should automatically restart if it terminates unexpectedly. When true, the thread will be monitored and restarted if it exits due to an error.

MaxRuntimeMs

Gets or sets the maximum runtime for the thread in milliseconds. If specified, the thread will be automatically terminated after this duration.

Name

Gets or sets the name of the thread for identification and management. If not specified, a name is automatically generated based on the method name.

Priority

Gets or sets the priority level for thread execution. Higher priority threads may receive more CPU time on capable platforms.

Methods

ToString()

Returns a string that represents the current .

Class ThreadAttribute ​

Inheritance ​

Inherited Members ​

Examples ​

Remarks ​

Constructors ​

Properties ​

Methods ​