Python
Official documentation for Autumn Labs Python SDK
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.