feat: Updates in AI Analyzer Engine Implementation (d18cfa85) · Commits · TFS / controller

src/tests/mwc26-f5ga/AI_analytics_engine/ai_model/ai_processor.py

+94 −67

Original line number	Diff line number	Diff line
		@@ -53,31 +53,53 @@ class AIModelProcessor:

		def ai_model_processor(
		self,
		metric_values: list[float]
		) -> Optional[list[float]]:
		metric_values: list[dict[str, Any]],
		) -> Optional[list[dict[str, Any]]]:
		"""
		Process device and performance data through AI models.

		Args:
		metric_values: List of performance metric values.
		metric_values: List of dictionaries containing performance metric values.
		Each dict has keys like 'bandwidth_utilization', 'latency', etc.
		Returns:
		List of 3 forecasted values, or None if insufficient data.
		List of dicts containing forecasted values for each metric,
		or None if insufficient data.

		"""
		LOGGER.debug("Processing data through AI models")
		# LOGGER.debug(f"Number of performance data points: {len(metric_values)}")
		LOGGER.debug(f"Number of performance data points: {len(metric_values)}")

		if not metric_values or len(metric_values) < 4:
		LOGGER.warning("Insufficient data for forecasting")
		if not metric_values or len(metric_values) < 3:
		LOGGER.warning("Insufficient data for forecasting (need at least 3 samples)")
		return None

		results = []

		try:
		# Convert metric_values to pandas Series
		data = pd.Series(metric_values)
		# Convert list of dicts to DataFrame for easier processing
		df = pd.DataFrame(metric_values)

		# Process each metric column separately
		for column in df.columns:
		data = df[column]

		# Skip non-numeric columns
		if not pd.api.types.is_numeric_dtype(data):
		LOGGER.debug(f"Skipping non-numeric column: {column}")
		continue

		# Remove NaN values
		data = data.dropna()

		if len(data) < 3:
		LOGGER.warning(f"Insufficient data for column {column} (need at least 3 samples)")
		continue

		LOGGER.debug(f"Processing column: {column} with {len(data)} samples")

		# Create and fit Exponential Smoothing model
		model = ExponentialSmoothing(
		data,
		endog = data,
		trend = "add",
		seasonal = None # No seasonal component for this data
		)
		@@ -86,18 +108,22 @@ class AIModelProcessor:

		# Forecast next 3 values
		forecast = fit.forecast(steps=3)

		forecasted_values = forecast.tolist()

		# Calculate confidence score based on model fit quality
		# Using residual standard error as inverse confidence metric
		error = {}
		residuals = fit.resid
		mse = (residuals ** 2).mean()
		rmse = mse ** 0.5

		error['mse'] = float(mse)
		error['rmse'] = float(rmse)

		# Normalize confidence: lower RMSE = higher confidence
		# Use data scale (std dev) to normalize RMSE
		data_std = data.std()

		if data_std > 0:
		normalized_error = rmse / data_std
		# Convert to confidence score (0-1 range, higher is better)
		@@ -105,21 +131,27 @@ class AIModelProcessor:
		else:
		confidence = 0.5 # Default if std dev is 0

		# LOGGER.info(f"Model RMSE: {rmse:.4f}, Confidence: {confidence:.4f}")
		LOGGER.info(f"Forecasted next 3 values: {forecasted_values}")
		LOGGER.info(f"Metric: {column}, RMSE: {rmse:.4f}, Data Std: {data_std:.4f}, Confidence: {confidence:.4f}")
		LOGGER.info(f"Forecasted next 3 values for {column}: {forecasted_values}")

		# return forecasted_values
		return [confidence]
		results.append({
		"metric_name": column,
		"forecasted_values": forecasted_values,
		"confidence": float(confidence),
		"sample_interval": 5,
		"error_metrics": error,
		})

		return results if results else None

		except Exception as e:
		LOGGER.error(f"Error during forecasting: {e}")
		LOGGER.error(f"Error during forecasting: {e}", exc_info=True)
		return None


		def process_data(
		self,
		performance_data: Dict[str, Any],
		sla_policy: SLAPolicyConfig
		) -> Dict[str, Any]:
		"""
		Process device and performance data through AI models.
		@@ -139,28 +171,23 @@ class AIModelProcessor:
		LOGGER.debug(f"Number of performance data points: {len(metric_values)}")
		# LOGGER.debug(f"Performance data values: {metric_values}")

		# forecasted_values = self.ai_model_processor(metric_values)
		# if forecasted_values is None:
		# LOGGER.warning("AI model processing failed or insufficient data")

		# if forecasted_values:
		# # Exponential weights: more weight on earlier (starting) values
		# # Example: for 3 values -> weights = [0.5, 0.33, 0.17] (exponential decay)
		# weights = [2**(-i) for i in range(len(forecasted_values))]
		# # Normalize weights to sum to 1
		# total_weight = sum(weights)
		# weights = [w / total_weight for w in weights]
		# score = average(forecasted_values, weights=weights)
		# # LOGGER.debug(f"Weighted average with exponential weights: {score}, weights: {weights}")
		# else:
		# score = None

		score = self.ai_model_processor(metric_values)
		if not metric_values:
		LOGGER.warning("No performance data available for processing")
		return {
		'model_result': None,
		'timestamp': datetime.now(UTC).isoformat()
		}
		result = self.ai_model_processor(metric_values)

		if result is None:
		# fallback score structure
		LOGGER.warning("AI model processing failed or insufficient data. See logs for details.")
		return {
		'confidence_scores': score,
		'summary': {
		'sla_policy': sla_policy.to_dict(),
		'model_result': None,
		'timestamp': datetime.now(UTC).isoformat()
		}

		return {
		'model_result': result,
		'timestamp': datetime.now(UTC).isoformat()
		}

src/tests/mwc26-f5ga/AI_analytics_engine/ai_model/sla_policy.py

+16 −8

Original line number	Diff line number	Diff line
		@@ -33,11 +33,15 @@ class SLAPolicyConfig:
		bandwidth_utilization_threshold_pct: Maximum acceptable bandwidth
		utilization as a percentage (0-100).
		time_window_seconds: Time window in seconds for data analysis.
		sample_interval_sec: Sampling interval in seconds for data collection.
		sample_count: Minimum number of samples to fetch from database.
		"""
		simap_id: str
		latency_threshold_ms: float\|None
		bandwidth_utilization_threshold_pct: float\|None
		bandwidth_utilization: float\|None
		time_window_seconds: int
		sample_interval_sec: int
		sample_count: int

		@classmethod
		def from_dict(cls, data: Dict[str, Any]) -> SLAPolicyConfig:
		@@ -47,7 +51,7 @@ class SLAPolicyConfig:
		Args:
		data: Dictionary containing the SLA policy configuration fields.
		Required keys: 'simap_id', 'latency_threshold_ms',
		'bandwidth_utilization_threshold_pct', 'time_window_seconds'.
		'bandwidth_utilization', 'time_window_seconds'.
		Supports nested 'sla_metrics' structure.

		Returns:
		@@ -62,14 +66,18 @@ class SLAPolicyConfig:
		simap_id = str(data['simap_id'])
		metrics = data['sla_metrics']
		latency_threshold_ms = float(metrics['latency_threshold_ms'])
		bandwidth_threshold = float(metrics.get('bandwidth_utilization_threshold_pct', 0.0))
		time_window = int(data.get('window_size_sec', 300))
		bandwidth_threshold = float(metrics.get('bandwidth_utilization', 0.0))
		time_window = int(data['history_window_size_sec'])
		sample_interval = int(data['sample_interval_sec'])
		sample_count = int(data['sample_count'])

		return cls(
		simap_id = simap_id,
		latency_threshold_ms = latency_threshold_ms,
		bandwidth_utilization_threshold_pct = bandwidth_threshold,
		time_window_seconds = time_window
		bandwidth_utilization = bandwidth_threshold,
		time_window_seconds = time_window,
		sample_interval_sec = sample_interval,
		sample_count = sample_count
		)
		except KeyError as e:
		raise KeyError(f"Missing required field: {e.args[0]}") from e

