q/owen-script
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: q:d0e87dc9c5b848b82b663682db6cb495bb50fcb5
Branches Tags
q:main
..
compare: q:5895cf64400e34897606c27c7ebc63fadd44de02
Branches Tags
q:main

No commits in common. "d0e87dc9c5b848b82b663682db6cb495bb50fcb5" and "5895cf64400e34897606c27c7ebc63fadd44de02" have entirely different histories.
d0e87dc9c5 ... 5895cf6440

7 changed files with 0 additions and 457 deletions
Show stats Expand all files Collapse all files

20
.gitignore vendored
View file

@ -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

157
kaluga/gen_graph.py
View file

@ -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)

89
kaluga/main.py
View file

@ -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()

52
parser/device_index.py
View file

@ -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()

65
parser/device_parser.py
View file

@ -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()

74
parser/main.py
View file

@ -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()

0
parser/test
View file