Spaces:
Sleeping
Sleeping
File size: 3,323 Bytes
285a08f f5dadbf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
import requests
import firebase_admin
from firebase_admin import credentials, firestore
import time
# Replace with your actual credentials
THINGSPEAK_API_KEY = "P54KXM40TA3CB6W4"
CHANNEL_ID = "2784385"
FIREBASE_CREDENTIALS_PATH = r"snippetscript-37175-firebase-adminsdk-cf1z8-7d509b09fd.json"
# Global variable to track the last entry ID
last_entry_id = None
def initialize_firebase():
"""Initialize Firebase connection"""
try:
# Initialize Firebase app (only once)
cred = credentials.Certificate(FIREBASE_CREDENTIALS_PATH)
firebase_admin.initialize_app(cred)
# Create Firestore client
return firestore.client()
except Exception as e:
print(f"Firebase initialization error: {e}")
return None
def fetch_thingspeak_data():
"""Fetch data from ThingSpeak"""
global last_entry_id
try:
url = f"https://api.thingspeak.com/channels/2784385/feeds.json?api_key=P54KXM40TA3CB6W4"
response = requests.get(url)
response.raise_for_status() # Raise an exception for bad responses
data = response.json()
# Check if there are feeds
if not data.get('feeds'):
print("No new feeds found.")
return None
# Get the latest feed
latest_feed = data['feeds'][-1]
# Check if this is a new entry
current_entry_id = latest_feed.get('entry_id')
if current_entry_id != last_entry_id:
last_entry_id = current_entry_id
return latest_feed
print("No new entries since last check.")
return None
except requests.RequestException as e:
print(f"ThingSpeak data fetch error: {e}")
return None
def store_data_in_firestore(firestore_client, data):
"""Store data in Firestore"""
try:
# Validate data
if not data:
print("No data to store.")
return
# Create a document with timestamp
doc_data = {
'field1': data.get('field1'),
'field2': data.get('field2'),
'entry_id': data.get('entry_id'),
'timestamp': firestore.SERVER_TIMESTAMP
}
# Add to Firestore collection
firestore_client.collection('thingspeak_data').add(doc_data)
print("New data successfully stored in Firestore")
except Exception as e:
print(f"Firestore storage error: {e}")
def main():
# Initialize Firebase
firestore_client = initialize_firebase()
if not firestore_client:
print("Failed to initialize Firestore. Exiting.")
return
# Main data collection loop
try:
while True:
# Fetch data from ThingSpeak
thingspeak_data = fetch_thingspeak_data()
if thingspeak_data:
# Store data in Firestore only if there's a new entry
store_data_in_firestore(firestore_client, thingspeak_data)
# Wait before next iteration
time.sleep(5) # Adjust interval as needed
except KeyboardInterrupt:
print("Data collection stopped by user.")
except Exception as e:
print(f"Unexpected error: {e}")
if __name__ == "__main__":
main() |