Python

Installation

pip install autumnlabs --extra-index-url https://client.autumnlabs.io/get/sdk/python

Quick Start

from autumnlabs.station import AutumnLabsStation

# Initialize a station instance
station = AutumnLabsStation()

# Track a unit coming in
station.track_unit_in("UNIT123", tags=["batch1"], slot="slot1")

# Push a station metric
station.push_metric("temperature", 25.5, value_unit="C", tags=["sensor1"])

# Push a unit metric
station.push_metric("temperature", 25.5, value_unit="C", tags=["sensor1"], unit_serial_number="UNIT123")

# Set station state
station.set_station_state("active", "Station is processing units")

# Track unit going out
station.track_unit_out("UNIT123", result="pass", tags=["completed"])

# Set station state
station.set_station_state("idle", "Station is ready to process units")

Constructor

AutumnLabsStation(
    station_alias: str = None,
    mode: str = None,
    state: str = None,
    debug: bool = False,
    fetch_remote: bool = True,
    fetch_timeout_sec: int = 5
)

Parameters:

station_alias (str, optional): Alias for the station. Run "al info" to get the station alias.
mode (str, optional): Operational mode of the station (e.g., "online", "offline").
state (str, optional): Initial state of the station (e.g., "active", "error", "idle", "maintenance").
debug (bool, optional): Enable debug mode for detailed logging. Defaults to False.
fetch_remote (bool, optional): Whether to fetch variables and limits remotely. Defaults to True.
fetch_timeout_sec (int, optional): Timeout duration in seconds for fetching remote data. Defaults to 5.

Example:

station = AutumnLabsStation(
    station_alias="station-alias-1",
    mode="production",
    state="active",
    debug=True,
    fetch_remote=True,
    fetch_timeout_sec=10
)

Management

`get_station_mode()`

Retrieve the current operational mode of the station.

Returns: str - The current mode of the station.

Example:

current_mode = station.get_station_mode()
print(f"Station mode: {current_mode}")

`set_station_mode(mode)`

Set the operational mode of the station and log the mode change.

Parameters:

mode (str): The new mode to set for the station.

Example:

station.set_station_mode("maintenance")

`get_station_state()`

Retrieve the current state of the station.

Returns: str - The current state of the station.

Example:

current_state = station.get_station_state()
print(f"Station state: {current_state}")

`set_station_state(state_name, state_description, tags, slot, unit_serial_number)`

Set the state of the station and update the internal state variable.

Parameters:

state_name (str): The new state to set for the station.
state_description (str, optional): A description of the state.
tags (list[str], optional): A list of tags associated with the state.
slot (str, optional): The slot associated with the state.
unit_serial_number (str, optional): The unit serial number associated with the state.

Example:

station.set_station_state(
    "processing",
    "Station is actively processing units",
    tags=["production", "active"],
    slot="slot1"
)

Unit Tracking

`track_unit_in(serial, tags, slot)`

Track a unit being brought into the station.

Parameters:

serial (str): The serial number of the unit being tracked.
tags (list[str], optional): A list of tags associated with the unit.
slot (str, optional): The slot associated with the unit. Defaults to empty string.

Example:

station.track_unit_in(
    "UNIT-12345",
    tags=["batch-A", "priority"],
    slot="slot1"
)

`track_unit_out(serial, result, tags, slot)`

Track a unit being removed from the station.

Parameters:

serial (str): The serial number of the unit being tracked.
result (str, optional): The result of the unit's process (e.g., "passed", "failed").
tags (list[str], optional): A list of tags associated with the unit.
slot (str, optional): The slot associated with the unit.

Example:

station.track_unit_out(
    "UNIT-12345",
    result="passed",
    tags=["completed", "quality-check"],
    slot="slot1"
)

`get_units()`

Retrieve the list of unit serial numbers currently associated with the station.

Returns: list[dict] - A list of dictionaries, each containing unit serial number details.

Example:

units = station.get_units()
for unit in units:
    print(f"Serial: {unit['serial']}, Slot: {unit['slot']}")

Logging

`push_log(message, name, level, tags, slot, unit_serial_number)`

Push a log message to the system with optional metadata.

Parameters:

message (str): The log message to be pushed.
name (str, optional): The name associated with the log message.
level (str, optional): The level of the log message. Options: trace, debug, info, warn, error, fatal, panic. Defaults to "info".
tags (list[str], optional): A list of tags associated with the log message.
slot (str, optional): The slot associated with the log message.
unit_serial_number (str, optional): The unit serial number associated with the log message.

Example:

station.push_log(
    "Temperature sensor reading completed",
    name="SensorReading",
    level="info",
    tags=["sensor", "temperature"],
    slot="slot1",
    unit_serial_number="UNIT-12345"
)

Metrics

`push_metric(name, value, value_unit, lower_limit, upper_limit, tags, slot, unit_serial_number, buffer)`

Push a metric to the system with optional metadata.

Parameters:

name (str): The name of the metric.
value (object): The value of the metric.
value_unit (str, optional): The unit of the metric value.
lower_limit (object, optional): The lower limit for the metric.
upper_limit (object, optional): The upper limit for the metric.
tags (list[str], optional): A list of tags associated with the metric.
slot (str, optional): The slot associated with the metric.
unit_serial_number (str, optional): The unit serial number associated with the metric.
buffer (bool, optional): Flag to determine if the metric should be buffered. Defaults to False.

