from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, Image from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib import colors from reportlab.lib.units import inch from database import get_farm_details_from_db, get_hives_from_db, get_hive_detail_from_db, get_history, get_user_profile from datetime import datetime import io import os import matplotlib matplotlib.use('Agg') # Non-GUI backend for server environments import matplotlib.pyplot as plt import tempfile import uuid import logging # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def create_table(data, col_widths=None): """Helper function to create a styled table.""" table = Table(data, colWidths=col_widths) table.setStyle(TableStyle([ ('GRID', (0, 0), (-1, -1), 1, colors.black), ('FONT', (0, 0), (-1, -1), 'Helvetica', 10), ('BACKGROUND', (0, 0), (-1, 0), colors.lightgrey), ('ALIGN', (0, 0), (-1, -1), 'CENTER'), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('INNERGRID', (0, 0), (-1, -1), 0.25, colors.black), ('BOX', (0, 0), (-1, -1), 0.25, colors.black), ('LEFTPADDING', (0, 0), (-1, -1), 6), ('RIGHTPADDING', (0, 0), (-1, -1), 6), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6), ])) return table def create_pie_chart(data_counts, output_path, title): """Generate a pie chart for given data distribution.""" try: if not data_counts: logger.warning(f"No data for {title} pie chart, skipping generation") return False labels = list(data_counts.keys()) sizes = list(data_counts.values()) colors_list = ['#ff9999', '#66b3ff', '#99ff99', '#ffcc99', '#ffb3e6', '#c2c2f0'] plt.figure(figsize=(4, 4)) plt.pie(sizes, labels=labels, colors=colors_list[:len(labels)], autopct='%1.1f%%', startangle=90) plt.axis('equal') # Ensure circular shape plt.title(title) plt.savefig(output_path, bbox_inches='tight', dpi=150, format='png') plt.close() if os.path.exists(output_path): return True else: logger.error(f"Pie chart file not found at {output_path} after saving") return False except Exception as e: logger.error(f"Error generating {title} pie chart: {str(e)}") return False def create_bar_chart(history, output_path): """Generate a bar chart for prediction results.""" try: result_counts = {'healthy': 0, 'no queen': 0, 'not bee': 0} for entry in history: result = entry['result'].lower() if result in result_counts: result_counts[result] += 1 labels = list(result_counts.keys()) counts = list(result_counts.values()) plt.figure(figsize=(6, 4)) plt.bar(labels, counts, color=['#66b3ff', '#ff9999', '#99ff99']) plt.xlabel('Prediction Result') plt.ylabel('Count') plt.title('Prediction Result Distribution') for i, v in enumerate(counts): plt.text(i, v + 0.1, str(v), ha='center') plt.savefig(output_path, bbox_inches='tight', dpi=150, format='png') plt.close() if os.path.exists(output_path): return True else: logger.error(f"Bar chart file not found at {output_path} after saving") return False except Exception as e: logger.error(f"Error generating bar chart: {str(e)}") return False def add_footer(canvas, doc): """Add a footer with page number to each page.""" canvas.saveState() canvas.setFont('Helvetica', 9) page_number = f"Page {doc.page}" canvas.drawCentredString(letter[0] / 2, 0.5 * inch, page_number) canvas.restoreState() def generate_report(user_id): """ Generates a PDF report for a user's bee hives with visualizations and returns a BytesIO buffer. Args: user_id (str): The ID of the user for whom the report is generated. Returns: io.BytesIO: A buffer containing the generated PDF report. Raises: Exception: If farm details are not found or an error occurs during report generation. """ logger.info(f"Generating report for user_id: {user_id}") # Create a unique temporary directory for this report temp_dir = os.path.join(tempfile.gettempdir(), f"report_{uuid.uuid4()}") os.makedirs(temp_dir, exist_ok=True) try: # Get user details user = get_user_profile(user_id) if 'error' in user: logger.error(f"User not found for user_id: {user_id}") user_details = ["User details not found."] else: user_details = [ f"Name: {user.get('fullname', 'N/A')}", f"Email: {user.get('email', 'N/A')}", f"Location: {user.get('city', 'N/A')}, {user.get('country', 'N/A')}", f"Gender: {user.get('gender', 'N/A')}", f"Phone: {user.get('phone_number', 'N/A')}" ] # Get farm details farm = get_farm_details_from_db(user_id) if not farm: logger.error(f"No farm details found for user_id: {user_id}") raise Exception("Farm details not found") # Get hives hives = get_hives_from_db(farm['farm_id']) hive_details = [] health_status_counts = {} bee_type_counts = {} for hive in hives: hive_detail = get_hive_detail_from_db(hive['hive_id']) if 'error' not in hive_detail: hive_details.append(hive_detail) health_status = hive_detail.get('health_status', 'Unknown') health_status_counts[health_status] = health_status_counts.get(health_status, 0) + 1 bee_type = hive_detail.get('bee_type', 'Unknown') bee_type_counts[bee_type] = bee_type_counts.get(bee_type, 0) + 1 # Get prediction history history = get_history(user_id) # Generate recommendations recommendations = [] if health_status_counts.get('Unhealthy', 0) > 0: recommendations.append("Inspect hives with 'Unhealthy' status immediately and consult a beekeeping expert.") if health_status_counts.get('Unknown', 0) > 0: recommendations.append("Update health status for hives marked as 'Unknown' to ensure accurate monitoring.") no_queen_count = sum(1 for entry in history if entry['result'].lower() == 'no queen') if no_queen_count > len(history) * 0.3: # More than 30% no queen results recommendations.append("Multiple hives lack a queen. Consider introducing new queens or requeening.") if len(history) > 0 and len([entry for entry in history if entry['result'].lower() == 'not bee']) > len(history) * 0.5: recommendations.append("High number of 'not bee' predictions. Verify audio recordings and hive activity.") # Create PDF buffer = io.BytesIO() doc = SimpleDocTemplate(buffer, pagesize=letter, topMargin=0.5*inch, bottomMargin=0.5*inch) styles = getSampleStyleSheet() # Custom styles styles.add(ParagraphStyle(name='CenteredTitle', parent=styles['Title'], alignment=1)) styles.add(ParagraphStyle(name='BoldNormal', parent=styles['Normal'], fontName='Helvetica-Bold')) elements = [] # Title elements.append(Paragraph("Bee Hive Monitoring Report", styles['CenteredTitle'])) elements.append(Paragraph(f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", styles['Normal'])) elements.append(Spacer(1, 12)) # User Details elements.append(Paragraph("User Details", styles['Heading2'])) for detail in user_details: elements.append(Paragraph(detail, styles['Normal'])) elements.append(Spacer(1, 12)) # Farm Details try: farm_data = [ ["Farm ID", farm.get('farm_id', 'N/A')], ["Name", farm.get('fullname', 'N/A')], ["Location", f"{farm.get('city', 'N/A')}, {farm.get('country', 'N/A')} {farm.get('zip', 'N/A')}"] ] farm_table = create_table(farm_data, col_widths=[2*inch, 4*inch]) elements.append(farm_table) except Exception as e: logger.error(f"Error creating farm details table: {e}") elements.append(Paragraph("Error: Unable to display farm details.", styles['Normal'])) elements.append(Spacer(1, 12)) # Hive Summary elements.append(Paragraph("Hive Summary", styles['Heading2'])) hive_data = [["Hive #", "Bee Type", "Frames", "Health", "Created"]] for hive in hive_details: hive_data.append([ hive.get('hive_number', 'N/A'), hive.get('bee_type', 'N/A'), hive.get('number_of_frames', 'N/A'), hive.get('health_status', 'Unknown'), hive.get('creation_date', datetime.now()).strftime('%Y-%m-%d') ]) hive_table = create_table(hive_data, col_widths=[1.2*inch, 1.8*inch, 1.2*inch, 1.2*inch, 1.6*inch]) elements.append(hive_table) elements.append(Spacer(1, 12)) # Bee Type Distribution with Pie Chart elements.append(Paragraph("Bee Type Distribution", styles['Heading2'])) bee_type_table = create_table([["Bee Type", "Count"]] + [[bt, count] for bt, count in bee_type_counts.items()], col_widths=[3*inch, 1*inch]) elements.append(bee_type_table) bee_type_chart_path = os.path.join(temp_dir, f"bee_type_pie_{user_id}.png") if bee_type_counts: if create_pie_chart(bee_type_counts, bee_type_chart_path, "Bee Type Distribution"): if os.path.exists(bee_type_chart_path): elements.append(Spacer(1, 12)) elements.append(Image(bee_type_chart_path, width=3*inch, height=3*inch, kind='proportional')) else: elements.append(Paragraph("Error: Unable to display bee type pie chart.", styles['Normal'])) else: elements.append(Paragraph("No bee type data available for pie chart.", styles['Normal'])) else: elements.append(Paragraph("No bee type data available for pie chart.", styles['Normal'])) elements.append(Spacer(1, 12)) # Health Status Overview with Pie Chart elements.append(Paragraph("Health Status Overview", styles['Heading2'])) health_data = [[status, count] for status, count in health_status_counts.items()] health_table = create_table([["Status", "Count"]] + health_data, col_widths=[3*inch, 1*inch]) elements.append(health_table) health_chart_path = os.path.join(temp_dir, f"health_pie_{user_id}.png") if health_status_counts: if create_pie_chart(health_status_counts, health_chart_path, "Health Status Distribution"): if os.path.exists(health_chart_path): elements.append(Spacer(1, 12)) elements.append(Image(health_chart_path, width=3*inch, height=3*inch, kind='proportional')) else: elements.append(Paragraph("Error: Unable to display health status pie chart.", styles['Normal'])) else: elements.append(Paragraph("No health status data available for pie chart.", styles['Normal'])) else: elements.append(Paragraph("No health status data available for pie chart.", styles['Normal'])) elements.append(Spacer(1, 12)) # Prediction History with Bar Chart elements.append(Paragraph("Prediction History", styles['Heading2'])) history_data = [["Timestamp", "Audio", "Result", "Hive #"]] for entry in history[:10]: history_data.append([ entry.get('timestamp', 'N/A'), entry.get('audio_name', 'N/A'), entry.get('result', 'N/A'), entry.get('hive_number', 'N/A') or "N/A" ]) history_table = create_table(history_data, col_widths=[1.5*inch, 2*inch, 1*inch, 1*inch]) elements.append(history_table) bar_chart_path = os.path.join(temp_dir, f"prediction_bar_{user_id}.png") if history: if create_bar_chart(history, bar_chart_path): if os.path.exists(bar_chart_path): elements.append(Spacer(1, 12)) elements.append(Image(bar_chart_path, width=4*inch, height=2.5*inch)) else: elements.append(Paragraph("Error: Unable to display prediction history bar chart.", styles['Normal'])) else: elements.append(Paragraph("No prediction data available for bar chart.", styles['Normal'])) else: elements.append(Paragraph("No prediction data available for bar chart.", styles['Normal'])) elements.append(Spacer(1, 12)) # Recommendations elements.append(Paragraph("Recommendations", styles['Heading2'])) for rec in recommendations: elements.append(Paragraph(f"• {rec}", styles['BoldNormal'])) elements.append(Spacer(1, 12)) # Build PDF with footer try: doc.build(elements, onFirstPage=add_footer, onLaterPages=add_footer) except Exception as e: logger.error(f"Error building PDF: {e}") raise Exception(f"Failed to generate PDF: {e}") finally: # Clean up temporary directory if os.path.exists(temp_dir): for file in os.listdir(temp_dir): try: os.remove(os.path.join(temp_dir, file)) except Exception as e: logger.error(f"Error cleaning up file {file}: {e}") try: os.rmdir(temp_dir) except Exception as e: logger.error(f"Error cleaning up directory {temp_dir}: {e}") buffer.seek(0) logger.info(f"Report generated successfully for user_id: {user_id}") return buffer