杰瑞科技汇

SnapConnect Python如何快速上手?

Of course! Let's dive deep into snapconnect, which is the Python library for communicating with the Particle Photon, P1, and Electron microcontrollers.

SnapConnect Python如何快速上手?-图1
(图片来源网络,侵删)

What is snapconnect?

snapconnect is a Python library that implements the SPark (Particle) Cloud Protocol. This protocol allows your Python application (running on a computer, Raspberry Pi, server, etc.) to securely and reliably communicate with a Particle device over the internet.

Think of it as the "other half" of the communication pair. Your Particle device runs code that sends/receives messages using the Particle.publish() and Particle.subscribe() functions. Your Python application uses snapconnect to listen for those published messages and to send its own messages back to the device.


Key Concepts

Before we dive into code, it's crucial to understand the core concepts:

  1. The Connection: A Snap object is the main entry point. It manages the connection to the Particle Cloud. You authenticate with your Particle account credentials (username and password).

    SnapConnect Python如何快速上手?-图2
    (图片来源网络,侵删)
  2. Subscribing to Events (Listening): Your Python app can "listen" for specific events that your Particle device publishes.

    • Particle Device Side: Particle.publish("my-event", "Hello from my device!")
    • Python Side: You subscribe to "my-event". When the device publishes, your Python code gets the message.
  3. Publishing Events (Sending): Your Python app can also publish events to the Particle Cloud. This is how you can send commands or data to your device.

    • Python Side: snap.publish("my-command", "turn-on-led")
    • Particle Device Side: Your device code would be subscribed to "my-command" and would act on the message it receives.
  4. Function Calls (The Modern Way): While snapconnect focuses on events, the Particle ecosystem has a more powerful and modern way for Python to control devices: Function Calls. This is the recommended approach for most use cases today.

    • Your Python app calls a function on a specific device.
    • The device runs a pre-defined C++ function and can return a value.
    • This is more direct, reliable, and less "noisy" than publishing generic events.

Installation

First, you need to install the library. It's available on PyPI.

SnapConnect Python如何快速上手?-图3
(图片来源网络,侵删)
pip install snapconnect

Example 1: Basic Event Subscription (Python Listens to Device)

This is the classic "hello world" of snapconnect.

Particle Device Code (Firmware)

Upload this code to your Particle device (Photon, P1, or Electron).

// This code runs on your Particle device
// It publishes a message every second
void setup() {
  Serial.begin(9600);
  // Subscribe to a function we'll call from Python later
  Particle.function("python-call", handlePythonCall);
}
void loop() {
  // Publish an event named "python-test-event" every 5 seconds
  Particle.publish("python-test-event", "Hello from my Particle device!");
  Serial.println("Published event to the cloud.");
  delay(5000);
}
// This function is called from Python
int handlePythonCall(String command) {
  Serial.printlnf("Python called function with command: %s", command.c_str());
  if (command == "led-on") {
    digitalWrite(D7, HIGH);
    return 1; // Return 1 for success
  } else if (command == "led-off") {
    digitalWrite(D7, LOW);
    return 0; // Return 0 for success
  }
  return -1; // Return -1 for error (invalid command)
}

Python Application Code

Save this as a Python file (e.g., listener.py) and run it from your computer.

import snapconnect
import time
import sys
# --- Configuration ---
# Get these from your Particle account dashboard (settings -> access tokens)
PARTICLE_USERNAME = "your_email@example.com"
PARTICLE_PASSWORD = "your_access_token"
# --- Event Handler ---
# This function is called automatically when a subscribed event is received
def event_handler(name, data):
    print(f"--- Event Received ---")
    print(f"Event Name: {name}")
    print(f"Event Data: {data}")
    print("---------------------\n")
# --- Main Application ---
def main():
    print("Starting SnapConnect client...")
    # 1. Create a Snap instance
    # The first argument is the product ID. 0 is fine for personal projects.
    # The second argument is the secret, which can be an empty string for public products.
    # We will authenticate with username/password instead.
    snap = snapconnect.Snap(0, "")
    # 2. Define the events you want to subscribe to
    # You can subscribe to specific events ("python-test-event")
    # or use wildcards ("python-*" to listen to all events starting with "python-")
    subscriptions = [
        "python-test-event",
        "python-#"
    ]
    # 3. Register the event handler function
    snap.register_event_handler(event_handler, subscriptions)
    # 4. Connect to the Particle Cloud
    # This will authenticate and establish the connection
    try:
        snap.connect(PARTICLE_USERNAME, PARTICLE_PASSWORD)
        print("Successfully connected to the Particle Cloud!")
    except Exception as e:
        print(f"Connection failed: {e}")
        sys.exit(1)
    # 5. Keep the script running to listen for events
    print("Listening for events. Press Ctrl+C to exit.")
    try:
        while True:
            # The process() call is crucial. It checks for incoming messages.
            snap.process()
            time.sleep(0.1) # Small delay to prevent 100% CPU usage
    except KeyboardInterrupt:
        print("\nExiting...")
    finally:
        snap.disconnect()
        print("Disconnected.")
