VALLONGOL/S1005403_RisCC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sistemata la visualizzazione dell'antenna durante la connessione al server

Browse Source
This commit is contained in:
VALLONGOL 2025-11-07 13:16:00 +01:00
parent 7c7bbe57ba
commit 05ae71420b
7 changed files with 162 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
logger_prefs.json
View File

@ -1,9 +1,10 @@
{
"saved_levels": {
"target_simulator.gui.payload_router": "INFO",
"target_simulator.gui.payload_router": "DEBUG",
"target_simulator.analysis.simulation_state_hub": "INFO",
"target_simulator.gui.main_view": "INFO",
"target_simulator.core.sfp_transport": "INFO"
"target_simulator.core.sfp_transport": "INFO",
"target_simulator.gui.ppi_display": "DEBUG"
},
"last_selected": "target_simulator.core.sfp_transport"
"last_selected": "target_simulator.gui.payload_router"
}

19
target_simulator/analysis/simulation_state_hub.py
View File

@ -270,6 +270,16 @@ class SimulationStateHub:
with self._lock:
self._antenna_azimuth_deg = az
self._antenna_azimuth_ts = ts
try:
# Debug log to aid diagnosis when antenna updates arrive
logger.debug(
"[SimulationStateHub] set_antenna_azimuth: az=%.3f ts=%s hub_id=%s",
az,
ts,
id(self),
)
except Exception:
pass
def get_antenna_azimuth(self) -> Tuple[Optional[float], Optional[float]]:
"""
@ -278,6 +288,15 @@ class SimulationStateHub:
If not available, returns (None, None).
"""
with self._lock:
try:
logger.debug(
"[SimulationStateHub] get_antenna_azimuth -> az=%s ts=%s hub_id=%s",
self._antenna_azimuth_deg,
self._antenna_azimuth_ts,
id(self),
)
except Exception:
pass
return (self._antenna_azimuth_deg, self._antenna_azimuth_ts)
# Backwards-compatible aliases for existing callers. These delegate to the

12
target_simulator/gui/main_view.py
View File

@ -642,6 +642,15 @@ class MainView(tk.Tk):
# Update antenna sweep line
if self.ppi_widget.animate_antenna_var.get():
az_deg, _ = self.simulation_hub.get_antenna_azimuth()
try:
# Debug: record the value read from the hub and hub identity
self.logger.debug(
"MainView: read antenna az from hub -> az=%s hub_id=%s",
az_deg,
id(self.simulation_hub),
)
except Exception:
pass
if az_deg is not None:
self.ppi_widget.render_antenna_line(az_deg)
else:
@ -679,6 +688,9 @@ class MainView(tk.Tk):
self._update_simulation_progress_display()
if hasattr(self, "ppi_widget") and self.ppi_widget.canvas:
self.ppi_widget.canvas.draw_idle()
self.after(GUI_REFRESH_RATE_MS, self._gui_refresh_loop)
def _refresh_analysis_list(self):

90
target_simulator/gui/payload_router.py
View File

