""" BI服务模块 提供自助BI看板和数据可视化服务 包括数据集管理、图表创建、看板配置等功能 """ from typing import Dict, List, Optional, Any, Tuple from datetime import datetime, timedelta import json import pandas as pd import numpy as np from .models import ( Chart, ChartType, DataSourceType, Dashboard, DataSource, Dataset, SupersetIntegration, MetabaseIntegration ) class BIService: """BI服务主类""" def __init__(self): # 初始化数据存储 self.charts: Dict[str, Chart] = {} self.dashboards: Dict[str, Dashboard] = {} self.data_sources: Dict[str, DataSource] = {} self.datasets: Dict[str, Dataset] = {} self.superset_integration: Optional[SupersetIntegration] = None self.metabase_integration: Optional[MetabaseIntegration] = None # 初始化默认数据源 self._init_default_data_sources() self._init_default_charts() self._init_default_dashboards() def _init_default_data_sources(self): """初始化默认数据源""" # 传感器数据源 sensor_source = DataSource( id="sensor_data", name="传感器数据", type=DataSourceType.SENSOR_DATA, description="所有IoT传感器实时和历史数据", config={ "time_field": "timestamp", "value_field": "value", "device_field": "device_id", "location_field": "location", "type_field": "data_type" }, query_template="SELECT * FROM sensor_data WHERE {filters}", columns=[ {"name": "id", "type": "integer", "description": "数据记录ID"}, {"name": "device_id", "type": "string", "description": "设备ID"}, {"name": "data_type", "type": "string", "description": "数据类型"}, {"name": "value", "type": "float", "description": "数值"}, {"name": "unit", "type": "string", "description": "单位"}, {"name": "timestamp", "type": "datetime", "description": "时间戳"}, {"name": "location", "type": "string", "description": "位置"}, {"name": "quality_score", "type": "float", "description": "质量评分"} ] ) self.data_sources[sensor_source.id] = sensor_source # 设备状态数据源 device_source = DataSource( id="device_data", name="设备状态", type=DataSourceType.DEVICE_DATA, description="所有设备的运行状态和配置信息", config={ "status_field": "status", "type_field": "device_type", "location_field": "location" }, query_template="SELECT * FROM device WHERE {filters}", columns=[ {"name": "id", "type": "string", "description": "设备ID"}, {"name": "name", "type": "string", "description": "设备名称"}, {"name": "device_type", "type": "string", "description": "设备类型"}, {"name": "location", "type": "string", "description": "位置"}, {"name": "status", "type": "string", "description": "状态"}, {"name": "install_date", "type": "datetime", "description": "安装日期"}, {"name": "metadata", "type": "json", "description": "元数据"} ] ) self.data_sources[device_source.id] = device_source # 警报数据源 alert_source = DataSource( id="alert_data", name="警报数据", type=DataSourceType.ALERT_DATA, description="系统警报和通知记录", config={ "level_field": "level", "type_field": "alert_type", "resolved_field": "resolved" }, query_template="SELECT * FROM alert WHERE {filters}", columns=[ {"name": "id", "type": "string", "description": "警报ID"}, {"name": "device_id", "type": "string", "description": "设备ID"}, {"name": "alert_type", "type": "string", "description": "警报类型"}, {"name": "level", "type": "string", "description": "警报级别"}, {"name": "message", "type": "string", "description": "警报信息"}, {"name": "timestamp", "type": "datetime", "description": "发生时间"}, {"name": "resolved", "type": "boolean", "description": "是否已解决"}, {"name": "resolved_at", "type": "datetime", "description": "解决时间"} ] ) self.data_sources[alert_source.id] = alert_source # 系统统计数据源 stats_source = DataSource( id="system_stats", name="系统统计", type=DataSourceType.SYSTEM_STATS, description="系统运行性能和使用统计", config={ "cpu_field": "cpu_usage_percent", "memory_field": "memory_usage_mb", "records_field": "total_records" }, query_template="SELECT * FROM system_stats WHERE {filters}", columns=[ {"name": "timestamp", "type": "datetime", "description": "统计时间"}, {"name": "total_records", "type": "integer", "description": "总记录数"}, {"name": "total_devices", "type": "integer", "description": "设备总数"}, {"name": "active_connections", "type": "integer", "description": "活跃连接数"}, {"name": "api_requests_count", "type": "integer", "description": "API请求数"}, {"name": "alerts_count", "type": "integer", "description": "警报数量"}, {"name": "data_quality_score", "type": "float", "description": "数据质量评分"}, {"name": "memory_usage_mb", "type": "float", "description": "内存使用量(MB)"}, {"name": "cpu_usage_percent", "type": "float", "description": "CPU使用率(%)"} ] ) self.data_sources[stats_source.id] = stats_source def _init_default_charts(self): """初始化默认图表""" # 流量趋势图 flow_trend = Chart( id="flow_trend", name="流量趋势分析", description="显示各区域流量随时间的变化趋势", chart_type=ChartType.LINE, data_source=DataSourceType.SENSOR_DATA, x_axis="timestamp", y_axis=["value"], group_by=["location"], aggregation="avg", filters={"data_type": "LL"}, options={ "title": "各区域流量趋势", "yAxis": {"title": "流量 (m³/h)"}, "xAxis": {"title": "时间"}, "legend": {"show": True}, "tooltip": {"trigger": "axis"} }, tags=["流量", "趋势", "区域"] ) self.charts[flow_trend.id] = flow_trend # 设备状态分布图 device_status = Chart( id="device_status_distribution", name="设备状态分布", description="显示不同状态设备的数量分布", chart_type=ChartType.PIE, data_source=DataSourceType.DEVICE_DATA, x_axis="status", y_axis=["count"], aggregation="count", options={ "title": "设备状态分布", "legend": {"show": True}, "tooltip": {"trigger": "item"} }, tags=["设备", "状态", "分布"] ) self.charts[device_status.id] = device_status # 警报级别统计图 alert_stats = Chart( id="alert_level_stats", name="警报级别统计", description="按级别统计警报数量", chart_type=ChartType.BAR, data_source=DataSourceType.ALERT_DATA, x_axis="level", y_axis=["count"], aggregation="count", filters={"resolved": False}, options={ "title": "未解决警报按级别统计", "yAxis": {"title": "数量"}, "xAxis": {"title": "警报级别"}, "legend": {"show": False} }, tags=["警报", "级别", "统计"] ) self.charts[alert_stats.id] = alert_stats # 系统性能监控图 system_performance = Chart( id="system_performance", name="系统性能监控", description="显示系统CPU和内存使用率趋势", chart_type=ChartType.LINE, data_source=DataSourceType.SYSTEM_STATS, x_axis="timestamp", y_axis=["cpu_usage_percent", "memory_usage_mb"], options={ "title": "系统性能监控", "yAxis": [{"title": "CPU使用率(%)"}, {"title": "内存使用量(MB)"}], "xAxis": {"title": "时间"}, "legend": {"show": True}, "tooltip": {"trigger": "axis"} }, tags=["系统", "性能", "监控"] ) self.charts[system_performance.id] = system_performance def _init_default_dashboards(self): """初始化默认看板""" # 水务运营总览看板 overview_dashboard = Dashboard( id="operation_overview", name="水务运营总览", description="水务系统整体运营情况综合看板", charts=["flow_trend", "device_status_distribution", "alert_level_stats", "system_performance"], layout=[ {"i": "flow_trend", "x": 0, "y": 0, "w": 12, "h": 8}, {"i": "device_status_distribution", "x": 12, "y": 0, "w": 6, "h": 6}, {"i": "alert_level_stats", "x": 18, "y": 0, "w": 6, "h": 6}, {"i": "system_performance", "x": 0, "y": 8, "w": 24, "h": 8} ], is_public=True, tags=["运营", "总览", "综合"] ) self.dashboards[overview_dashboard.id] = overview_dashboard # 设备管理看板 device_dashboard = Dashboard( id="device_management", name="设备管理看板", description="设备状态监控和维护管理", charts=["device_status_distribution"], layout=[ {"i": "device_status_distribution", "x": 0, "y": 0, "w": 12, "h": 8} ], tags=["设备", "管理", "监控"] ) self.dashboards[device_dashboard.id] = device_dashboard # 安全监控看板 security_dashboard = Dashboard( id="security_monitoring", name="安全监控看板", description="系统安全和警报监控", charts=["alert_level_stats"], layout=[ {"i": "alert_level_stats", "x": 0, "y": 0, "w": 12, "h": 8} ], tags=["安全", "监控", "警报"] ) self.dashboards[security_dashboard.id] = security_dashboard def get_chart(self, chart_id: str) -> Optional[Chart]: """获取图表""" return self.charts.get(chart_id) def get_all_charts(self) -> List[Chart]: """获取所有图表""" return list(self.charts.values()) def create_chart(self, chart_data: Dict[str, Any]) -> Chart: """创建图表""" chart = Chart.from_dict(chart_data) self.charts[chart.id] = chart return chart def update_chart(self, chart_id: str, chart_data: Dict[str, Any]) -> Optional[Chart]: """更新图表""" if chart_id in self.charts: chart = Chart.from_dict(chart_data) chart.id = chart_id # 保持ID不变 self.charts[chart_id] = chart return chart return None def delete_chart(self, chart_id: str) -> bool: """删除图表""" if chart_id in self.charts: del self.charts[chart_id] # 从所有看板中移除该图表 for dashboard in self.dashboards.values(): if chart_id in dashboard.charts: dashboard.charts.remove(chart_id) return True return False def get_dashboard(self, dashboard_id: str) -> Optional[Dashboard]: """获取看板""" return self.dashboards.get(dashboard_id) def get_all_dashboards(self) -> List[Dashboard]: """获取所有看板""" return list(self.dashboards.values()) def create_dashboard(self, dashboard_data: Dict[str, Any]) -> Dashboard: """创建看板""" dashboard = Dashboard.from_dict(dashboard_data) self.dashboards[dashboard.id] = dashboard return dashboard def update_dashboard(self, dashboard_id: str, dashboard_data: Dict[str, Any]) -> Optional[Dashboard]: """更新看板""" if dashboard_id in self.dashboards: dashboard = Dashboard.from_dict(dashboard_data) dashboard.id = dashboard_id # 保持ID不变 self.dashboards[dashboard_id] = dashboard return dashboard return None def delete_dashboard(self, dashboard_id: str) -> bool: """删除看板""" if dashboard_id in self.dashboards: del self.dashboards[dashboard_id] return True return False def get_data_source(self, source_id: str) -> Optional[DataSource]: """获取数据源""" return self.data_sources.get(source_id) def get_all_data_sources(self) -> List[DataSource]: """获取所有数据源""" return list(self.data_sources.values()) def get_dataset(self, dataset_id: str) -> Optional[Dataset]: """获取数据集""" return self.datasets.get(dataset_id) def get_all_datasets(self) -> List[Dataset]: """获取所有数据集""" return list(self.datasets.values()) def execute_chart_data(self, chart_id: str) -> Dict[str, Any]: """执行图表数据查询""" chart = self.get_chart(chart_id) if not chart: return {"error": "Chart not found"} # 这里模拟数据查询，实际应该连接到数据库或数据源 data = self._generate_chart_data(chart) return { "chart_id": chart_id, "chart_name": chart.name, "data": data, "columns": chart.options.get("columns", []), "chart_type": chart.chart_type.value, "options": chart.options } def _generate_chart_data(self, chart: Chart) -> List[Dict[str, Any]]: """生成图表数据（模拟）""" # 根据图表类型和数据源生成模拟数据 if chart.data_source == DataSourceType.SENSOR_DATA: return self._generate_sensor_data(chart) elif chart.data_source == DataSourceType.DEVICE_DATA: return self._generate_device_data(chart) elif chart.data_source == DataSourceType.ALERT_DATA: return self._generate_alert_data(chart) elif chart.data_source == DataSourceType.SYSTEM_STATS: return self._generate_stats_data(chart) else: return [] def _generate_sensor_data(self, chart: Chart) -> List[Dict[str, Any]]: """生成传感器数据""" data = [] # 生成时间序列数据 base_time = datetime.now() - timedelta(days=7) locations = ["A区", "B区", "C区", "D区"] for i in range(24 * 7): # 7天，每小时一个点 timestamp = base_time + timedelta(hours=i) for location in locations: # 添加一些随机波动 base_value = 50 if chart.filters.get("data_type") == "LL" else 1.0 value = base_value + np.random.normal(0, 10) data.append({ "timestamp": timestamp.isoformat(), "location": location, "value": round(value, 2), "device_id": f"device_{hash(location) % 10 + 1}", "data_type": chart.filters.get("data_type", "LL"), "quality_score": round(np.random.uniform(0.8, 1.0), 2) }) # 应用过滤和聚合 if chart.group_by: # 简单的分组聚合 grouped_data = {} for item in data: key = tuple(item.get(field) for field in chart.group_by) if key not in grouped_data: grouped_data[key] = [] grouped_data[key].append(item) result = [] for key, items in grouped_data.items(): group_data = {} for i, field in enumerate(chart.group_by): group_data[field] = key[i] # 聚合计算 values = [item["value"] for item in items] if chart.aggregation == "avg": group_data["value"] = sum(values) / len(values) elif chart.aggregation == "sum": group_data["value"] = sum(values) elif chart.aggregation == "max": group_data["value"] = max(values) elif chart.aggregation == "min": group_data["value"] = min(values) else: group_data["value"] = sum(values) / len(values) result.append(group_data) return result return data def _generate_device_data(self, chart: Chart) -> List[Dict[str, Any]]: """生成设备数据""" devices = [ {"id": "device_1", "name": "流量计-001", "device_type": "流量计", "location": "A区", "status": "active"}, {"id": "device_2", "name": "压力计-001", "device_type": "压力计", "location": "A区", "status": "active"}, {"id": "device_3", "name": "水位计-001", "device_type": "水位计", "location": "B区", "status": "maintenance"}, {"id": "device_4", "name": "浊度计-001", "device_type": "浊度计", "location": "B区", "status": "active"}, {"id": "device_5", "name": "pH计-001", "device_type": "pH计", "location": "C区", "status": "inactive"}, ] # 按状态分组 status_groups = {} for device in devices: status = device["status"] if status not in status_groups: status_groups[status] = [] status_groups[status].append(device) # 生成统计数据 result = [] for status, devices_in_status in status_groups.items(): result.append({ "status": status, "count": len(devices_in_status), "devices": [d["name"] for d in devices_in_status] }) return result def _generate_alert_data(self, chart: Chart) -> List[Dict[str, Any]]: """生成警报数据""" alerts = [ {"level": "info", "count": 5, "description": "信息级别警报"}, {"level": "warning", "count": 3, "description": "警告级别警报"}, {"level": "error", "count": 1, "description": "错误级别警报"}, {"level": "critical", "count": 0, "description": "严重级别警报"}, ] return alerts def _generate_stats_data(self, chart: Chart) -> List[Dict[str, Any]]: """生成系统统计数据""" data = [] base_time = datetime.now() - timedelta(days=1) for i in range(24): # 24小时数据 timestamp = base_time + timedelta(hours=i) data.append({ "timestamp": timestamp.isoformat(), "total_records": 1000 + np.random.randint(-100, 100), "total_devices": 25 + np.random.randint(-5, 5), "active_connections": 5 + np.random.randint(-2, 3), "api_requests_count": 150 + np.random.randint(-30, 30), "alerts_count": np.random.randint(0, 5), "data_quality_score": round(np.random.uniform(0.9, 1.0), 2), "memory_usage_mb": 100 + np.random.randint(-20, 20), "cpu_usage_percent": 30 + np.random.randint(-10, 10) }) return data def setup_superset_integration(self, integration_data: Dict[str, Any]) -> SupersetIntegration: """设置Superset集成""" integration = SupersetIntegration.from_dict(integration_data) self.superset_integration = integration return integration def setup_metabase_integration(self, integration_data: Dict[str, Any]) -> MetabaseIntegration: """设置Metabase集成""" integration = MetabaseIntegration.from_dict(integration_data) self.metabase_integration = integration return integration def get_chart_data_api(self, chart_id: str) -> Dict[str, Any]: """获取图表数据API接口""" return self.execute_chart_data(chart_id) def get_dashboard_data_api(self, dashboard_id: str) -> Dict[str, Any]: """获取看板数据API接口""" dashboard = self.get_dashboard(dashboard_id) if not dashboard: return {"error": "Dashboard not found"} charts_data = {} for chart_id in dashboard.charts: charts_data[chart_id] = self.get_chart_data_api(chart_id) return { "dashboard_id": dashboard_id, "dashboard_name": dashboard.name, "charts": charts_data, "layout": dashboard.layout } def get_public_dashboards(self) -> List[Dashboard]: """获取公开看板""" return [db for db in self.dashboards.values() if db.is_public] def get_charts_by_tag(self, tag: str) -> List[Chart]: """根据标签获取图表""" return [chart for chart in self.charts.values() if tag in chart.tags] def get_dashboards_by_tag(self, tag: str) -> List[Dashboard]: """根据标签获取看板""" return [dashboard for dashboard in self.dashboards.values() if tag in dashboard.tags] def search_charts(self, keyword: str) -> List[Chart]: """搜索图表""" keyword = keyword.lower() return [ chart for chart in self.charts.values() if keyword in chart.name.lower() or keyword in chart.description.lower() or any(keyword in tag.lower() for tag in chart.tags) ] def search_dashboards(self, keyword: str) -> List[Dashboard]: """搜索看板""" keyword = keyword.lower() return [ dashboard for dashboard in self.dashboards.values() if keyword in dashboard.name.lower() or keyword in dashboard.description.lower() or any(keyword in tag.lower() for tag in dashboard.tags) ]