Данные о потребителях Kafka не отображаются в Dash Plotly ⇐ Python

[Anonymous]

Я пытаюсь отобразить данные о частоте сердечных сокращений и артериальном давлении в режиме реального времени на штриховом графике с использованием Python. В консоли app.py отображаются данные о потребителях, но в приложении Dash на графике ничего не отображается.
Вот мой код Producer.py

Код: Выделить всё

import random
from kafka import KafkaProducer
import json
import time

producer = KafkaProducer(bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.dumps(v).encode('utf-8'))

def produce_heart_rate_data():
patient_id = "P001"
while True:
data = {
"patient_id": patient_id,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"heart_rate": random.randint(60, 100)  # Simulated heart rate
}
producer.send('topic_health_heart', value=data)
print(f"Produced heart rate: {data}")
time.sleep(60)

produce_heart_rate_data()

вот мой Consumer.py

Код: Выделить всё

from kafka import KafkaConsumer
import json
import threading
import pandas as pd

# Global variable to store heart rate data
heart_rate_data = pd.DataFrame(columns=['patient_id', 'timestamp', 'heart_rate'])

# Create a lock for thread-safe access to the heart_rate_data DataFrame
data_lock = threading.Lock()

def consume_heart_rate_data():
global heart_rate_data
try:
consumer = KafkaConsumer(
'topic_health_heart',
bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest',
enable_auto_commit=True,
group_id='heart_rate_monitor_group',
value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)
for message in consumer:
data = message.value
print(f"Patient ID: {data['patient_id']}, Timestamp: {data['timestamp']}, Heart Rate: {data['heart_rate']}")

# Append new data to the DataFrame safely
new_data = pd.DataFrame([{
"patient_id": data['patient_id'],
"timestamp": data['timestamp'],
"heart_rate": data['heart_rate']
}])

# Acquire the lock before updating the DataFrame
with data_lock:
heart_rate_data = pd.concat([heart_rate_data, new_data], ignore_index=True)
print(f"Updated heart_rate_data size: {heart_rate_data.shape}")  # Log the size of DataFrame

except Exception as e:
print(f"Error consuming data: {e}")

# Run the Kafka consumer in a separate thread
consumer_thread = threading.Thread(target=consume_heart_rate_data)
consumer_thread.start()

а вот мой app.py

Код: Выделить всё

import dash
from dash import dcc, html
from dash.dependencies import Input, Output
import plotly.express as px
import pandas as pd
import threading
import time

from consumer1 import heart_rate_data, data_lock  # Import shared variables and lock
from datetime import datetime
import pytz

# Your local timezone (replace with appropriate timezone)
local_timezone = pytz.timezone('Asia/Karachi')  # Example for Pakistan Standard Time

app = dash.Dash(__name__)

# Delay to ensure the Kafka consumer has time to populate the data
time.sleep(15)

# App layout with a line chart that updates every 2 seconds
app.layout = html.Div([
html.H1("Real-Time Heart Rate Data"),
dcc.Graph(id='live-update-graph'),
dcc.Interval(
id='interval-component',
interval=2*1000,  # Update every 2 seconds
n_intervals=0
)
])

@app.callback(
Output('live-update-graph', 'figure'),
Input('interval-component', 'n_intervals')
)
def update_graph_live(n):
global heart_rate_data

# Safely access the heart_rate_data DataFrame using the lock
with data_lock:
df = heart_rate_data.copy()

print(f"Heart rate data in callback: {df}")  # Log DataFrame contents

if df.empty:
print("DataFrame is empty in callback")
return px.line(title="No data yet")

# Proceed if DataFrame is not empty
if 'timestamp' in df.columns:
# Convert timestamp to datetime for better plotting
df['timestamp'] = pd.to_datetime(df['timestamp'])
print(f"DataFrame after timestamp conversion:  {df}")

# Create the line graph
fig = px.line(df, x='timestamp', y='heart_rate', color='patient_id',
title="Heart Rate Over Time")

fig.update_layout(
xaxis_title="Timestamp",
yaxis_title="Heart Rate (BPM)",
xaxis=dict(showline=True, showgrid=False),
yaxis=dict(showline=True, showgrid=False),
showlegend=True
)

return fig

return px.line(title="No data yet")

if __name__ == '__main__':
app.run(debug=True, threaded=False)

Я использую библиотеку Python-kafka для создания и использования данных. Данные успешно создаются в темах Kafka и даже отображаются в консоли. Потребитель также успешно отображает данные в консоли. Но когда я запускаю app.py, данные отображаются в консоли, но не отображаются в графиках на веб-сайте. Пожалуйста, нужно решение.

Подробнее здесь: https://stackoverflow.com/questions/790 ... ash-plotly