defmodule ToweropsWeb.DeviceLive.Helpers.ChartBuilders do @moduledoc """ Chart data building functions for DeviceLive.Show. Pure functions that transform time-series data into Chart.js-compatible JSON. All functions return nil for empty data or JSON-encoded strings for valid datasets. """ alias Towerops.Monitoring alias Towerops.Snmp require Logger @doc """ Loads processor CPU usage chart data for the last 24 hours. Returns JSON-encoded chart data or nil if no data available. """ @spec load_processor_chart_data(list()) :: String.t() | nil def load_processor_chart_data([]), do: nil def load_processor_chart_data(processors) do twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) datasets = processors |> Enum.map(&processor_to_chart_dataset(&1, twenty_four_hours_ago)) |> Enum.reject(&Enum.empty?(&1.data)) if Enum.empty?(datasets) do nil else safe_encode_chart_data(%{datasets: datasets}) end end defp processor_to_chart_dataset(processor, since) do readings = processor.id |> Snmp.get_processor_readings(since: since, limit: 1000) |> Enum.reverse() label = if processor.description do processor.description else "CPU #{processor.processor_index}" end %{ label: label, data: Enum.map(readings, &processor_reading_to_data_point/1) } end defp processor_reading_to_data_point(reading) do %{ x: DateTime.to_unix(reading.checked_at, :millisecond), y: if(reading.load_percent, do: Float.round(reading.load_percent, 1)) } end @doc """ Loads storage volume usage chart data for the last 24 hours. Returns JSON-encoded chart data or nil if no data available. """ @spec load_storage_volume_chart_data(list()) :: String.t() | nil def load_storage_volume_chart_data([]), do: nil def load_storage_volume_chart_data(storage_volumes) do twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) datasets = storage_volumes |> Enum.map(&storage_volume_to_chart_dataset(&1, twenty_four_hours_ago)) |> Enum.reject(&Enum.empty?(&1.data)) if Enum.empty?(datasets) do nil else safe_encode_chart_data(%{datasets: datasets}) end end defp storage_volume_to_chart_dataset(storage, since) do readings = storage.id |> Snmp.get_storage_readings(since: since, limit: 1000) |> Enum.reverse() label = storage.description || "Storage #{storage.storage_index}" %{ label: label, data: Enum.map(readings, &storage_reading_to_data_point/1) } end defp storage_reading_to_data_point(reading) do %{ x: DateTime.to_unix(reading.checked_at, :millisecond), y: if(reading.usage_percent, do: Float.round(reading.usage_percent, 1)) } end @doc """ Loads sensor chart data for specific sensor types. Returns JSON-encoded chart data or nil if no matching sensors found. """ @spec load_sensor_chart_data(Snmp.SNMPDevice.t() | nil, list(String.t())) :: String.t() | nil def load_sensor_chart_data(nil, _sensor_types), do: nil def load_sensor_chart_data(snmp_device, sensor_types) when is_list(sensor_types) do build_sensor_datasets_json(snmp_device, sensor_types) end defp build_sensor_datasets_json(device, sensor_types) do sensors = device.sensors |> Enum.filter(&(&1.sensor_type in sensor_types)) |> Enum.sort_by(& &1.sensor_index) if Enum.empty?(sensors) do nil else twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) datasets = Enum.map(sensors, &sensor_to_chart_dataset(&1, twenty_four_hours_ago)) safe_encode_chart_data(%{datasets: datasets}) end end defp sensor_to_chart_dataset(sensor, since) do readings = sensor.id |> Snmp.get_sensor_readings(since: since, limit: 1000) |> Enum.reverse() %{ label: sensor.sensor_descr, data: Enum.map(readings, &reading_to_data_point/1) } end defp reading_to_data_point(reading) do %{ x: DateTime.to_unix(reading.checked_at, :millisecond), y: if(reading.value, do: Float.round(reading.value, 1)) } end @doc """ Loads ICMP latency chart data for the last 24 hours. Returns JSON-encoded chart data or nil if no latency data available. """ @spec load_latency_chart_data(binary()) :: String.t() | nil def load_latency_chart_data(device_id) do twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) checks = device_id |> Monitoring.get_latency_data(since: twenty_four_hours_ago, limit: 1000) |> Enum.reverse() if Enum.empty?(checks) do nil else dataset = %{ label: "ICMP Latency", data: Enum.map(checks, &latency_check_to_data_point/1) } safe_encode_chart_data(%{datasets: [dataset]}) end end defp latency_check_to_data_point(check) do %{ x: DateTime.to_unix(check.checked_at, :millisecond), y: check.response_time_ms } end @doc """ Loads overall device traffic chart data (aggregate of all interfaces). Returns JSON-encoded chart data or nil if no traffic data available. """ @spec load_overall_traffic_chart_data(Snmp.SNMPDevice.t() | nil) :: String.t() | nil def load_overall_traffic_chart_data(nil), do: nil def load_overall_traffic_chart_data(snmp_device) do build_overall_traffic_chart_json(snmp_device) end defp build_overall_traffic_chart_json(device) do if Enum.empty?(device.interfaces) do nil else twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) all_stats = aggregate_interface_traffic_stats(device.interfaces, twenty_four_hours_ago) if Enum.empty?(all_stats) do nil else build_traffic_json_datasets(all_stats, device) end end end # Calculate BPS per interface first, then aggregate by timestamp # This avoids issues with misaligned timestamps between interfaces # Processes all interfaces (regardless of monitored status) to show total device traffic defp aggregate_interface_traffic_stats(interfaces, since) do interfaces |> Enum.flat_map(fn interface -> interface.id |> Snmp.get_interface_stats(since: since, limit: 1000) |> Enum.reverse() |> calculate_interface_bps() end) |> Enum.group_by(& &1.timestamp_ms) |> Enum.map(fn {ts, bps_list} -> total_in = Enum.reduce(bps_list, 0.0, fn x, acc -> acc + x.in_bps end) total_out = Enum.reduce(bps_list, 0.0, fn x, acc -> acc + x.out_bps end) %{timestamp_ms: ts, in_bps: total_in, out_bps: total_out} end) |> Enum.sort_by(& &1.timestamp_ms) end # Calculate BPS for a single interface's stats defp calculate_interface_bps(stats) do stats |> Enum.chunk_every(2, 1, :discard) |> Enum.map(fn [s1, s2] -> time_diff = s2.checked_at |> DateTime.diff(s1.checked_at, :second) |> max(1) in_bps = if s2.if_in_octets && s1.if_in_octets do calculate_counter_bps(s1.if_in_octets, s2.if_in_octets, time_diff) else 0.0 end out_bps = if s2.if_out_octets && s1.if_out_octets do calculate_counter_bps(s1.if_out_octets, s2.if_out_octets, time_diff) else 0.0 end # Round timestamp to nearest minute for aggregation (avoids microsecond misalignment) ts_seconds = DateTime.to_unix(s2.checked_at) rounded_ts_ms = div(ts_seconds, 60) * 60 * 1000 %{timestamp_ms: rounded_ts_ms, in_bps: in_bps, out_bps: out_bps} end) end defp calculate_counter_bps(prev_octets, curr_octets, time_diff) do Towerops.Capacity.calculate_bps(prev_octets, curr_octets, time_diff) end defp build_traffic_json_datasets(all_stats, device) do in_data = Enum.map(all_stats, fn s -> %{x: s.timestamp_ms, y: Float.round(s.in_bps, 2)} end) out_data = Enum.map(all_stats, fn s -> %{x: s.timestamp_ms, y: -Float.round(s.out_bps, 2)} end) datasets = [ %{label: "Outbound", data: out_data}, %{label: "Inbound", data: in_data} ] # For backhaul devices, add capacity reference lines datasets = if device.device.device_role == "backhaul" do add_capacity_datasets(datasets, all_stats, device) else datasets end Jason.encode!(%{datasets: datasets}) end # Add capacity reference lines for backhaul devices # Calculates capacity per timestamp based on which interfaces have data at that point # For radios with Rx/Tx Capacity sensors, uses those values as total device capacity defp add_capacity_datasets(datasets, all_stats, device) do # Get capacity sensors if available (for radios like AirFiber) rx_capacity_sensor = Enum.find(device.sensors, &(&1.sensor_descr == "Rx Capacity")) tx_capacity_sensor = Enum.find(device.sensors, &(&1.sensor_descr == "Tx Capacity")) # If we have capacity sensors, use them as the total device capacity # Otherwise, calculate from interface speeds capacity_data = if rx_capacity_sensor || tx_capacity_sensor do calculate_sensor_capacity_data(all_stats, rx_capacity_sensor, tx_capacity_sensor) else calculate_interface_capacity_data(all_stats, device) end # Only add capacity datasets if we have at least some capacity data max_rx = capacity_data |> Enum.map(fn {_, rx, _} -> rx end) |> Enum.max(fn -> 0 end) max_tx = capacity_data |> Enum.map(fn {_, _, tx} -> tx end) |> Enum.max(fn -> 0 end) if max_rx > 0 || max_tx > 0 do capacity_in_data = Enum.map(capacity_data, fn {ts, rx, _tx} -> %{x: ts, y: Float.round(rx * 1.0, 2)} end) capacity_out_data = Enum.map(capacity_data, fn {ts, _rx, tx} -> %{x: ts, y: -Float.round(tx * 1.0, 2)} end) # Add capacity datasets with special styling datasets ++ [ %{ label: "Capacity (In)", data: capacity_in_data, borderDash: [5, 5], borderWidth: 2, pointRadius: 0, fill: false }, %{ label: "Capacity (Out)", data: capacity_out_data, borderDash: [5, 5], borderWidth: 2, pointRadius: 0, fill: false } ] else datasets end end # Calculate capacity data using sensor values (for radios with Rx/Tx Capacity sensors) # Sensor values represent total device capacity, not per-interface # Loads time-series sensor readings to create a dynamic capacity line defp calculate_sensor_capacity_data(all_stats, rx_capacity_sensor, tx_capacity_sensor) do twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) # Load time-series readings for both sensors rx_readings = load_capacity_sensor_readings(rx_capacity_sensor, twenty_four_hours_ago) tx_readings = load_capacity_sensor_readings(tx_capacity_sensor, twenty_four_hours_ago) # If we have sensor readings, use them directly (they have their own timestamps) # Otherwise fall back to static values at traffic timestamps if map_size(rx_readings) > 0 || map_size(tx_readings) > 0 do build_capacity_data_from_sensors( rx_readings, tx_readings, rx_capacity_sensor, tx_capacity_sensor ) else build_static_capacity_data(all_stats, rx_capacity_sensor, tx_capacity_sensor) end end # Load sensor readings and convert to timestamp -> bps map defp load_capacity_sensor_readings(nil, _since), do: %{} defp load_capacity_sensor_readings(sensor, since) do sensor.id |> Snmp.get_sensor_readings(since: since, limit: 1000) |> Map.new(&sensor_reading_to_capacity_point/1) end # Convert a sensor reading to {timestamp_ms, bps} tuple # Sensor values are in Mbps, convert to bps defp sensor_reading_to_capacity_point(reading) do bps = capacity_value_to_bps(reading.value) ts_ms = DateTime.to_unix(reading.checked_at, :millisecond) {ts_ms, bps} end # Convert capacity sensor value (in Mbps) to bps defp capacity_value_to_bps(nil), do: 0 defp capacity_value_to_bps(value) when is_number(value), do: round(value * 1_000_000) defp capacity_value_to_bps(_), do: 0 # Build capacity data points from sensor readings defp build_capacity_data_from_sensors(rx_readings, tx_readings, rx_sensor, tx_sensor) do # Combine all unique timestamps from both sensors all_timestamps = (Map.keys(rx_readings) ++ Map.keys(tx_readings)) |> MapSet.new() |> Enum.sort() # Build capacity data points using sensor readings Enum.map(all_timestamps, fn ts_ms -> rx_bps = Map.get(rx_readings, ts_ms, get_fallback_capacity_bps(rx_sensor)) tx_bps = Map.get(tx_readings, ts_ms, get_fallback_capacity_bps(tx_sensor)) {ts_ms, rx_bps, tx_bps} end) end # Build static capacity data at traffic timestamps defp build_static_capacity_data(all_stats, rx_sensor, tx_sensor) do rx_cap_bps = get_fallback_capacity_bps(rx_sensor) tx_cap_bps = get_fallback_capacity_bps(tx_sensor) Enum.map(all_stats, fn stat -> {stat.timestamp_ms, rx_cap_bps, tx_cap_bps} end) end # Get fallback capacity in bps from sensor's last_value (already in Mbps) defp get_fallback_capacity_bps(nil), do: 0 defp get_fallback_capacity_bps(sensor) do capacity_value_to_bps(sensor.last_value) end # Calculate capacity data from interface configured_capacity_bps or if_speed # Sums capacity of active interfaces at each timestamp defp calculate_interface_capacity_data(all_stats, device) do # Build a map of interface_id -> {rx_capacity, tx_capacity} for quick lookup # Includes all interfaces to match the total device traffic interface_capacity_map = Map.new(device.interfaces, fn interface -> cap = interface.configured_capacity_bps || interface.if_speed || 0 {interface.id, {cap, cap}} end) # Get the raw interface stats to determine which interfaces were active at each timestamp twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour) # Track which interfaces have data at each timestamp interface_activity_by_timestamp = device.interfaces |> Enum.flat_map(fn interface -> interface.id |> Snmp.get_interface_stats(since: twenty_four_hours_ago, limit: 1000) |> Enum.map(fn stat -> ts_seconds = DateTime.to_unix(stat.checked_at) rounded_ts_ms = div(ts_seconds, 60) * 60 * 1000 {rounded_ts_ms, interface.id} end) end) |> Enum.group_by(fn {ts, _} -> ts end, fn {_, iface_id} -> iface_id end) # Calculate capacity at each timestamp based on active interfaces Enum.map(all_stats, fn stat -> active_interfaces = Map.get(interface_activity_by_timestamp, stat.timestamp_ms, []) {total_rx, total_tx} = active_interfaces |> Enum.uniq() |> Enum.reduce({0, 0}, fn iface_id, {rx_acc, tx_acc} -> {rx_cap, tx_cap} = Map.get(interface_capacity_map, iface_id, {0, 0}) {rx_acc + rx_cap, tx_acc + tx_cap} end) {stat.timestamp_ms, total_rx, total_tx} end) end # Safely encode chart data to JSON, returning empty dataset on error defp safe_encode_chart_data(data) do case Jason.encode(data) do {:ok, encoded} -> encoded {:error, reason} -> Logger.error("Failed to encode chart data: #{inspect(reason)}") Jason.encode!(%{datasets: []}) end end end