src/tests/mwc26-f5ga/AI_analytics_engine/api/api_blueprint.py

+12 −3

Original line number	Diff line number	Diff line
		@@ -124,10 +124,11 @@ def create_ai_analytics_blueprint(
		# >>> Step 3: Process data through AI models
		LOGGER.debug(">>> Step 3: Processing data through AI models")
		results = ai_processor.process_data(
		performance_data, sla_policy
		performance_data
		)

		# >>> Step 4: Send results to Decision Engine
		results['simap_id'] = sla_policy.simap_id # Include SIMAP ID in results
		LOGGER.debug(">>> Step 4: Sending results to Decision Engine")
		if not decision_client.send_results(results):
		LOGGER.error("Failed to send results to Decision Engine")
		@@ -181,20 +182,28 @@ def create_ai_analytics_blueprint(

		Returns:
		JSON response with SIMAP and InfluxDB connection details.
		Sensitive values (passwords, tokens) are masked.
		"""
		LOGGER.debug("Configuration requested")

		def mask_secret(value: str) -> str:
		"""Mask sensitive values for display."""
		if not value:
		return '(not set)'
		return f"{value[:2]}{'' (len(value) - 2)}" if len(value) > 2 else '***'

		return jsonify({
		'simap': {
		'scheme': Config.SIMAP_SERVER_SCHEME,
		'address': Config.SIMAP_SERVER_ADDRESS,
		'port': Config.SIMAP_SERVER_PORT,
		'username': Config.SIMAP_SERVER_USERNAME,
		'password': Config.SIMAP_SERVER_PASSWORD
		'password': mask_secret(Config.SIMAP_SERVER_PASSWORD)
		},
		'influxdb': {
		'host': Config.INFLUXDB_HOST,
		'port': Config.INFLUXDB_PORT,
		'token': Config.INFLUXDB_TOKEN,
		'token': mask_secret(Config.INFLUXDB_TOKEN),
		'database': Config.INFLUXDB_DATABASE
		},
		'api': {

src/tests/mwc26-f5ga/AI_analytics_engine/clients/influxdb_fetcher.py

+97 −71

Original line number	Diff line number	Diff line
		@@ -100,7 +100,6 @@ class InfluxDBFetcher:
		def process_response_table(
		self,
		table: Any,
		metric_to_process: str
		) -> Dict[str, Any]:
		"""
		Process InfluxDB response table into structured data.
		@@ -111,45 +110,36 @@ class InfluxDBFetcher:
		Returns:
		Dictionary containing processed performance metrics and values.
		"""
		metrics = []
		if table is not None and isinstance(table, pd.DataFrame):
		metrics = table.to_dict('records')
		else:
		if table is None or not isinstance(table, pd.DataFrame):
		LOGGER.warning("No data returned from InfluxDB query")
		return {
		'metrics': [],
		'metric_values': []
		}

		LOGGER.debug(f"Processing {len(table)} rows from InfluxDB response")

		# Define columns for each output dataframe
		full_columns = ['bandwidth_utilization', 'latency', 'time', 'link_id']
		metric_columns = ['bandwidth_utilization', 'latency']

		keys_to_check = [ 'time', 'link_id']
		sla_metric = 'bandwidth_utilization'
		keys_to_check.append(sla_metric)

		# if metric_to_process == 'latency_threshold_ms':
		# sla_metric = 'latency'
		# keys_to_check.append(sla_metric)
		# elif metric_to_process == 'bandwidth_utilization_threshold_pct':
		# sla_metric = 'bandwidth_utilization'
		# keys_to_check.append(sla_metric)
		# else:
		# sla_metric = None
		# LOGGER.warning(f"Unknown metric to process: {metric_to_process}")


		LOGGER.debug(f"Processed {len(metrics)} metric records from InfluxDB response")
		data = []
		metric_value = []
		if sla_metric is not None:
		for row in metrics:
		# LOGGER.debug(f"Metric record: {row}")
		new_row = {}
		for key, value in row.items():
		if key in keys_to_check:
		new_row[key] = value
		if key == sla_metric:
		metric_value.append(value)
		data.append(new_row)
		LOGGER.debug(f">>> Processed metric values: {metric_value}")
		# Create DataFrame 1: Full metrics with time and link_id
		# Filter only columns that exist in the table
		available_full_cols = [col for col in full_columns if col in table.columns]
		df_full = table[available_full_cols]
		metrics = df_full.to_dict('records')

		# Create DataFrame 2: Only metric values (bandwidth_utilization, latency)
		available_metric_cols = [col for col in metric_columns if col in table.columns]
		df_metrics = table[available_metric_cols]
		metric_values = df_metrics.to_dict('records')

		LOGGER.debug(f"Processed {len(metrics)} metric records with {len(available_full_cols)} columns")
		LOGGER.debug(f"Extracted {len(metric_values)} metric value records with {len(available_metric_cols)} columns")

		return {
		'metrics': data,
		'metric_values': metric_value
		'metrics': metrics,
		'metric_values': metric_values
		}


		@@ -165,13 +155,19 @@ class InfluxDBFetcher:
		SLA policy parameters and device information. The retry decorator
		ensures resilience against transient failures.

		If the initial query returns fewer samples than required by
		sla_policy.sample_count, the method will automatically fetch
		older data with an extended time window until the required
		sample count is met or max attempts are reached.

		Args:
		sla_policy: The SLA policy configuration containing time window
		and threshold parameters.
		sla_policy: The SLA policy configuration containing time window,
		threshold parameters, and required sample count.

		Returns:
		Dictionary containing:
		- 'metrics': List of performance metric records.
		- 'metric_values': List of metric values only.
		- 'timestamp_range': Dictionary with 'start' and 'end'
		timestamps for the queried data.

		@@ -186,49 +182,79 @@ class InfluxDBFetcher:
		if sla_policy.latency_threshold_ms is None:
		raise ValueError("SLA policy missing latency threshold for data fetch")

		metric_to_process = sla_policy.latency_threshold_ms

		LOGGER.debug(
		f"Fetching performance data for simap_id={sla_policy.simap_id}, "
		f"time_window={sla_policy.time_window_seconds}s "
		f"for metric={metric_to_process} "
		f"time_window={sla_policy.time_window_seconds}s, "
		f"required_samples={sla_policy.sample_count}"
		)

		try:
		# Initial time window
		current_time_window = sla_policy.time_window_seconds
		max_attempts = 3
		attempt = 1
		final_table = None

		while attempt <= max_attempts:
		query = (
		f"SELECT * FROM link_telemetry "
		f"WHERE link_id = '{sla_policy.simap_id}' "
		f"AND time >= now() - INTERVAL '{sla_policy.time_window_seconds} seconds' "
		f"AND time >= now() - INTERVAL '{current_time_window} seconds' "
		f"ORDER BY time DESC"
		)

		LOGGER.debug(f"Executing query: {query}")
		LOGGER.debug(f"Attempt {attempt}/{max_attempts}: Executing query with time_window={current_time_window}s")
		LOGGER.debug(f"Query: {query}")

		table = self._client.query(query=query, language="sql", mode="pandas")
		final_table = self._client.query(query=query, language="sql", mode="pandas")

		result = self.process_response_table(table, metric_to_process)
		# metrics = result.get('metrics', [])
		# Count samples from raw table
		samples_fetched = 0 if final_table is None or not isinstance(final_table, pd.DataFrame) else len(final_table)

		# start_time = datetime.now(timezone.utc)
		# end_time = datetime.now(timezone.utc)
		# if metrics:
		# times = [m.get('time') for m in metrics if m.get('time')]
		# if times is not None:
		# start_time = min(times)
		# end_time = max(times)
		LOGGER.info(
		f"Attempt {attempt}: Fetched {samples_fetched} samples "
		f"(required: {sla_policy.sample_count})"
		)

		LOGGER.info(f"Fetched {len(result.get('metrics', []))} metric records for simap_id={sla_policy.simap_id}")
		# Check if we have enough samples
		if samples_fetched >= sla_policy.sample_count:
		LOGGER.info(f"Required samples met")
		break

		# If not enough samples and not last attempt, calculate new time window
		if attempt < max_attempts:
		if samples_fetched > 0:
		# Calculate required time window based on sample density
		# Formula: new_window = current_window * (required_samples / fetched_samples) * 1.2
		# The 1.2 factor adds 20% buffer to account for non-uniform data distribution
		ratio = sla_policy.sample_count / samples_fetched
		current_time_window = int(current_time_window * ratio * 1.2)
		LOGGER.debug(f"Extending time window to {current_time_window}s(ratio: {ratio:.2f})")
		else:
		# If no samples, double the time window
		current_time_window *= 2
		LOGGER.warning(f"No samples found, doubling time window to {current_time_window}s")

		attempt += 1
		else:
		LOGGER.warning(
		f"Max attempts reached. Returning {samples_fetched} samples "
		f"(required: {sla_policy.sample_count})"
		)
		break

		# Process the response table after fetch is completed
		result = self.process_response_table(final_table)

		return {
		'metrics': result.get('metrics', []),
		'metric_values': result.get('metric_values', []),
		'timestamp_range': {
		# 'start': start_time.isoformat() if isinstance(start_time, datetime) else str(start_time),
		# 'end': end_time.isoformat() if isinstance(end_time, datetime) else str(end_time)
		'fetch_window_size_sec': current_time_window,
		'timestamp_range': {},
		}
		}
		finally:
		self._client.close()
		except Exception as e:
		LOGGER.error(f"Error fetching performance data from InfluxDB: {e}", exc_info=True)
		raise e

		@RETRY_DECORATOR
		def notify_telemetry_update(

src/tests/mwc26-f5ga/AI_analytics_engine/requirements.in

+3 −0

Original line number	Diff line number	Diff line
		@@ -2,3 +2,6 @@
		flask>=2.3.0
		requests>=2.31.0
		influxdb3-python>=0.8.0
		pandas>=2.0.0
		statsmodels>=0.14.0
		numpy>=1.24.0