Compare commits
No commits in common. "d0e87dc9c5b848b82b663682db6cb495bb50fcb5" and "5895cf64400e34897606c27c7ebc63fadd44de02" have entirely different histories.
d0e87dc9c5
...
5895cf6440
7 changed files with 0 additions and 457 deletions
20
.gitignore
vendored
20
.gitignore
vendored
|
|
@ -1,20 +0,0 @@
|
|||
.DS_Store
|
||||
.DS_Store?
|
||||
._*
|
||||
.Spotlight-V100
|
||||
.Trashes
|
||||
ehthumbs.db
|
||||
Thumbs.db
|
||||
**/.env
|
||||
.db
|
||||
.png
|
||||
bin/
|
||||
combined_graph.png
|
||||
device_data.db
|
||||
etc/
|
||||
include/
|
||||
lib/
|
||||
lib64/
|
||||
pyvenv.cfg
|
||||
share/
|
||||
kaluga/lib64
|
||||
|
|
@ -1,157 +0,0 @@
|
|||
import os
|
||||
import json
|
||||
import time
|
||||
import requests
|
||||
import dash
|
||||
import dash_core_components as dcc
|
||||
import dash_html_components as html
|
||||
from dash.dependencies import Input, Output
|
||||
import plotly.graph_objects as go
|
||||
from dotenv import load_dotenv
|
||||
from datetime import datetime
|
||||
import sqlite3
|
||||
|
||||
# Load environment variables for credentials
|
||||
load_dotenv()
|
||||
|
||||
# Function to get token for OwenCloud
|
||||
def get_token():
|
||||
login = os.getenv("LOGIN")
|
||||
password = os.getenv("PASSWORD")
|
||||
if not login or not password:
|
||||
raise ValueError("Логин или пароль не заданы в .env")
|
||||
|
||||
data = {'login': login, 'password': password}
|
||||
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
||||
req = requests.post("https://api.owencloud.ru/v1/auth/open/", json=data, headers=headers)
|
||||
|
||||
if req.status_code == 200:
|
||||
answer = req.json()
|
||||
token = "Bearer " + answer["token"]
|
||||
return token
|
||||
else:
|
||||
print(f"Ошибка авторизации: {req.status_code}, {req.text}")
|
||||
return None
|
||||
|
||||
# Function to fetch device data
|
||||
def fetch_device_data(device_id, token, codes_to_search):
|
||||
headers = {'Authorization': token}
|
||||
data = json.dumps({'filter': ''})
|
||||
|
||||
req = requests.post(f"https://api.owencloud.ru/v1/device/{device_id}", data=data, headers=headers)
|
||||
|
||||
if req.status_code != 200:
|
||||
print(f"Ошибка запроса устройства: {req.status_code}, {req.text}")
|
||||
return None
|
||||
|
||||
response = req.json()
|
||||
result = {}
|
||||
|
||||
# Process parameters to check if they are among the codes we need
|
||||
if 'parameters' in response and isinstance(response['parameters'], list):
|
||||
for item in response['parameters']:
|
||||
if isinstance(item, dict) and 'code' in item and item['code'] in codes_to_search:
|
||||
result[item['code']] = item['value']
|
||||
|
||||
return result
|
||||
|
||||
# SQLite Database Setup
|
||||
def init_db():
|
||||
conn = sqlite3.connect('device_data.db')
|
||||
c = conn.cursor()
|
||||
c.execute('''
|
||||
CREATE TABLE IF NOT EXISTS device_data (
|
||||
timestamp TEXT,
|
||||
code TEXT,
|
||||
value REAL
|
||||
)
|
||||
''')
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# Function to store device data in the database
|
||||
def store_data(timestamp, code, value):
|
||||
conn = sqlite3.connect('device_data.db')
|
||||
c = conn.cursor()
|
||||
c.execute('''
|
||||
INSERT INTO device_data (timestamp, code, value)
|
||||
VALUES (?, ?, ?)
|
||||
''', (timestamp, code, value))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# Initialize Dash app
|
||||
app = dash.Dash(__name__)
|
||||
|
||||
# List of codes to monitor
|
||||
codes_to_search = [
|
||||
"P16385", "P16386", "P16387", "P16388", "P16389", "P16390", "P16391",
|
||||
"P16392", "P16393", "P16394", "P16395", "P16396", "P16397", "P16398"
|
||||
]
|
||||
|
||||
device_id = "459205"
|
||||
token = get_token()
|
||||
|
||||
# Data structure to store the collected data for each device code
|
||||
collected_data = {code: [] for code in codes_to_search}
|
||||
timestamps = []
|
||||
|
||||
# Generate a Plotly figure
|
||||
def generate_figure():
|
||||
fig = go.Figure()
|
||||
for code in codes_to_search:
|
||||
fig.add_trace(go.Scatter(
|
||||
x=timestamps, y=collected_data[code], mode='lines', name=code
|
||||
))
|
||||
|
||||
fig.update_layout(
|
||||
title="Real-time Device Data Comparison",
|
||||
xaxis_title="Time",
|
||||
yaxis_title="Value",
|
||||
template="plotly_dark"
|
||||
)
|
||||
return fig
|
||||
|
||||
# Function to update data every 10 seconds and store it in the database
|
||||
def update_data():
|
||||
global timestamps
|
||||
data = fetch_device_data(device_id, token, codes_to_search)
|
||||
if data:
|
||||
# Get current timestamp
|
||||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||||
timestamps.append(timestamp)
|
||||
|
||||
# Update the collected data for each device and store it in the database
|
||||
for code in codes_to_search:
|
||||
if code in data:
|
||||
value = data[code]
|
||||
collected_data[code].append(value)
|
||||
store_data(timestamp, code, value)
|
||||
|
||||
# Dash layout
|
||||
app.layout = html.Div([
|
||||
html.H1("Real-time Device Data Comparison"),
|
||||
dcc.Graph(id='live-update-graph'),
|
||||
dcc.Interval(
|
||||
id='interval-component',
|
||||
interval=10000, # Update every 10 seconds
|
||||
n_intervals=0
|
||||
)
|
||||
])
|
||||
|
||||
# Callback to update the graph
|
||||
@app.callback(
|
||||
Output('live-update-graph', 'figure'),
|
||||
Input('interval-component', 'n_intervals')
|
||||
)
|
||||
def update_graph(n_intervals):
|
||||
# Update data
|
||||
update_data()
|
||||
|
||||
# Generate the updated figure
|
||||
return generate_figure()
|
||||
|
||||
# Run the Dash app
|
||||
if __name__ == '__main__':
|
||||
init_db() # Initialize the database
|
||||
app.run_server(debug=True)
|
||||
|
|
@ -1,89 +0,0 @@
|
|||
import requests
|
||||
import json
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load information for owencloud and telegram
|
||||
load_dotenv()
|
||||
#Send To Telegram
|
||||
def send_telegram(text: str):
|
||||
token = os.getenv("TELEGRAM_TOKEN")
|
||||
channel_id = os.getenv("TELEGRAM_CHANNEL_ID")
|
||||
if not token or not channel_id:
|
||||
raise ValueError("Telegram токен или ID канала не заданы в .env")
|
||||
|
||||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||||
|
||||
r = requests.post(url, data={
|
||||
"chat_id": channel_id,
|
||||
"text": text
|
||||
})
|
||||
print(r.text)
|
||||
if r.status_code != 200:
|
||||
raise Exception(f"Ошибка отправки в Telegram: {r.text}")
|
||||
#Get Token for OwenCloud
|
||||
def get_token():
|
||||
login = os.getenv("LOGIN")
|
||||
password = os.getenv("PASSWORD")
|
||||
if not login or not password:
|
||||
raise ValueError("Логин или пароль не заданы в .env")
|
||||
|
||||
data = {'login': login, 'password': password}
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json"
|
||||
}
|
||||
req = requests.post("https://api.owencloud.ru/v1/auth/open/", json=data, headers=headers)
|
||||
if req.status_code == 200:
|
||||
answer = req.json()
|
||||
token = "Bearer " + answer["token"]
|
||||
return token
|
||||
else:
|
||||
print(f"Ошибка авторизации: {req.status_code}, {req.text}")
|
||||
return None
|
||||
#Device parsing
|
||||
def main():
|
||||
token = get_token()
|
||||
if not token:
|
||||
print("Не удалось получить токен. Завершаем работу.")
|
||||
return
|
||||
|
||||
device_id = "296613"
|
||||
headers = {'Authorization': token}
|
||||
data = json.dumps({'filter': ''})
|
||||
|
||||
req = requests.post(f"https://api.owencloud.ru/v1/device/{device_id}", data=data, headers=headers)
|
||||
if req.status_code != 200:
|
||||
print(f"Ошибка запроса устройства: {req.status_code}, {req.text}")
|
||||
return
|
||||
|
||||
nswer = req.json()
|
||||
|
||||
if "parameters" not in nswer:
|
||||
print("Не удалось найти параметры в ответе.")
|
||||
return
|
||||
|
||||
parameters = nswer['parameters']
|
||||
|
||||
#Find temperature
|
||||
try:
|
||||
temperature_1 = next(param for param in parameters if param['name'] == "Температура_1 (НИЗ)")
|
||||
temperature_2 = next(param for param in parameters if param['name'] == "Температура_2 (ВЕРХ)")
|
||||
except StopIteration:
|
||||
print("Не удалось найти необходимые параметры.")
|
||||
return
|
||||
|
||||
message = (f"Шача:\n"
|
||||
f"Температура_1 (НИЗ): {temperature_1['value']}\n"
|
||||
f"Температура_2 (ВЕРХ): {temperature_2['value']}")
|
||||
|
||||
print(message)
|
||||
|
||||
# Send to Telegram
|
||||
try:
|
||||
send_telegram(message)
|
||||
except Exception as e:
|
||||
print(f"Ошибка отправки сообщения: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
import requests
|
||||
import json
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load information for owencloud and telegram
|
||||
load_dotenv()
|
||||
#Get Token for OwenCloud
|
||||
def get_token():
|
||||
login = os.getenv("LOGIN")
|
||||
password = os.getenv("PASSWORD")
|
||||
if not login or not password:
|
||||
raise ValueError("Логин или пароль не заданы в .env")
|
||||
|
||||
data = {'login': login, 'password': password}
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json"
|
||||
}
|
||||
req = requests.post("https://api.owencloud.ru/v1/auth/open/", json=data, headers=headers)
|
||||
if req.status_code == 200:
|
||||
answer = req.json()
|
||||
token = "Bearer " + answer["token"]
|
||||
return token
|
||||
else:
|
||||
print(f"Ошибка авторизации: {req.status_code}, {req.text}")
|
||||
return None
|
||||
#Device parsing
|
||||
def main():
|
||||
token = get_token()
|
||||
if not token:
|
||||
print("Не удалось получить токен. Завершаем работу.")
|
||||
return
|
||||
|
||||
device_id = "296613"
|
||||
headers = {'Authorization': token}
|
||||
data = json.dumps({'filter': ''})
|
||||
|
||||
req = requests.post("https://api.owencloud.ru/v1/device/index/", data=data, headers=headers)
|
||||
|
||||
if req.status_code != 200:
|
||||
print(f"Ошибка запроса устройства: {req.status_code}, {req.text}")
|
||||
return
|
||||
|
||||
nswer = req.json()
|
||||
|
||||
|
||||
print(req.text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,65 +0,0 @@
|
|||
import requests
|
||||
import json
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load information for owencloud and telegram
|
||||
load_dotenv()
|
||||
|
||||
#Get Token for OwenCloud
|
||||
def get_token():
|
||||
login = os.getenv("LOGIN")
|
||||
password = os.getenv("PASSWORD")
|
||||
if not login or not password:
|
||||
raise ValueError("Логин или пароль не заданы в .env")
|
||||
|
||||
data = {'login': login, 'password': password}
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json"
|
||||
}
|
||||
req = requests.post("https://api.owencloud.ru/v1/auth/open/", json=data, headers=headers)
|
||||
if req.status_code == 200:
|
||||
answer = req.json()
|
||||
token = "Bearer " + answer["token"]
|
||||
return token
|
||||
else:
|
||||
print(f"Ошибка авторизации: {req.status_code}, {req.text}")
|
||||
return None
|
||||
#Device parsing
|
||||
def main():
|
||||
|
||||
codes_to_search = [
|
||||
"P16385", "P16386", "P16387", "P16388", "P16389", "P16390", "P16391",
|
||||
"P16392", "P16393", "P16394", "P16395", "P16396", "P16397", "P16398"
|
||||
]
|
||||
token = get_token()
|
||||
if not token:
|
||||
print("Не удалось получить токен. Завершаем работу.")
|
||||
return
|
||||
|
||||
device_id = "459205"
|
||||
headers = {'Authorization': token}
|
||||
data = json.dumps({'filter': ''})
|
||||
|
||||
req = requests.post(f"https://api.owencloud.ru/v1/device/{device_id}", data=data, headers=headers)
|
||||
if req.status_code != 200:
|
||||
print(f"Ошибка запроса устройства: {req.status_code}, {req.text}")
|
||||
return
|
||||
|
||||
#nswer = req.json()
|
||||
#print(req.text)
|
||||
#print("Response structure:", json.dumps(nswer, indent=2))
|
||||
nwer = req.json() # Store the response in 'nwer'
|
||||
|
||||
# Check and process the parameters list
|
||||
if 'parameters' in nwer and isinstance(nwer['parameters'], list):
|
||||
for item in nwer['parameters']:
|
||||
if isinstance(item, dict) and 'code' in item and item['code'] in codes_to_search:
|
||||
print(f"Code: {item['code']}, Value: {item['value']}")
|
||||
else:
|
||||
print("Unexpected response format or 'parameters' key is missing.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,74 +0,0 @@
|
|||
import requests
|
||||
import json
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load information for owencloud and telegram
|
||||
load_dotenv()
|
||||
|
||||
#Get Token for OwenCloud
|
||||
def get_token():
|
||||
login = os.getenv("LOGIN")
|
||||
password = os.getenv("PASSWORD")
|
||||
if not login or not password:
|
||||
raise ValueError("Логин или пароль не заданы в .env")
|
||||
|
||||
data = {'login': login, 'password': password}
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json"
|
||||
}
|
||||
req = requests.post("https://api.owencloud.ru/v1/auth/open/", json=data, headers=headers)
|
||||
if req.status_code == 200:
|
||||
answer = req.json()
|
||||
token = "Bearer " + answer["token"]
|
||||
return token
|
||||
else:
|
||||
print(f"Ошибка авторизации: {req.status_code}, {req.text}")
|
||||
return None
|
||||
#Device parsing
|
||||
def main():
|
||||
token = get_token()
|
||||
if not token:
|
||||
print("Не удалось получить токен. Завершаем работу.")
|
||||
return
|
||||
|
||||
device_id = "296613"
|
||||
headers = {'Authorization': token}
|
||||
data = json.dumps({'filter': ''})
|
||||
|
||||
req = requests.post(f"https://api.owencloud.ru/v1/device/{device_id}", data=data, headers=headers)
|
||||
if req.status_code != 200:
|
||||
print(f"Ошибка запроса устройства: {req.status_code}, {req.text}")
|
||||
return
|
||||
|
||||
nswer = req.json()
|
||||
|
||||
if "parameters" not in nswer:
|
||||
print("Не удалось найти параметры в ответе.")
|
||||
return
|
||||
|
||||
parameters = nswer['parameters']
|
||||
|
||||
#Find temperature
|
||||
try:
|
||||
temperature_1 = next(param for param in parameters if param['name'] == "Температура_1 (НИЗ)")
|
||||
temperature_2 = next(param for param in parameters if param['name'] == "Температура_2 (ВЕРХ)")
|
||||
except StopIteration:
|
||||
print("Не удалось найти необходимые параметры.")
|
||||
return
|
||||
|
||||
message = (f"Шача:\n"
|
||||
f"Температура_1 (НИЗ): {temperature_1['value']}\n"
|
||||
f"Температура_2 (ВЕРХ): {temperature_2['value']}")
|
||||
|
||||
print(message)
|
||||
|
||||
# Send to Telegram
|
||||
try:
|
||||
send_telegram(message)
|
||||
except Exception as e:
|
||||
print(f"Ошибка отправки сообщения: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Reference in a new issue