Example:

# Push immediate metric
station.push_metric(
    "temperature",
    25.5,
    value_unit="Celsius",
    lower_limit=20.0,
    upper_limit=30.0,
    tags=["sensor1", "environment"],
    slot="slot1",
    unit_serial_number="UNIT-12345"
)

# Buffer metric for bulk push
station.push_metric(
    "pressure",
    1013.25,
    value_unit="hPa",
    tags=["barometer"],
    buffer=True
)

`push_buffered_metrics()`

Push all buffered metrics to the system and clear the buffer.

Example:

# Add several buffered metrics
station.push_metric("temp1", 23.1, buffer=True)
station.push_metric("temp2", 24.2, buffer=True)
station.push_metric("temp3", 25.3, buffer=True)

# Push all buffered metrics at once
station.push_buffered_metrics()

`get_buffered_metrics()`

Retrieve the current list of bulk buffered metrics.

Returns: list[dict] - A list of dictionaries, each representing a buffered metric.

Example:

buffered = station.get_buffered_metrics()
print(f"Number of buffered metrics: {len(buffered)}")

Asset Management

`push_asset(path, tags, slot, unit_serial_number)`

Push an asset to the system with optional metadata.

Parameters:

path (str): The file path of the asset to be pushed.
tags (list[str], optional): A list of tags associated with the asset.
slot (str, optional): The slot associated with the asset.
unit_serial_number (str, optional): The unit serial number associated with the asset.

Returns: tuple[bool, str] - Success status and message.

Example:

success, message = station.push_asset(
    "/path/to/image.png",
    unit_serial_number="UNIT-12345"
)

Data Retrieval

`pull_variables(fetch_remote, fetch_timeout_sec)`

Retrieve and parse station variables.

Parameters:

fetch_remote (bool, optional): Whether to fetch variables remotely. Defaults to False.
fetch_timeout_sec (int, optional): Timeout duration in seconds. Defaults to 5.

Returns: dict - A dictionary of parsed station variables.

Example:

variables = station.pull_variables(fetch_remote=True, fetch_timeout_sec=10)
for var_name, var_info in variables.items():
    print(f"{var_name}: {var_info['value']} ({var_info['type']})")

`pull_limits(fetch_remote, fetch_timeout_sec)`

Retrieve and parse station limits.

Parameters:

fetch_remote (bool, optional): Whether to fetch limits remotely. Defaults to False.
fetch_timeout_sec (int, optional): Timeout duration in seconds. Defaults to 5.

Returns: dict - A dictionary of parsed station limits.

Example:

limits = station.pull_limits(fetch_remote=True)
for limit_name, limit_info in limits.items():
    print(f"{limit_name}: {limit_info['lower_limit']} - {limit_info['upper_limit']} {limit_info['limit_unit']}")

`update_station_inputs(timeout_sec)`

Update the station's variables and limits with a specified timeout.

Parameters:

timeout_sec (int, optional): Timeout duration in seconds. Defaults to 5.

Returns: bool - True if both variables and limits were successfully updated.

Example:

success = station.update_station_inputs(timeout_sec=10)
if success:
    print("Station inputs updated successfully")
else:
    print("Failed to update station inputs")

Usage

from autumnlabs.station import AutumnLabsStation
import time

# Initialize station
station = AutumnLabsStation()

try:
    # Set initial state
    station.set_station_state("active", "Station ready for processing")

    # Track unit coming in
    unit_serial = "UNIT-12345"
    station.track_unit_in(unit_serial, tags=["batch-A"], slot="A1")

    # Log start of processing
    station.push_log(
        f"Started processing unit {unit_serial}",
        name="ProcessStart",
        level="info"
    )

    # Simulate processing with metrics
    for i in range(5):
        temperature = 25.0 + i * 0.5
        station.push_metric(
            "temperature",
            temperature,
            value_unit="Celsius",
            lower_limit=20.0,
            upper_limit=30.0,
            unit_serial_number=unit_serial,
            buffer=True
        )
        time.sleep(1)

    # Upload all buffered metrics
    station.push_buffered_metrics()

    # Upload asset
    success, msg = station.push_asset(
        "/path/to/image.png",
        unit_serial_number=unit_serial
    )

    # Track unit going out
    station.track_unit_out(unit_serial, result="passed", tags=["completed"])

    # Set final state
    station.set_station_state("idle", "Processing completed")

except Exception as e:
    station.push_log(f"Error occurred: {str(e)}", level="error")
    station.set_station_state("error", f"Station error: {str(e)}")

Best Practices

Error Handling: Always wrap station operations in try-catch blocks and log errors appropriately.
Resource Management: The station object automatically handles cleanup when destroyed, but call push_buffered_metrics() before termination to ensure all metrics are sent.
Buffered Metrics: Use buffered metrics for high-frequency data collection, then push in batches for better performance.
State Management: Always update station state to reflect current operations for better monitoring and debugging.
Logging: Use appropriate log levels and include relevant context (serial numbers, slots, tags) for better traceability.
Asset Management: Ensure file paths exist before calling push_asset() and handle the returned success status.

Error Handling

The SDK includes comprehensive error handling and logging. Enable debug mode during development to see detailed logs:

station = AutumnLabsStation(debug=True)

All file operations and API calls are wrapped in error handling, with failures logged appropriately. Check the logs at ~/.autumnlabs/logs/sdk_py.log for troubleshooting.

Python

On this page