@ -336,7 +336,9 @@ class DebugPayloadRouter:
# --- Propagate antenna azimuth to hub ---
if self._hub:
# Get platform heading and relative antenna sweep
heading_rad = scenario_dict.get("true_heading", scenario_dict.get("platform_azimuth"))
heading_rad = scenario_dict.get(
"true_heading", scenario_dict.get("platform_azimuth")
)
sweep_rad = scenario_dict.get("ant_nav_az")
total_az_rad = None
@ -372,61 +374,65 @@ class DebugPayloadRouter:
except Exception:
return
# The JSON may follow the same structure as the RIS debug JSON
# we generate elsewhere: {"scenario": {...}, "targets": [...]}
def _find_scenario_field(dct, key):
# Helper to find fields in varied JSON structures
def _find_val(key_candidates, dct):
if not isinstance(dct, dict):
return None
# Direct scenario container
sc = dct.get("scenario") or dct.get("sc") or dct
if isinstance(sc, dict) and key in sc:
return sc.get(key)
# Top-level key
if key in dct:
return dct.get(key)
# Nested search
for v in dct.values():
if isinstance(v, dict) and key in v:
return v.get(key)
# Check top-level
for key in key_candidates:
if key in dct:
return dct[key]
# Check scenario sub-dict if present
sc = dct.get("scenario") or dct.get("sc")
if isinstance(sc, dict):
for key in key_candidates:
if key in sc:
return sc[key]
return None
# Find platform heading and relative antenna sweep
heading_val = _find_scenario_field(obj, "true_heading") or _find_scenario_field(
obj, "platform_azimuth"
# Find platform heading and relative antenna sweep using multiple potential keys
heading_val = _find_val(
["true_heading", "platform_azimuth", "heading"], obj
)
sweep_val = _find_val(
["ant_nav_az", "antenna_azimuth", "sweep_azimuth"], obj
)
sweep_val = _find_scenario_field(obj, "ant_nav_az")
total_az_val = None
if heading_val is not None:
total_az_val = float(heading_val)
if sweep_val is not None:
total_az_val += float(sweep_val)
if total_az_val is not None:
try:
# Values may be in radians or degrees; if small absolute
# value (< 2*pi) assume radians and convert.
val = float(total_az_val)
if abs(val) <= (2 * math.pi + 0.01):
az_deg = math.degrees(val)
else:
az_deg = val
h_rad = float(heading_val)
# If heading is in degrees (> 2*pi), convert to radians
if abs(h_rad) > (2 * math.pi + 0.01):
h_rad = math.radians(h_rad)
total_az_rad = h_rad
if sweep_val is not None:
s_rad = float(sweep_val)
# If sweep is in degrees, convert
if abs(s_rad) > (2 * math.pi + 0.01):
s_rad = math.radians(s_rad)
total_az_rad += s_rad
az_deg = math.degrees(total_az_rad)
# LOGGARE IL SUCCESSO PER DEBUG
self._logger.debug(
f"Found azimuth info in JSON: heading={heading_val}, sweep={sweep_val} -> total_deg={az_deg}"
)
self._hub.set_antenna_azimuth(az_deg, timestamp=time.monotonic())
except Exception:
except Exception as e:
self._logger.debug(f"Error processing azimuth values: {e}")
pass
else:
pass
# self._logger.debug("No heading info found in JSON payload")
# Optionally capture elevation for future UI use
plat_el = _find_scenario_field(obj, "ant_nav_el") or _find_scenario_field(
obj, "platform_elevation"
)
if plat_el is not None:
try:
_ = float(plat_el)
# For now we don't store elevation in the hub; leave as TODO.
except Exception:
pass
# plat_el = _find_val(["ant_nav_el", "platform_elevation"], obj)
# if plat_el is not None:
# pass
except Exception:
except Exception as e:
self._logger.exception("Error handling JSON payload")
def get_and_clear_latest_payloads(self) -> Dict[str, Any]:

4
target_simulator/gui/ppi_adapter.py
View File

@ -56,7 +56,9 @@ def build_display_data(
sim_target = Target(target_id=tid, trajectory=[])
setattr(sim_target, "_pos_x_ft", rel_x_ft)
setattr(sim_target, "_pos_y_ft", rel_y_ft)
setattr(sim_target, "_pos_z_ft", z_ft) # Altitude is handled relative to ground
setattr(
sim_target, "_pos_z_ft", z_ft
) # Altitude is handled relative to ground
sim_target.current_velocity_fps = vel_fps
sim_target.current_vertical_velocity_fps = vert_vel_fps
sim_target._update_current_polar_coords()

66
target_simulator/gui/ppi_display.py
View File

@ -236,7 +236,9 @@ class PPIDisplay(ttk.Frame):
(self._path_plot,) = self.ax.plot([], [], "g--", linewidth=1.5, zorder=3)
(self._start_plot,) = self.ax.plot([], [], "go", markersize=8, zorder=3)
(self._waypoints_plot,) = self.ax.plot([], [], "y+", markersize=10, mew=2, zorder=3)
(self._waypoints_plot,) = self.ax.plot(
[], [], "y+", markersize=10, mew=2, zorder=3
)
self.preview_artists = [self._path_plot, self._start_plot, self._waypoints_plot]
limit_rad = np.deg2rad(self.scan_limit_deg)
@ -413,7 +415,13 @@ class PPIDisplay(ttk.Frame):
r_nm = target.current_range_nm
theta_rad_plot = np.deg2rad(target.current_azimuth_deg)
(dot,) = self.ax.plot(
theta_rad_plot, r_nm, "o", markersize=6, color=color, alpha=0.6, zorder=5
theta_rad_plot,
r_nm,
"o",
markersize=6,
color=color,
alpha=0.6,
zorder=5,
)
artist_list.append(dot)
(x_mark,) = self.ax.plot(
@ -424,7 +432,7 @@ class PPIDisplay(ttk.Frame):
markersize=8,
markeredgewidth=0.9,
linestyle="",
zorder=6
zorder=6,
)
label_artist_list.append(x_mark)
except Exception:
@ -489,7 +497,13 @@ class PPIDisplay(ttk.Frame):
if len(trail) > 1:
thetas, rs = zip(*trail)
(line,) = self.ax.plot(
thetas, rs, color=color, linestyle="-", linewidth=0.8, alpha=0.7, zorder=3
thetas,
rs,
color=color,
linestyle="-",
linewidth=0.8,
alpha=0.7,
zorder=3,
)
artist_list.append(line)
@ -620,21 +634,53 @@ class PPIDisplay(ttk.Frame):
def render_antenna_line(self, az_deg: Optional[float]):
"""Directly renders the antenna line at a given absolute azimuth."""
# Aggiunto log di debug per tracciare la chiamata
# logger.debug(f"PPIDisplay.render_antenna_line called with az_deg={az_deg}")
try:
# If the artist was not created (unexpected), create it lazily so
# the antenna can still be rendered. This avoids silent failures
# when the initial plot setup didn't complete for some reason.
created_artist = False
if self._antenna_line_artist is None:
return
try:
(self._antenna_line_artist,) = self.ax.plot(
[],
[],
color="lightgray",
linestyle="--",
linewidth=1.2,
alpha=0.85,
zorder=2,
)
created_artist = True
logger.debug(
"PPIDisplay: lazily created _antenna_line_artist for widget id=%s",
id(self),
)
except Exception:
# If creation fails, give up gracefully
return
if az_deg is None or not self.animate_antenna_var.get():
self._antenna_line_artist.set_visible(False)
else:
theta = np.deg2rad(float(az_deg))
# Assicuriamoci che az_deg sia un float valido
az_float = float(az_deg)
theta = np.deg2rad(az_float % 360)
max_r = self.ax.get_ylim()[1]
self._antenna_line_artist.set_data([theta, theta], [0, max_r])
self._antenna_line_artist.set_visible(True)
# This method is called frequently, so we rely on the main loop's
# final draw call to update the canvas, avoiding redundant draws.
# self.canvas.draw_idle()
# logger.debug(f"Antenna line drawn at theta={theta}")
# If we created the artist now, or we made it visible, schedule a
# canvas redraw. Use draw_idle to avoid immediate blocking draws.
try:
if (
created_artist
or (az_deg is not None and self.animate_antenna_var.get())
) and self.canvas:
self.canvas.draw_idle()
except Exception:
pass
except Exception:
# Silently fail to prevent logging floods
pass

13
target_simulator/utils/logger.py
View File

@ -144,11 +144,24 @@ def setup_basic_logging(
_actual_console_handler = logging.StreamHandler()
_actual_console_handler.setFormatter(_base_formatter)
_actual_console_handler.setLevel(logging.DEBUG)
try:
# Also attach console handler directly to the root logger so
# console output appears immediately (helps during development
# and when the Tk polling loop hasn't started yet).
root_logger.addHandler(_actual_console_handler)
except Exception:
pass
queue_putter = QueuePuttingHandler(handler_queue=_global_log_queue)
queue_putter.setLevel(logging.DEBUG)
root_logger.addHandler(queue_putter)
# Emit a small startup message so users running from console see logging is active
try:
root_logger.debug("Logging system initialized (queue-based).")
except Exception:
pass
_logging_system_active = True
_log_processor_after_id = _tk_root_instance_for_processing.after(
GLOBAL_LOG_QUEUE_POLL_INTERVAL_MS, _process_global_log_queue