heboba/goodgarden
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Kleine veranderingen mqtt

Browse Source
This commit is contained in:
Atilla
2024-03-13 09:19:15 +01:00
parent 9de5db355c
commit fee3d4a564
4 changed files with 20 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
script/devices.py
View File

@@ -6,6 +6,18 @@ def on_message(client, userdata, message):
payload_str = message.payload.decode("utf-8")
data = json.loads(payload_str)
device_256 = 0
device_322 = 0
for key in data["results"]:
if int(key["id"]) == 256:
device_256 = int(key["id"])
print(f"{device_256}")
elif int(key["id"]) == 322:
device_322 = int(key["id"])
print(f"{device_322}")
print(f"Message received on topic {message.topic}: {data}")
if __name__ == "__main__":

202
script/samenvoegen_databases.py
View File

@@ -5,24 +5,6 @@ from db_connect import database_connect
##########################* DEVICES #######################
def fetch_and_display_all(url, access_token):
try:
headers = {
"Authorization": f"Token {access_token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
print(f"Data from {url}:")
print(data)
load_data(data)
except requests.exceptions.RequestException as e:
print(f"Error fetching data from {url}: {e}")
print("Waiting for the next retrieval action...")
time.sleep(1)
def load_data(data):
mydb = database_connect()
if mydb.is_connected():
@@ -50,12 +32,6 @@ def load_data(data):
print("Data inserted into the database.")
if __name__ == "__main__":
url = "https://garden.inajar.nl/api/devices/?format=json"
access_token = "33bb3b42452306c58ecedc3c86cfae28ba22329c"
fetch_and_display_all(url, access_token)
############################### EINDE ########################
# #
# #
@@ -63,33 +39,6 @@ if __name__ == "__main__":
# #
##########################* PAR_EVENTS #######################
import requests
import time
from db_connect import database_connect
def fetch_and_display_all(url, access_token, repeat_count=5):
for _ in range(repeat_count):
try:
headers = {
"Authorization": f"Token {access_token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
print(f"Data from {url}:")
print(data)
load_data(data)
except requests.exceptions.RequestException as e:
print(f"Error fetching data from {url}: {e}")
print("Waiting for the next retrieval action...")
time.sleep(1) # Time here is in seconds.
def load_data(data):
mydb = database_connect()
if mydb.is_connected():
@@ -120,46 +69,12 @@ def load_data(data):
print("Data inserted into the database.")
if __name__ == "__main__":
url = "https://garden.inajar.nl/api/par_events/?format=json"
access_token = "33bb3b42452306c58ecedc3c86cfae28ba22329c"
# You can change the repeat_count to control how many times you want to repeat the process
repeat_count = 10
fetch_and_display_all(url, access_token, repeat_count)
############################### EINDE ########################
# #
# #
# #
# #
##########################* RELATIVE_HUMIDITY_EVENTS #######################
import requests
import time
from db_connect import database_connect
def fetch_and_display_all(url, access_token, repeat_count=5):
for _ in range(repeat_count):
try:
headers = {
"Authorization": f"Token {access_token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
print(f"Data from {url}:")
print(data)
load_data(data)
except requests.exceptions.RequestException as e:
print(f"Error fetching data from {url}: {e}")
print("Waiting for the next retrieval action...")
time.sleep(1) # Time here is in seconds.
def load_data(data):
mydb = database_connect()
@@ -191,47 +106,12 @@ def load_data(data):
print("Data inserted into the database.")
if __name__ == "__main__":
url = "https://garden.inajar.nl/api/relative_humidity_events/?format=json"
access_token = "33bb3b42452306c58ecedc3c86cfae28ba22329c"
# You can change the repeat_count to control how many times you want to repeat the process
repeat_count = 10
fetch_and_display_all(url, access_token, repeat_count)
############################### EINDE ########################
# #
# #
# #
# #
##########################* SOIL_ELECTRIC_CONDUCTIVITY_EVENTS #######################
import requests
import time
from db_connect import database_connect
def fetch_and_display_all(url, access_token, repeat_count=5):
for _ in range(repeat_count):
try:
headers = {
"Authorization": f"Token {access_token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
print(f"Data from {url}:")
print(data)
load_data(data)
except requests.exceptions.RequestException as e:
print(f"Error fetching data from {url}: {e}")
print("Waiting for the next retrieval action...")
time.sleep(1) # Time here is in seconds.
def load_data(data):
mydb = database_connect()
@@ -262,14 +142,6 @@ def load_data(data):
mydb.close()
print("Data inserted into the database.")
if __name__ == "__main__":
url = "https://garden.inajar.nl/api/soil_electric_conductivity_events/?format=json"
access_token = "33bb3b42452306c58ecedc3c86cfae28ba22329c" # Replace this with your actual access token
# You can change the repeat_count to control how many times you want to repeat the process
repeat_count = 10
fetch_and_display_all(url, access_token, repeat_count)
############################### EINDE ########################
# #
@@ -278,33 +150,6 @@ if __name__ == "__main__":
# #
##########################* SOIL_TEMPERATURE_EVENTS #######################
import requests
import time
from db_connect import database_connect
def fetch_and_display_all(url, access_token, repeat_count=5):
for _ in range(repeat_count):
try:
headers = {
"Authorization": f"Token {access_token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
print(f"Data from {url}:")
print(data)
load_data(data)
except requests.exceptions.RequestException as e:
print(f"Error fetching data from {url}: {e}")
print("Waiting for the next retrieval action...")
time.sleep(1) # Time here is in seconds.
def load_data(data):
mydb = database_connect()
if mydb.is_connected():
@@ -335,49 +180,12 @@ def load_data(data):
print("Data inserted into the database.")
if __name__ == "__main__":
url = "https://garden.inajar.nl/api/soil_relative_permittivity_events/?format=json"
access_token = "33bb3b42452306c58ecedc3c86cfae28ba22329c"
# You can change the repeat_count to control how many times you want to repeat the process
repeat_count = 10
# fetch_and_display_all(urls, access_token)
fetch_and_display_all(url, access_token, repeat_count)
############################### EINDE ########################
# #
# #
# #
# #
##########################* SOIL_TEMPERATURE_EVENTS #######################
import requests
import time
from db_connect import database_connect
def fetch_and_display_all(url, access_token, repeat_count=5):
for _ in range(repeat_count):
try:
headers = {
"Authorization": f"Token {access_token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
print(f"Data from {url}:")
print(data)
load_data(data)
except requests.exceptions.RequestException as e:
print(f"Error fetching data from {url}: {e}")
print("Waiting for the next retrieval action...")
time.sleep(1) # Time here is in seconds.
def load_data(data):
mydb = database_connect()
@@ -407,12 +215,4 @@ def load_data(data):
mycursor.close()
mydb.close()
print("Data inserted into the database.")
if __name__ == "__main__":
url = "https://garden.inajar.nl/api/soil_temperature_events/?format=json"
access_token = "33bb3b42452306c58ecedc3c86cfae28ba22329c" # Replace this with your actual access token
# You can change the repeat_count to control how many times you want to repeat the process
repeat_count = 10
fetch_and_display_all(url, access_token, repeat_count)
print("Data inserted into the database.")