defmodule ToweropsWeb.GraphLive.Show do @moduledoc false use ToweropsWeb, :live_view alias Towerops.Agents alias Towerops.Devices alias Towerops.Monitoring alias Towerops.Numeric alias Towerops.Snmp require Logger # Maximum live data points (5 minutes at 1 second intervals) @max_live_points 300 @impl true def mount(_params, _session, socket) do {:ok, socket} end @impl true def handle_params(%{"id" => device_id, "check_id" => check_id} = params, _, socket) do organization = socket.assigns.current_scope.organization with {:ok, device} <- find_device(device_id), :ok <- verify_device_access(device, organization), {:ok, check} <- find_check(check_id), :ok <- verify_check_device(check, device_id) do socket = initialize_check_graph_view(socket, device, check, params) {:noreply, socket} else {:error, :not_found} -> {:noreply, socket |> put_flash(:error, t("Device not found")) |> push_navigate(to: ~p"/devices")} {:error, :check_not_found} -> {:noreply, socket |> put_flash(:error, t("Check not found")) |> push_navigate(to: ~p"/devices/#{device_id}?tab=checks")} {:error, :check_device_mismatch} -> {:noreply, socket |> put_flash(:error, t("Check does not belong to this device")) |> push_navigate(to: ~p"/devices/#{device_id}?tab=checks")} {:error, :access_denied} -> {:noreply, socket |> put_flash(:error, t("You don't have access to this device")) |> push_navigate(to: ~p"/devices")} end end @impl true def handle_params(%{"id" => device_id, "sensor_type" => sensor_type} = params, _, socket) do organization = socket.assigns.current_scope.organization with {:ok, device} <- find_device(device_id), :ok <- verify_device_access(device, organization) do socket = initialize_graph_view(socket, device_id, sensor_type, params) {:noreply, socket} else {:error, :not_found} -> {:noreply, socket |> put_flash(:error, t("Device not found")) |> push_navigate(to: ~p"/devices")} {:error, :access_denied} -> {:noreply, socket |> put_flash(:error, t("You don't have access to this device")) |> push_navigate(to: ~p"/devices")} end end defp find_device(device_id) do case Devices.get_device(device_id) do nil -> {:error, :not_found} device -> {:ok, device} end end defp verify_device_access(device, organization) do if device.organization_id == organization.id do :ok else {:error, :access_denied} end end defp find_check(check_id) do case Monitoring.get_check(check_id) do nil -> {:error, :check_not_found} check -> {:ok, check} end end defp verify_check_device(check, device_id) do if check.device_id == device_id do :ok else {:error, :check_device_mismatch} end end defp initialize_check_graph_view(socket, device, check, params) do range = Map.get(params, "range", "24h") # Load graph data for the check {from_time, to_time} = get_time_range_for_graph(range) graph_data_points = Monitoring.get_check_graph_data(check.id, from_time, to_time) # Convert to chart format chart_data = if Enum.empty?(graph_data_points) do nil else dataset = %{ label: check.name, data: Enum.map(graph_data_points, fn point -> %{ x: DateTime.to_unix(point.timestamp, :millisecond), y: Float.round(point.value, 1) } end) } Jason.encode!(%{datasets: [dataset]}) end # Get unit and title based on check type and config {unit, auto_scale} = get_check_unit_and_scale(check) {max_value, min_value} = calculate_chart_stats(chart_data) socket |> assign(:device, device) |> assign(:device_id, device.id) |> assign(:check, check) |> assign(:check_id, check.id) |> assign(:range, range) |> assign(:page_title, "#{device.name} - #{check.name}") |> assign(:chart_title, check.name) |> assign(:chart_data, chart_data) |> assign(:unit, unit) |> assign(:auto_scale, auto_scale) |> assign(:show_zero_line, false) |> assign(:max_value, max_value) |> assign(:min_value, min_value) |> assign(:is_live_mode, false) |> assign(:has_agent_assignment, false) end defp get_time_range_for_graph("1h"), do: {DateTime.add(DateTime.utc_now(), -1, :hour), DateTime.utc_now()} defp get_time_range_for_graph("6h"), do: {DateTime.add(DateTime.utc_now(), -6, :hour), DateTime.utc_now()} defp get_time_range_for_graph("12h"), do: {DateTime.add(DateTime.utc_now(), -12, :hour), DateTime.utc_now()} defp get_time_range_for_graph("24h"), do: {DateTime.add(DateTime.utc_now(), -24, :hour), DateTime.utc_now()} defp get_time_range_for_graph("7d"), do: {DateTime.add(DateTime.utc_now(), -7, :day), DateTime.utc_now()} defp get_time_range_for_graph("30d"), do: {DateTime.add(DateTime.utc_now(), -30, :day), DateTime.utc_now()} defp get_time_range_for_graph(_), do: {DateTime.add(DateTime.utc_now(), -24, :hour), DateTime.utc_now()} defp get_check_unit_and_scale(%{check_type: "snmp_sensor", config: config}) do # Get unit from sensor config unit = config["sensor_unit"] || "" {unit, true} end defp get_check_unit_and_scale(%{check_type: "snmp_processor"}), do: {"%", false} defp get_check_unit_and_scale(%{check_type: "snmp_storage"}), do: {"%", false} defp get_check_unit_and_scale(%{check_type: "http"}), do: {"ms", true} defp get_check_unit_and_scale(%{check_type: "tcp"}), do: {"ms", true} defp get_check_unit_and_scale(%{check_type: "dns"}), do: {"ms", true} defp get_check_unit_and_scale(_), do: {"", true} defp initialize_graph_view(socket, device_id, sensor_type, params) do maybe_subscribe_to_device(socket, device_id) range = Map.get(params, "range", "24h") interface_id = Map.get(params, "interface_id") sensor_id = Map.get(params, "sensor_id") storage_id = Map.get(params, "storage_id") # Check if device has agent assignment has_agent = Agents.get_device_assignment(device_id) != nil socket |> assign(:device_id, device_id) |> assign(:sensor_type, sensor_type) |> assign(:range, range) |> assign(:interface_id, interface_id) |> assign(:sensor_id, sensor_id) |> assign(:storage_id, storage_id) |> assign(:is_live_mode, range == "live") |> assign(:has_agent_assignment, has_agent) |> maybe_start_live_polling() |> load_graph_data() end defp maybe_subscribe_to_device(socket, device_id) do _ = if connected?(socket) do Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device_id}") end :ok end @impl true def handle_event("change_range", %{"range" => range}, socket) do # Check if this is a check-based graph or sensor-based graph if Map.has_key?(socket.assigns, :check_id) do # Check-based graph params = %{"range" => range, "check_id" => socket.assigns.check_id} {:noreply, push_patch(socket, to: ~p"/devices/#{socket.assigns.device_id}/graph/check?#{params}" )} else # Sensor-based graph (existing behavior) params = build_graph_params(socket.assigns, range) {:noreply, push_patch(socket, to: ~p"/devices/#{socket.assigns.device_id}/graph/#{socket.assigns.sensor_type}?#{params}" )} end end defp build_graph_params(assigns, range) do base_params = %{"range" => range} base_params = if assigns.interface_id do Map.put(base_params, "interface_id", assigns.interface_id) else base_params end base_params = if assigns.sensor_id do Map.put(base_params, "sensor_id", assigns.sensor_id) else base_params end if assigns.storage_id do Map.put(base_params, "storage_id", assigns.storage_id) else base_params end end # Private functions defp load_graph_data(socket) do if socket.assigns.is_live_mode do load_live_mode_graph_data(socket) else # Historical mode - existing logic device_id = socket.assigns.device_id sensor_type = socket.assigns.sensor_type range = socket.assigns.range device = Devices.get_device!(device_id) {chart_data, title_suffix} = load_chart_data_for_type(socket.assigns, range) {title, unit, auto_scale} = get_chart_config(sensor_type) chart_title = if title_suffix, do: "#{title} - #{title_suffix}", else: title show_zero_line = sensor_type == "traffic" # Calculate max/min values from chart data {max_value, min_value} = calculate_chart_stats(chart_data) socket |> assign(:device, device) |> assign(:page_title, "#{device.name} - #{chart_title}") |> assign(:chart_title, chart_title) |> assign(:chart_data, chart_data) |> assign(:unit, unit) |> assign(:auto_scale, auto_scale) |> assign(:show_zero_line, show_zero_line) |> assign(:max_value, max_value) |> assign(:min_value, min_value) end end defp load_live_mode_graph_data(socket) do device_id = socket.assigns.device_id sensor_type = socket.assigns.sensor_type device = Devices.get_device!(device_id) {title, unit, auto_scale} = get_chart_config(sensor_type) title_suffix = get_title_suffix(socket.assigns) chart_title = if title_suffix, do: "#{title} - #{title_suffix}", else: title socket |> assign(:device, device) |> assign(:page_title, "#{device.name} - #{chart_title} (Live)") |> assign(:chart_title, chart_title) |> assign(:chart_data, Jason.encode!(%{datasets: []})) |> assign(:unit, unit) |> assign(:auto_scale, auto_scale) |> assign(:show_zero_line, sensor_type == "traffic") |> assign(:max_value, nil) |> assign(:min_value, nil) end defp get_title_suffix(assigns) do cond do assigns[:sensor_id] -> get_sensor_name(assigns[:sensor_id]) assigns[:interface_id] && assigns[:sensor_type] == "traffic" -> get_interface_name(assigns[:interface_id]) assigns[:storage_id] && assigns[:sensor_type] == "storage_volume" -> get_storage_name(assigns[:storage_id]) true -> nil end end defp load_chart_data_for_type(%{sensor_type: "latency", device_id: device_id}, range) do {load_latency_chart_data(device_id, range), nil} end defp load_chart_data_for_type(%{sensor_type: "traffic", interface_id: interface_id}, range) when not is_nil(interface_id) do {load_interface_traffic_chart_data(interface_id, range), get_interface_name(interface_id)} end defp load_chart_data_for_type(%{sensor_type: "traffic", device_id: device_id}, range) do {load_traffic_chart_data(device_id, range), nil} end defp load_chart_data_for_type(%{sensor_type: "storage_volume", storage_id: storage_id}, range) when not is_nil(storage_id) do {load_storage_chart_data(storage_id, range), get_storage_name(storage_id)} end defp load_chart_data_for_type(%{sensor_id: sensor_id}, range) when not is_nil(sensor_id) do {load_single_sensor_chart_data(sensor_id, range), get_sensor_name(sensor_id)} end defp load_chart_data_for_type(%{sensor_type: sensor_type, device_id: device_id}, range) do sensor_types = get_sensor_types_for_chart(sensor_type) {load_sensor_chart_data(device_id, sensor_types, range), nil} end defp get_sensor_types_for_chart("processors"), do: ["cpu_load"] defp get_sensor_types_for_chart("memory"), do: ["memory_usage"] defp get_sensor_types_for_chart("storage"), do: ["disk_usage"] defp get_sensor_types_for_chart("temperature"), do: ["temperature", "cpu_temperature", "celsius"] defp get_sensor_types_for_chart("voltage"), do: ["voltage", "volts"] defp get_sensor_types_for_chart("traffic"), do: nil # For unknown sensor types, use the type directly (supports count, pppoe_sessions, etc.) defp get_sensor_types_for_chart(sensor_type), do: [sensor_type] defp get_chart_config("latency"), do: {"Ping Latency", "ms", true} defp get_chart_config("processors"), do: {"Processor Usage", "%", false} defp get_chart_config("memory"), do: {"Memory Usage", "%", false} defp get_chart_config("storage"), do: {"Storage Usage", "%", false} defp get_chart_config("storage_volume"), do: {"Storage Usage", "%", false} defp get_chart_config("temperature"), do: {"Temperature", "°C", true} defp get_chart_config("voltage"), do: {"Voltage", "V", true} defp get_chart_config("traffic"), do: {"Overall Traffic", "bps", true} defp get_chart_config("count"), do: {"Count", "", true} defp get_chart_config("pppoe_sessions"), do: {"PPPoE Sessions", "", true} defp get_chart_config("connections"), do: {"Connections", "", true} defp get_chart_config("wireless"), do: {"Wireless Metrics", "", true} # Humanize unknown sensor types for title defp get_chart_config(sensor_type), do: {humanize_sensor_type(sensor_type), "", true} defp humanize_sensor_type(sensor_type) do sensor_type |> String.replace("_", " ") |> String.split() |> Enum.map_join(" ", &String.capitalize/1) end defp load_sensor_chart_data(device_id, sensor_types, range) do case Snmp.get_device_with_associations(device_id) do nil -> nil device -> build_sensor_chart_json(device, sensor_types, range) end end defp load_single_sensor_chart_data(sensor_id, range) do case Snmp.get_sensor(sensor_id) do nil -> nil sensor -> since = get_datetime_from_range(range) limit = get_limit_for_range(range) dataset = sensor_to_dataset(sensor, since, limit) # Return nil if no data points (empty readings) if Enum.empty?(dataset.data) do nil else Jason.encode!(%{datasets: [dataset]}) end end end defp get_sensor_name(sensor_id) do case Snmp.get_sensor(sensor_id) do nil -> nil sensor -> sensor.sensor_descr end end defp load_storage_chart_data(storage_id, range) do case Snmp.get_storage(storage_id) do nil -> nil storage -> since = get_datetime_from_range(range) limit = get_limit_for_range(range) dataset = storage_to_dataset(storage, since, limit) Jason.encode!(%{datasets: [dataset]}) end end defp get_storage_name(storage_id) do case Snmp.get_storage(storage_id) do nil -> nil storage -> storage.description || "Storage #{storage.storage_index}" end end defp storage_to_dataset(storage, since, limit) do readings = storage.id |> Snmp.get_storage_readings(since: since, limit: limit) |> Enum.reverse() %{ label: storage.description || "Storage #{storage.storage_index}", data: Enum.map(readings, &storage_reading_to_chart_point/1) } end defp storage_reading_to_chart_point(reading) do %{ x: DateTime.to_unix(reading.checked_at, :millisecond), y: if(reading.usage_percent, do: Float.round(reading.usage_percent, 1)) } end defp build_sensor_chart_json(device, sensor_types, range) 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 since = get_datetime_from_range(range) limit = get_limit_for_range(range) datasets = Enum.map(sensors, &sensor_to_dataset(&1, since, limit)) Jason.encode!(%{datasets: datasets}) end end defp sensor_to_dataset(sensor, since, limit) do readings = sensor.id |> Snmp.get_sensor_readings(since: since, limit: limit) |> Enum.reverse() %{ label: sensor.sensor_descr, data: Enum.map(readings, &reading_to_chart_point/1) } end defp reading_to_chart_point(reading) do %{ x: DateTime.to_unix(reading.checked_at, :millisecond), y: if(reading.value, do: Float.round(reading.value, 1)) } end defp get_datetime_from_range("1h"), do: DateTime.add(DateTime.utc_now(), -1, :hour) defp get_datetime_from_range("6h"), do: DateTime.add(DateTime.utc_now(), -6, :hour) defp get_datetime_from_range("12h"), do: DateTime.add(DateTime.utc_now(), -12, :hour) defp get_datetime_from_range("24h"), do: DateTime.add(DateTime.utc_now(), -24, :hour) defp get_datetime_from_range("7d"), do: DateTime.add(DateTime.utc_now(), -7, :day) defp get_datetime_from_range("30d"), do: DateTime.add(DateTime.utc_now(), -30, :day) defp get_datetime_from_range(_), do: DateTime.add(DateTime.utc_now(), -24, :hour) defp get_limit_for_range("1h"), do: 360 defp get_limit_for_range("6h"), do: 720 defp get_limit_for_range("12h"), do: 1440 defp get_limit_for_range("24h"), do: 2880 defp get_limit_for_range("7d"), do: 10_080 defp get_limit_for_range("30d"), do: 43_200 defp get_limit_for_range(_), do: 2880 defp load_traffic_chart_data(device_id, range) do case Snmp.get_device_with_associations(device_id) do nil -> nil device -> build_traffic_chart_json(device, range) end end defp load_interface_traffic_chart_data(interface_id, range) do case Snmp.get_interface(interface_id) do nil -> nil interface -> build_interface_traffic_chart_json(interface, range) end end defp get_interface_name(interface_id) do case Snmp.get_interface(interface_id) do nil -> "Unknown Interface" interface -> interface.if_name || interface.if_descr || "Interface #{interface.if_index}" end end defp load_latency_chart_data(device_id, range) do since = get_datetime_from_range(range) limit = get_limit_for_range(range) checks = device_id |> Monitoring.get_latency_data(since: since, limit: limit) |> Enum.reverse() if Enum.empty?(checks) do nil else dataset = %{ label: "Ping Latency", data: Enum.map(checks, &latency_check_to_chart_point/1) } Jason.encode!(%{datasets: [dataset]}) end end defp latency_check_to_chart_point(check) do %{ x: DateTime.to_unix(check.checked_at, :millisecond), y: Float.round(check.response_time_ms, 1) } end # Calculate max and min values from chart data defp calculate_chart_stats(nil), do: {nil, nil} defp calculate_chart_stats(chart_data) when is_binary(chart_data) do case Jason.decode(chart_data) do {:ok, %{"datasets" => datasets}} -> extract_min_max_from_datasets(datasets) _ -> {nil, nil} end end defp extract_min_max_from_datasets(datasets) do values = Enum.flat_map(datasets, &extract_values_from_dataset/1) case values do [] -> {nil, nil} _ -> {Enum.max(values), Enum.min(values)} end end defp extract_values_from_dataset(%{"data" => data}) when is_list(data) do Enum.flat_map(data, &extract_y_value/1) end defp extract_values_from_dataset(_), do: [] defp extract_y_value(%{"y" => y}) when is_number(y), do: [y] defp extract_y_value(_), do: [] defp build_traffic_chart_json(device, range) do if Enum.empty?(device.interfaces), do: nil, else: build_traffic_datasets(device, range) end defp build_traffic_datasets(device, range) do since = get_datetime_from_range(range) limit = get_limit_for_range(range) active_interfaces = Enum.filter(device.interfaces, &(&1.if_oper_status == "up")) if Enum.empty?(active_interfaces), do: nil, else: encode_interface_datasets(active_interfaces, since, limit) end defp encode_interface_datasets(interfaces, since, limit) do datasets = build_per_interface_datasets(interfaces, since, limit) if Enum.empty?(datasets), do: nil, else: Jason.encode!(%{datasets: datasets}) end defp build_per_interface_datasets(interfaces, since, limit) do Enum.flat_map(interfaces, fn interface -> stats = interface.id |> Snmp.get_interface_stats(since: since, limit: limit) |> Enum.reverse() if Enum.empty?(stats) do [] else interface_name = interface.if_name || interface.if_descr in_data = calculate_interface_traffic_data(stats, :inbound) out_data = calculate_interface_traffic_data(stats, :outbound) [ %{label: "#{interface_name} Out", data: out_data}, %{label: "#{interface_name} In", data: in_data} ] end end) end defp build_interface_traffic_chart_json(interface, range) do since = get_datetime_from_range(range) limit = get_limit_for_range(range) stats = interface.id |> Snmp.get_interface_stats(since: since, limit: limit) |> Enum.reverse() if Enum.empty?(stats) do nil else in_data = calculate_interface_traffic_data(stats, :inbound) out_data = calculate_interface_traffic_data(stats, :outbound) datasets = [ %{label: "Outbound", data: out_data}, %{label: "Inbound", data: in_data} ] Jason.encode!(%{datasets: datasets}) end end defp calculate_interface_traffic_data(stats, :inbound) do stats |> Enum.chunk_every(2, 1, :discard) |> Enum.map(fn [stat1, stat2] -> # Skip if octets are nil if is_nil(stat1.if_in_octets) or is_nil(stat2.if_in_octets) do nil else time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1) bps = ((stat2.if_in_octets - stat1.if_in_octets) * 8 / time_diff) |> max(0) |> :erlang.float() |> Float.round(2) %{ x: DateTime.to_unix(stat2.checked_at, :millisecond), y: -bps } end end) |> Enum.reject(&is_nil/1) end defp calculate_interface_traffic_data(stats, :outbound) do stats |> Enum.chunk_every(2, 1, :discard) |> Enum.map(fn [stat1, stat2] -> # Skip if octets are nil if is_nil(stat1.if_out_octets) or is_nil(stat2.if_out_octets) do nil else time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1) bps = ((stat2.if_out_octets - stat1.if_out_octets) * 8 / time_diff) |> max(0) |> :erlang.float() |> Float.round(2) %{ x: DateTime.to_unix(stat2.checked_at, :millisecond), y: bps } end end) |> Enum.reject(&is_nil/1) end # Format value with unit for display defp format_value(value, "bps") do abs_value = abs(value) cond do abs_value >= 1_000_000_000 -> "#{Float.round(abs_value / 1_000_000_000, 2)} Gbps" abs_value >= 1_000_000 -> "#{Float.round(abs_value / 1_000_000, 2)} Mbps" abs_value >= 1_000 -> "#{Float.round(abs_value / 1_000, 2)} Kbps" true -> "#{Float.round(abs_value, 2)} bps" end end # For counts (empty unit), format as integer defp format_value(value, "") do "#{round(value)}" end # For other units, format with one decimal place defp format_value(value, unit) do "#{Float.round(value, 1)}#{unit}" end # Live polling functions # Start/stop live polling based on mode defp maybe_start_live_polling(socket) do if socket.assigns[:is_live_mode] && connected?(socket) do cancel_live_polling_timer(socket) socket |> assign(:live_data_buffer, %{}) |> assign(:live_data_points, 0) |> schedule_live_poll() else cancel_live_polling_timer(socket) end end # Schedule next poll in 1 second defp schedule_live_poll(socket) do timer_ref = Process.send_after(self(), :live_poll, 1_000) assign(socket, :live_poll_timer, timer_ref) end # Cancel existing timer defp cancel_live_polling_timer(socket) do case Map.get(socket.assigns, :live_poll_timer) do nil -> socket timer_ref -> Process.cancel_timer(timer_ref) assign(socket, :live_poll_timer, nil) end end # Handle device status changes from PubSub @impl true def handle_info({:device_status_changed, _device_id, _new_status, _response_time}, socket) do # Reload device to get updated status device = Devices.get_device(socket.assigns.device_id) {:noreply, assign(socket, :device, device)} end # Handle sensor updates from PubSub @impl true def handle_info({:state_sensors_updated, _device_id}, socket) do {:noreply, load_graph_data(socket)} end # Handle neighbor updates from PubSub (ignored - not relevant to graphs) @impl true def handle_info({:neighbors_updated, _device_id}, socket) do {:noreply, socket} end # Handle live poll event @impl true def handle_info(:live_poll, socket) do if socket.assigns[:is_live_mode] do socket = perform_live_poll(socket) {:noreply, schedule_live_poll(socket)} else {:noreply, socket} end end # Catch-all for unexpected messages @impl true def handle_info(_msg, socket), do: {:noreply, socket} # Perform SNMP polling and update buffer defp perform_live_poll(socket) do device_id = socket.assigns.device_id # Verify device still exists and access is valid case Devices.get_device(device_id) do nil -> socket |> put_flash(:error, t("Device no longer exists")) |> push_navigate(to: ~p"/devices") device -> organization = socket.assigns.current_scope.organization if device.organization_id == organization.id do poll_and_update_buffer(socket, device) else socket |> put_flash(:error, t("Access to device revoked")) |> push_navigate(to: ~p"/devices") end end end # Poll sensors and update buffer defp poll_and_update_buffer(socket, device) do now = DateTime.utc_now() timestamp_ms = DateTime.to_unix(now, :millisecond) # Check effective agent (device → site → org cascade) agent_token_id = Agents.get_effective_agent_token(device) Logger.info("Live poll for device #{device.name}: agent_token_id=#{inspect(agent_token_id)}") new_points = if agent_token_id do # Request agent to poll NOW and wait for result Logger.info("Requesting live poll from agent") request_agent_live_poll_and_wait(device.id, agent_token_id, socket.assigns) else # Do direct SNMP polling for specific sensors only Logger.info("Using direct SNMP for live polling") client_opts = build_snmp_client_opts(device) poll_sensors_for_live_mode(socket.assigns, client_opts, timestamp_ms) end Logger.info("Live poll fetched #{length(new_points)} data points: #{inspect(new_points)}") # Update buffer with new points (rolling window) {updated_buffer, point_count} = update_live_data_buffer( socket.assigns.live_data_buffer, socket.assigns.live_data_points, new_points ) # Push update to client socket |> assign(:live_data_buffer, updated_buffer) |> assign(:live_data_points, point_count) |> push_event("live_data_update", %{ timestamp: timestamp_ms, points: new_points }) end # Request live poll from agent and wait for response defp request_agent_live_poll_and_wait(device_id, agent_token_id, assigns) do # Generate unique reply topic for this request reply_topic = "live_poll_reply:#{:erlang.unique_integer([:positive])}" # Subscribe to reply topic :ok = Phoenix.PubSub.subscribe(Towerops.PubSub, reply_topic) # Get sensors we're interested in sensors = get_sensors_for_live_mode(assigns) sensor_oids = Enum.map(sensors, & &1.sensor_oid) # Broadcast live poll request with reply topic and sensor OIDs Phoenix.PubSub.broadcast( Towerops.PubSub, "agent:#{agent_token_id}:live_poll", {:live_poll_requested, device_id, sensor_oids, reply_topic} ) # Wait for response with timeout receive do {:live_poll_result, oid_values} -> # Process the OID values into data points process_live_poll_oids(sensors, oid_values) after 3000 -> Logger.warning("Live poll timeout waiting for agent response") [] end end # Convert OID values from agent into data points defp process_live_poll_oids(sensors, oid_values) do timestamp_ms = DateTime.to_unix(DateTime.utc_now(), :millisecond) sensors |> Enum.map(&build_live_poll_data_point(&1, oid_values, timestamp_ms)) |> Enum.reject(&is_nil/1) end defp build_live_poll_data_point(sensor, oid_values, timestamp_ms) do with value when not is_nil(value) <- Map.get(oid_values, sensor.sensor_oid), parsed_value when not is_nil(parsed_value) <- parse_sensor_value(value) do final_value = parsed_value / sensor.sensor_divisor %{ sensor_id: sensor.id, label: sensor.sensor_descr, value: Float.round(final_value, 1), timestamp: timestamp_ms } else _ -> nil end end defp parse_sensor_value(value), do: Numeric.parse_float(value) # Build SNMP client options defp build_snmp_client_opts(device) do snmp_config = Devices.get_snmp_config(device) [ ip: device.ip_address, community: snmp_config.community, version: snmp_config.version, port: device.snmp_port || 161, # Increased for live polling - devices may be slow to respond timeout: 3000 ] end # Poll sensors in parallel defp poll_sensors_for_live_mode(assigns, client_opts, timestamp_ms) do sensors = get_sensors_for_live_mode(assigns) sensors |> Task.async_stream( fn sensor -> case poll_single_sensor(sensor, client_opts) do {:ok, value} -> %{ sensor_id: sensor.id, label: sensor.sensor_descr, value: Float.round(value, 1), timestamp: timestamp_ms } {:error, _reason} -> # Skip failed polls - chart will show gap nil end end, max_concurrency: 10, # Give SNMP client (3000ms timeout) enough time plus overhead timeout: 4000, on_timeout: :kill_task ) |> Enum.reduce([], fn {:ok, nil}, acc -> acc {:ok, point}, acc -> [point | acc] {:exit, _reason}, acc -> acc end) end # Get sensors based on graph type defp get_sensors_for_live_mode(%{sensor_id: sensor_id}) when not is_nil(sensor_id) do case Snmp.get_sensor(sensor_id) do nil -> [] sensor -> [sensor] end end defp get_sensors_for_live_mode(%{sensor_type: sensor_type, device_id: device_id}) do sensor_types = get_sensor_types_for_chart(sensor_type) device = Snmp.get_device_with_associations(device_id) device.sensors |> Enum.filter(&(&1.sensor_type in sensor_types)) |> Enum.sort_by(& &1.sensor_index) end # Poll single sensor via SNMP defp poll_single_sensor(sensor, client_opts) do case Snmp.Client.get(client_opts, sensor.sensor_oid) do {:ok, raw_value} when is_number(raw_value) -> value = raw_value / sensor.sensor_divisor {:ok, value} {:ok, _non_numeric} -> {:error, :non_numeric} {:error, reason} -> {:error, reason} end end # Update live data buffer with rolling window defp update_live_data_buffer(buffer, current_count, new_points) do updated_buffer = Enum.reduce(new_points, buffer, fn point, acc -> sensor_buffer = Map.get(acc, point.sensor_id, []) # Add new point at the front new_buffer = [ %{x: point.timestamp, y: point.value} | sensor_buffer ] # Trim to max size trimmed_buffer = if current_count >= @max_live_points do Enum.take(new_buffer, @max_live_points) else new_buffer end Map.put(acc, point.sensor_id, trimmed_buffer) end) new_count = min(current_count + 1, @max_live_points) {updated_buffer, new_count} end # Cleanup on LiveView termination @impl true def terminate(_reason, socket) do cancel_live_polling_timer(socket) :ok end end