add parser and kaluga

.gitignore

@@ -0,0 +1,20 @@
+.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

kaluga/gen_graph.py

@@ -0,0 +1,157 @@
+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)

kaluga/main.py

@@ -0,0 +1,89 @@
+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()

parser/device_index.py

@@ -0,0 +1,52 @@
+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()

parser/device_parser.py

@@ -0,0 +1,65 @@
+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()

parser/main.py

@@ -0,0 +1,74 @@
+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()