# Fostrom Device SDK for Python
[Fostrom](https://fostrom.io) is an IoT Cloud Platform built for developers. Monitor and control your fleet of devices, from microcontrollers to industrial IoT. Designed to be simple, secure, and fast. Experience first-class tooling with Device SDKs, type-safe schemas, programmable actions, and more.
The Fostrom Device SDK for Python works with Python 3.8+ and helps you quickly integrate, start monitoring, and controlling your IoT devices in just a few lines of code.
## Installation
```bash
pip install fostrom
```
## Quick Start
```python
from fostrom import Fostrom
# Create SDK instance
fostrom = Fostrom({
"fleet_id": "<your-fleet-id>",
"device_id": "<your-device-id>",
"device_secret": "<your-device-secret>",
})
# Setup mail handler for incoming messages
def handle_mail(mail):
print(f"Received: {mail.name} ({mail.id})")
mail.ack() # Acknowledge the message
fostrom.on_mail = handle_mail
# Connect and start sending data
if fostrom.connect():
print("Connected successfully!")
else:
print("Connection failed - check event handlers for details")
return
# Send sensor data
fostrom.send_datapoint("sensors", {
"temperature": 23.5,
"humidity": 65,
"timestamp": time.time()
})
# Send status messages
fostrom.send_msg("status", {"online": True})
# Keep running to receive events
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("Stopping...")
```
## Features
- **Simple API**: Clean, Pythonic interface with full type annotations
- **Real-time Events**: Automatic event streaming for instant mail delivery
- **Background Agent**: Handles connection management and reconnection automatically
- **Error Handling**: Comprehensive error handling with detailed error messages
- **Mail System**: Send and receive messages with ack/reject/requeue operations
- **Type Safety**: Full typing support with mypy compatibility
## Event Handlers
The SDK provides several event handlers you can customize:
```python
fostrom = Fostrom(config)
# Handle incoming mail
fostrom.on_mail = lambda mail: (
print(f"Mail: {mail.name}"),
mail.ack()
)
# Handle connection events
fostrom.on_connected = lambda: print("Connected to Fostrom!")
fostrom.on_unauthorized = lambda reason, after: print(f"Auth failed: {reason}")
fostrom.on_reconnecting = lambda reason, after: print(f"Reconnecting: {reason}")
```
## API Reference
### Fostrom Class
#### `__init__(config)`
Create a new Fostrom instance.
**Parameters:**
- `config` (dict): Configuration dictionary with:
- `fleet_id` (str): Your fleet ID
- `device_id` (str): Your device ID
- `device_secret` (str): Your device secret
- `log` (bool, optional): Enable logging (default: True)
#### `connect() -> bool`
Connect to Fostrom and start the event stream. Returns True on success, False on failure.
On failure, calls appropriate event handlers (`on_unauthorized` or `on_reconnecting`) instead of raising exceptions.
#### `send_datapoint(name: str, payload: dict) -> None`
Send a datapoint to Fostrom.
#### `send_msg(name: str, payload: dict) -> None`
Send a message to Fostrom.
#### `mailbox_status() -> dict`
Get current mailbox status.
#### `next_mail() -> Mail | None`
Get the next mail from the mailbox.
### Mail Class
#### Properties
- `id` (str): Mail ID
- `name` (str): Mail name/type
- `payload` (dict): Mail payload data
- `mailbox_size` (int): Current mailbox size
#### Methods
- `ack()`: Acknowledge the mail
- `reject()`: Reject the mail
- `requeue()`: Requeue the mail
## Error Handling
The SDK uses `FostromError` for all Fostrom-related errors:
```python
from fostrom import Fostrom, FostromError
try:
fostrom.connect()
except FostromError as e:
print(f"Error [{e.atom}]: {e.message}")
```
## Complete Example
```python
import time
import random
from fostrom import Fostrom, FostromError
def main():
fostrom = Fostrom({
"fleet_id": "your-fleet-id",
"device_id": "your-device-id",
"device_secret": "your-device-secret",
})
# Event handlers
fostrom.on_mail = lambda mail: (
print(f"📧 Mail: {mail.name}"),
mail.ack()
)
fostrom.on_connected = lambda: print("🟢 Connected!")
fostrom.on_unauthorized = lambda r, a: print(f"🔒 Auth failed: {r}")
print("Connecting to Fostrom...")
if fostrom.connect():
# Send periodic sensor data
try:
while True:
data = {
"temperature": round(random.uniform(20, 25), 1),
"humidity": random.randint(40, 80),
"timestamp": time.time()
}
fostrom.send_datapoint("sensors", data)
print(f"Sent: {data}")
time.sleep(10)
except KeyboardInterrupt:
print("🛑 Stopping...")
else:
print("❌ Failed to connect - check event handlers for details")
if __name__ == "__main__":
try:
main()
except FostromError as e:
print(f"❌ Error: [{e.atom}] {e.message}")
```
## Device Agent
The Fostrom Device SDK downloads and runs the Fostrom Device Agent in the background. The Agent is downloaded automatically when the package is installed. The Device Agent starts when `fostrom.connect()` is called and handles all communication with the Fostrom platform.
The Agent runs continuously in the background for optimal reconnection performance. To stop the Agent completely:
```python
Fostrom.stop_agent()
```
**Note**: Your Python program will continue running to receive real-time events from the Agent. Use Ctrl+C to stop your program gracefully.
## Requirements
- Python 3.8+
- Linux or macOS
- Network connectivity
## Links
- **Fostrom Platform**: [https://fostrom.io](https://fostrom.io)
- **Documentation**: [https://docs.fostrom.io/sdk/py](https://docs.fostrom.io/sdk/py)
- **Python SDK**: [https://pypi.org/project/fostrom/](https://pypi.org/project/fostrom/)
## License
Apache 2.0
Raw data
{
"_id": null,
"home_page": null,
"name": "fostrom",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "actions, cloud, datapoint, device, fleet, fostrom, hardware, iot, message, mqtt, raspberry, sdk, websockets",
"author": null,
"author_email": "Fostrom <support@fostrom.io>",
"download_url": "https://files.pythonhosted.org/packages/fb/b7/dd6776f4ca30f58f3cef3a8709dc7c8e38118e3c64a1e30fe0f9a6235c41/fostrom-0.0.4.tar.gz",
"platform": null,
"description": "# Fostrom Device SDK for Python\n\n[Fostrom](https://fostrom.io) is an IoT Cloud Platform built for developers. Monitor and control your fleet of devices, from microcontrollers to industrial IoT. Designed to be simple, secure, and fast. Experience first-class tooling with Device SDKs, type-safe schemas, programmable actions, and more.\n\nThe Fostrom Device SDK for Python works with Python 3.8+ and helps you quickly integrate, start monitoring, and controlling your IoT devices in just a few lines of code.\n\n## Installation\n\n```bash\npip install fostrom\n```\n\n## Quick Start\n\n```python\nfrom fostrom import Fostrom\n\n# Create SDK instance\nfostrom = Fostrom({\n \"fleet_id\": \"<your-fleet-id>\",\n \"device_id\": \"<your-device-id>\",\n \"device_secret\": \"<your-device-secret>\",\n})\n\n# Setup mail handler for incoming messages\ndef handle_mail(mail):\n print(f\"Received: {mail.name} ({mail.id})\")\n mail.ack() # Acknowledge the message\n\nfostrom.on_mail = handle_mail\n\n# Connect and start sending data\nif fostrom.connect():\n print(\"Connected successfully!\")\nelse:\n print(\"Connection failed - check event handlers for details\")\n return\n\n# Send sensor data\nfostrom.send_datapoint(\"sensors\", {\n \"temperature\": 23.5,\n \"humidity\": 65,\n \"timestamp\": time.time()\n})\n\n# Send status messages\nfostrom.send_msg(\"status\", {\"online\": True})\n\n# Keep running to receive events\ntry:\n while True:\n time.sleep(1)\nexcept KeyboardInterrupt:\n print(\"Stopping...\")\n```\n\n## Features\n\n- **Simple API**: Clean, Pythonic interface with full type annotations\n- **Real-time Events**: Automatic event streaming for instant mail delivery\n- **Background Agent**: Handles connection management and reconnection automatically\n- **Error Handling**: Comprehensive error handling with detailed error messages\n- **Mail System**: Send and receive messages with ack/reject/requeue operations\n- **Type Safety**: Full typing support with mypy compatibility\n\n## Event Handlers\n\nThe SDK provides several event handlers you can customize:\n\n```python\nfostrom = Fostrom(config)\n\n# Handle incoming mail\nfostrom.on_mail = lambda mail: (\n print(f\"Mail: {mail.name}\"),\n mail.ack()\n)\n\n# Handle connection events\nfostrom.on_connected = lambda: print(\"Connected to Fostrom!\")\nfostrom.on_unauthorized = lambda reason, after: print(f\"Auth failed: {reason}\")\nfostrom.on_reconnecting = lambda reason, after: print(f\"Reconnecting: {reason}\")\n```\n\n## API Reference\n\n### Fostrom Class\n\n#### `__init__(config)`\nCreate a new Fostrom instance.\n\n**Parameters:**\n- `config` (dict): Configuration dictionary with:\n - `fleet_id` (str): Your fleet ID\n - `device_id` (str): Your device ID\n - `device_secret` (str): Your device secret\n - `log` (bool, optional): Enable logging (default: True)\n\n#### `connect() -> bool`\nConnect to Fostrom and start the event stream. Returns True on success, False on failure.\nOn failure, calls appropriate event handlers (`on_unauthorized` or `on_reconnecting`) instead of raising exceptions.\n\n#### `send_datapoint(name: str, payload: dict) -> None`\nSend a datapoint to Fostrom.\n\n#### `send_msg(name: str, payload: dict) -> None`\nSend a message to Fostrom.\n\n#### `mailbox_status() -> dict`\nGet current mailbox status.\n\n#### `next_mail() -> Mail | None`\nGet the next mail from the mailbox.\n\n\n\n### Mail Class\n\n#### Properties\n- `id` (str): Mail ID\n- `name` (str): Mail name/type\n- `payload` (dict): Mail payload data\n- `mailbox_size` (int): Current mailbox size\n\n#### Methods\n- `ack()`: Acknowledge the mail\n- `reject()`: Reject the mail\n- `requeue()`: Requeue the mail\n\n## Error Handling\n\nThe SDK uses `FostromError` for all Fostrom-related errors:\n\n```python\nfrom fostrom import Fostrom, FostromError\n\ntry:\n fostrom.connect()\nexcept FostromError as e:\n print(f\"Error [{e.atom}]: {e.message}\")\n```\n\n## Complete Example\n\n```python\nimport time\nimport random\nfrom fostrom import Fostrom, FostromError\n\ndef main():\n fostrom = Fostrom({\n \"fleet_id\": \"your-fleet-id\",\n \"device_id\": \"your-device-id\",\n \"device_secret\": \"your-device-secret\",\n })\n\n # Event handlers\n fostrom.on_mail = lambda mail: (\n print(f\"\ud83d\udce7 Mail: {mail.name}\"),\n mail.ack()\n )\n\n fostrom.on_connected = lambda: print(\"\ud83d\udfe2 Connected!\")\n fostrom.on_unauthorized = lambda r, a: print(f\"\ud83d\udd12 Auth failed: {r}\")\n\n print(\"Connecting to Fostrom...\")\n if fostrom.connect():\n # Send periodic sensor data\n try:\n while True:\n data = {\n \"temperature\": round(random.uniform(20, 25), 1),\n \"humidity\": random.randint(40, 80),\n \"timestamp\": time.time()\n }\n\n fostrom.send_datapoint(\"sensors\", data)\n print(f\"Sent: {data}\")\n\n time.sleep(10)\n\n except KeyboardInterrupt:\n print(\"\ud83d\uded1 Stopping...\")\n else:\n print(\"\u274c Failed to connect - check event handlers for details\")\n\nif __name__ == \"__main__\":\n try:\n main()\n except FostromError as e:\n print(f\"\u274c Error: [{e.atom}] {e.message}\")\n```\n\n## Device Agent\n\nThe Fostrom Device SDK downloads and runs the Fostrom Device Agent in the background. The Agent is downloaded automatically when the package is installed. The Device Agent starts when `fostrom.connect()` is called and handles all communication with the Fostrom platform.\n\nThe Agent runs continuously in the background for optimal reconnection performance. To stop the Agent completely:\n\n```python\nFostrom.stop_agent()\n```\n\n**Note**: Your Python program will continue running to receive real-time events from the Agent. Use Ctrl+C to stop your program gracefully.\n\n## Requirements\n\n- Python 3.8+\n- Linux or macOS\n- Network connectivity\n\n## Links\n\n- **Fostrom Platform**: [https://fostrom.io](https://fostrom.io)\n- **Documentation**: [https://docs.fostrom.io/sdk/py](https://docs.fostrom.io/sdk/py)\n- **Python SDK**: [https://pypi.org/project/fostrom/](https://pypi.org/project/fostrom/)\n\n## License\n\nApache 2.0\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Fostrom Device SDK",
"version": "0.0.4",
"project_urls": {
"Fostrom": "https://fostrom.io",
"SDK Docs": "https://docs.fostrom.io/sdk/py"
},
"split_keywords": [
"actions",
" cloud",
" datapoint",
" device",
" fleet",
" fostrom",
" hardware",
" iot",
" message",
" mqtt",
" raspberry",
" sdk",
" websockets"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "c3b3986241b58886c0bbda486a76592ed8774bb7aaca8930ea1cf5ed0a1a004a",
"md5": "82880cb51b2fc57738755fa41237a617",
"sha256": "0878bb3ec5e3f531a6243867a08eb322d2b6293a23353f045c639e90ae07db91"
},
"downloads": -1,
"filename": "fostrom-0.0.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "82880cb51b2fc57738755fa41237a617",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 10057,
"upload_time": "2025-07-18T13:26:39",
"upload_time_iso_8601": "2025-07-18T13:26:39.927011Z",
"url": "https://files.pythonhosted.org/packages/c3/b3/986241b58886c0bbda486a76592ed8774bb7aaca8930ea1cf5ed0a1a004a/fostrom-0.0.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "fbb7dd6776f4ca30f58f3cef3a8709dc7c8e38118e3c64a1e30fe0f9a6235c41",
"md5": "d1b0d86f0bcb45e6dfa10acd39373a77",
"sha256": "ce853e478a7c7bf6496b3f0f175cb8062bad40e6c21ad7fe1c464361f029be2c"
},
"downloads": -1,
"filename": "fostrom-0.0.4.tar.gz",
"has_sig": false,
"md5_digest": "d1b0d86f0bcb45e6dfa10acd39373a77",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 9584,
"upload_time": "2025-07-18T13:26:41",
"upload_time_iso_8601": "2025-07-18T13:26:41.072335Z",
"url": "https://files.pythonhosted.org/packages/fb/b7/dd6776f4ca30f58f3cef3a8709dc7c8e38118e3c64a1e30fe0f9a6235c41/fostrom-0.0.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-18 13:26:41",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "fostrom"
}
JSON