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 380764d1a6
commit 4b3d6ea7cc
4 changed files with 20 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ReadMe.md
View File

@@ -1,6 +1,6 @@
# GoodGarden # GoodGarden
Welkom bij ons project genaamd "GoodGarden". Wij hebben besloten om er een semi-monolitische project van te maken. Alles is te vinden binnen deze repository, inclusief de "mqtt". Welkom bij ons project genaamd "GoodGarden". Wij hebben besloten om er een semi-monolithic project van te maken. Alles is te vinden binnen deze repository, inclusief de "mqtt" publisher.
### Vereisten ### Vereisten

9
mqtt/publisher.py
View File

@@ -28,7 +28,8 @@ def publish_to_mqtt(topic, data):
json_data = json.dumps(data) # Serialiseer de data naar een JSON-string json_data = json.dumps(data) # Serialiseer de data naar een JSON-string
client.publish(topic, json_data) client.publish(topic, json_data)
print(f"Data published to MQTT topic {topic}.") print(f"\033[92mData published to MQTT topic {topic}.\033[0m")
def fetch_and_publish_data(): def fetch_and_publish_data():
@@ -52,6 +53,8 @@ if __name__ == "__main__":
client.loop_start() # Start de niet-blokkerende loop client.loop_start() # Start de niet-blokkerende loop
while True: while True:
fetch_and_publish_data() fetch_and_publish_data()
print("Wachten, wachten en nog eens wachten...") print(f"\033[91mWachten, wachten en nog eens wachten... {publish_interval} secondes!\033[0m\n")
time.sleep(publish_interval) time.sleep(publish_interval)
client.loop_stop() client.loop_stop()

12
script/devices.py
View File

@@ -6,6 +6,18 @@ def on_message(client, userdata, message):
payload_str = message.payload.decode("utf-8") payload_str = message.payload.decode("utf-8")
data = json.loads(payload_str) 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}") print(f"Message received on topic {message.topic}: {data}")
if __name__ == "__main__": if __name__ == "__main__":

202
script/samenvoegen_databases.py
View File

@@ -5,24 +5,6 @@ from db_connect import database_connect
##########################* DEVICES ####################### ##########################* 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): def load_data(data):
mydb = database_connect() mydb = database_connect()
if mydb.is_connected(): if mydb.is_connected():
@@ -50,12 +32,6 @@ def load_data(data):
print("Data inserted into the database.") 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 ######################## ############################### EINDE ########################
# # # #
# # # #
@@ -63,33 +39,6 @@ if __name__ == "__main__":
# # # #
##########################* PAR_EVENTS ####################### ##########################* 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): def load_data(data):
mydb = database_connect() mydb = database_connect()
if mydb.is_connected(): if mydb.is_connected():
@@ -120,46 +69,12 @@ def load_data(data):
print("Data inserted into the database.") 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 ######################## ############################### EINDE ########################
# # # #
# # # #
# # # #
# # # #
##########################* RELATIVE_HUMIDITY_EVENTS ####################### ##########################* 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): def load_data(data):
mydb = database_connect() mydb = database_connect()
@@ -191,47 +106,12 @@ def load_data(data):
print("Data inserted into the database.") 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 ######################## ############################### EINDE ########################
# # # #
# # # #
# # # #
# # # #
##########################* SOIL_ELECTRIC_CONDUCTIVITY_EVENTS ####################### ##########################* 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): def load_data(data):
mydb = database_connect() mydb = database_connect()
@@ -262,14 +142,6 @@ def load_data(data):
mydb.close() mydb.close()
print("Data inserted into the database.") 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 ######################## ############################### EINDE ########################
# # # #
@@ -278,33 +150,6 @@ if __name__ == "__main__":
# # # #
##########################* SOIL_TEMPERATURE_EVENTS ####################### ##########################* 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): def load_data(data):
mydb = database_connect() mydb = database_connect()
if mydb.is_connected(): if mydb.is_connected():
@@ -335,49 +180,12 @@ def load_data(data):
print("Data inserted into the database.") 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 ######################## ############################### EINDE ########################
# # # #
# # # #
# # # #
# # # #
##########################* SOIL_TEMPERATURE_EVENTS ####################### ##########################* 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): def load_data(data):
mydb = database_connect() mydb = database_connect()
@@ -407,12 +215,4 @@ def load_data(data):
mycursor.close() mycursor.close()
mydb.close() mydb.close()
print("Data inserted into the database.") 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)