if __name__ == "__main__":
    main()

To run this:

  1. Upload the Particle firmware.
  2. Modify the PARTICLE_USERNAME and PARTICLE_PASSWORD in the Python script.
  3. Run python listener.py.
  4. You should see the events printed in your terminal every 5 seconds, matching the loop() delay on your device.

Example 2: Publishing Events & Making Function Calls (Python Controls Device)

Now let's make the Python app the "master" and send commands to the device.

We will use the same Particle firmware as above, since it already has a Particle.function defined.

Python Application Code (Controller)

This script will connect and then send a command to the device.

import snapconnect
import time
import sys
# --- Configuration ---
PARTICLE_USERNAME = "your_email@example.com"
PARTICLE_PASSWORD = "your_access_token"
DEVICE_ID = "your_device_id_here" # Get this from the Particle console
# --- Main Application ---
def main():
    print("Starting SnapConnect controller...")
    snap = snapconnect.Snap(0, "")
    try:
        snap.connect(PARTICLE_USERNAME, PARTICLE_PASSWORD)
        print("Successfully connected to the Particle Cloud!")
    except Exception as e:
        print(f"Connection failed: {e}")
        sys.exit(1)
    # --- Wait for the device to come online ---
    print(f"Waiting for device {DEVICE_ID} to come online...")
    while not snap.is_connected(DEVICE_ID):
        snap.process()
        time.sleep(1)
    print(f"Device {DEVICE_ID} is online.")
    # --- Method 1: Publishing an Event (Less Recommended for Control) ---
    print("\nPublishing an event to the device...")
    snap.publish("python-command", "turn-on-led")
    time.sleep(2) # Give the device time to process
    # --- Method 2: Making a Function Call (Recommended) ---
    print("\nCalling a function on the device...")
    # The call_function signature is:
    # device_id, function_name, argument_string, callback_function
    def function_callback(result):
        print(f"Function call returned: {result}")
    snap.call_function(DEVICE_ID, "python-call", "led-on", function_callback)
    # Wait for the callback to be processed
    time.sleep(1)
    print("\nCalling function again to turn LED off...")
    snap.call_function(DEVICE_ID, "python-call", "led-off", function_callback)
    print("\nCommands sent. Keeping connection open for 10 seconds...")
    time.sleep(10)
    print("Exiting...")
    snap.disconnect()
if __name__ == "__main__":
    main()

To run this:

  1. Ensure the Particle device from Example 1 is running and online.
  2. Get your device ID from the Particle Console.
  3. Update the DEVICE_ID in the Python script.
  4. Run python controller.py.
  5. You should see the messages in your Python console, and the LED (D7) on your Particle device should blink on and off!

snapconnect vs. The Modern Particle API (particle-python)

It's important to know that snapconnect is one way to do this, but the Particle team now maintains a more official and feature-rich Python library called particle-python.

Feature snapconnect particle-python (Official)
Primary Use Device-to-Cloud and Cloud-to-Device event messaging. Full control over Particle Cloud (devices, events, variables, functions, webhooks).
Authentication Username/Password or Token. Username/Password or Token.
Event Listening Excellent, built-in, real-time event subscription. Excellent, built-in, real-time event subscription.
Sending to Device snap.publish() (event) or snap.call_function() (function call). p.call_function() (cleaner, more direct).
Device Management Not its primary focus. Excellent. Can list devices, claim devices, get variables, etc.
Webhooks No support. Excellent support for creating, listing, and managing webhooks.
Recommendation Good for simple device<->cloud event systems. Recommended for all new projects. It's more powerful and actively maintained.

When should you use snapconnect?

  • If you have an existing project built on it.
  • If you specifically need a lightweight, protocol-level implementation and don't need the broader Cloud management features.

For almost any new project, you should start with the official particle-python library. However, understanding snapconnect is very valuable as it deals with the underlying SPark protocol that the official library also uses under the hood.

分享:
扫描分享到社交APP
上一篇
下一篇