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

aggiunta il reset dei target via comando sia unico che mediante disattivazione target

Browse Source
This commit is contained in:
VALLONGOL 2025-10-29 14:58:44 +01:00
parent ca13144c1d
commit 20b7fd41ad
5 changed files with 236 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
settings.json
View File

@ -2,8 +2,8 @@
"general": {
"scan_limit": 60,
"max_range": 100,
"geometry": "1599x1089+169+45",
"last_selected_scenario": "scenario2",
"geometry": "1599x1089+587+179",
"last_selected_scenario": null,
"connection": {
"target": {
"type": "sfp",

4
target_simulator/gui/main_view.py
View File

@ -1083,7 +1083,9 @@ class MainView(tk.Tk):
):
real_rate = float(self.simulation_hub.get_real_rate(1.0))
# Prefer packet rate if available (packets/sec vs events/sec)
if self.simulation_hub and hasattr(self.simulation_hub, "get_packet_rate"):
if self.simulation_hub and hasattr(
self.simulation_hub, "get_packet_rate"
):
packet_rate = float(self.simulation_hub.get_packet_rate(1.0))
except Exception:
real_rate = 0.0

10
target_simulator/gui/payload_router.py
View File

@ -100,7 +100,9 @@ class DebugPayloadRouter:
with self._lock:
self._latest_payloads[flow_id] = payload
def _parse_ris_payload_to_targets(self, payload: bytearray) -> Tuple[List[Target], List[int]]:
def _parse_ris_payload_to_targets(
self, payload: bytearray
) -> Tuple[List[Target], List[int]]:
"""
Parse RIS payload and return a tuple:
- list of Target objects for entries with flags != 0 (active targets)
@ -178,11 +180,13 @@ class DebugPayloadRouter:
# If payload included inactive targets (flags==0), clear their stored
# real data so they disappear from the PPI immediately.
try:
for tid in (inactive_ids or []):
for tid in inactive_ids or []:
if hasattr(self._hub, "clear_real_target_data"):
self._hub.clear_real_target_data(tid)
except Exception:
self._logger.debug("Failed to clear inactive target data in hub", exc_info=True)
self._logger.debug(
"Failed to clear inactive target data in hub", exc_info=True
)
# Add real states for active targets
for target in real_targets:

357
target_simulator/gui/sfp_debug_window.py
View File

@ -348,7 +348,18 @@ class SfpDebugWindow(tk.Toplevel):
self.send_probe_btn.pack(side=tk.LEFT, padx=(6, 4))
def _create_target_sender_widgets(self, parent):
grid = ttk.Frame(parent, padding=5)
# Create a horizontal split: left 60% for editable fields, right 40% for
# a notebook containing CMD and JSON quick-action tabs.
paned = ttk.Panedwindow(parent, orient=tk.HORIZONTAL)
paned.pack(fill=tk.X, expand=False)
left = ttk.Frame(paned, padding=5)
right = ttk.Frame(paned, padding=5)
paned.add(left, weight=3)
paned.add(right, weight=2)
# Left side: editable fields laid out in a grid (keeps existing controls)
grid = ttk.Frame(left)
grid.pack(fill=tk.X)
grid.columnconfigure(1, pad=15)
grid.columnconfigure(3, pad=15)
@ -377,70 +388,93 @@ class SfpDebugWindow(tk.Toplevel):
ttk.Spinbox(
grid, from_=0, to=80000, textvariable=self.tgt_alt_var, width=12
).grid(row=1, column=5, sticky=tk.W)
controls_frame = ttk.Frame(grid)
controls_frame.grid(row=1, column=6, sticky="nsew", padx=(20, 0))
ttk.Checkbutton(
controls_frame, text="Active", variable=self.tgt_active_var
).pack(side=tk.LEFT, anchor="w")
ttk.Checkbutton(
controls_frame, text="Traceable", variable=self.tgt_traceable_var
).pack(side=tk.LEFT, anchor="w", padx=5)
ttk.Checkbutton(
controls_frame, text="Restart", variable=self.tgt_restart_var
).pack(side=tk.LEFT, anchor="w", padx=5)
# Send mode toggle: choose between sending a command-line string or a JSON info payload
ttk.Label(controls_frame, text="Send Mode:").pack(side=tk.LEFT, padx=(10, 4))
ttk.Radiobutton(
controls_frame, text="cmd line", variable=self.send_mode_var, value="cmd"
).pack(side=tk.LEFT)
ttk.Radiobutton(
controls_frame, text="json info", variable=self.send_mode_var, value="json"
).pack(side=tk.LEFT, padx=(0, 6))
send_button = ttk.Button(
controls_frame, text="Send Target", command=self._on_send_target
# Flags / options row under the numeric fields (keeps numeric inputs on one line)
flags_frame = ttk.Frame(grid)
# place flags_frame below the spinboxes, spanning the visible columns
flags_frame.grid(row=2, column=0, columnspan=7, sticky="w", pady=(8, 0))
ttk.Checkbutton(flags_frame, text="Active", variable=self.tgt_active_var).pack(
side=tk.LEFT, anchor="w"
)
send_button.pack(side=tk.LEFT, padx=(10, 0))
ttk.Label(
controls_frame,
text="(debug: sends /s /t /r if set)",
foreground="#555555",
font=("Segoe UI", 8),
).pack(side=tk.LEFT, padx=(6, 0))
quick_cmd_frame = ttk.Frame(parent)
quick_cmd_frame.pack(fill=tk.X, pady=(6, 0))
ttk.Button(
quick_cmd_frame,
text="tgtreset",
command=lambda: self._on_send_simple_command(
command_builder.build_tgtreset()
ttk.Checkbutton(
flags_frame, text="Traceable", variable=self.tgt_traceable_var
).pack(side=tk.LEFT, anchor="w", padx=8)
ttk.Checkbutton(
flags_frame, text="Restart", variable=self.tgt_restart_var
).pack(side=tk.LEFT, anchor="w", padx=8)
# Right side: notebook with CMD and JSON tabs
notebook = ttk.Notebook(right)
notebook.pack(fill=tk.BOTH, expand=True)
cmd_tab = ttk.Frame(notebook)
json_tab = ttk.Frame(notebook)
notebook.add(cmd_tab, text="CMD")
notebook.add(json_tab, text="JSON")
# Ensure CMD tab is selected by default so quick CMD buttons are visible
try:
notebook.select(cmd_tab)
except Exception:
pass
# CMD tab: legacy textual quick commands (compact grid to save vertical space)
cmd_btn_frame = ttk.Frame(cmd_tab)
# Pack to the top-left to make buttons immediately visible and reduce
# the chance they're clipped on small widths
cmd_btn_frame.pack(side=tk.TOP, anchor="w", padx=4, pady=4)
btn_specs = [
(
"tgtreset",
lambda: self._on_send_simple_command(command_builder.build_tgtreset()),
),
).pack(side=tk.LEFT, padx=4)
# Add a dedicated Reset button that sends either the JSON reset
# payload or the legacy textual reset sequence depending on the
# selected send mode. The legacy textual commands are sent exactly
# as requested (no leading '$', newline-terminated lines).
ttk.Button(
quick_cmd_frame,
text="reset",
command=self._on_send_reset_button,
).pack(side=tk.LEFT, padx=4)
ttk.Button(
quick_cmd_frame,
text="pause",
command=lambda: self._on_send_simple_command(command_builder.build_pause()),
).pack(side=tk.LEFT, padx=4)
ttk.Button(
quick_cmd_frame,
text="continue",
command=lambda: self._on_send_simple_command(
command_builder.build_continue()
(
"pause",
lambda: self._on_send_simple_command(command_builder.build_pause()),
),
).pack(side=tk.LEFT, padx=4)
(
"continue",
lambda: self._on_send_simple_command(command_builder.build_continue()),
),
(
"Send Target",
lambda: (self.send_mode_var.set("cmd"), self._on_send_target()),
),
("reset (legacy)", lambda: self._on_send_reset_button()),
]
# Arrange buttons in up to 3 columns to reduce vertical height
for idx, (text, cmd) in enumerate(btn_specs):
r = idx // 3
c = idx % 3
ttk.Button(cmd_btn_frame, text=text, command=cmd, width=12).grid(
row=r, column=c, padx=6, pady=2, sticky="w"
)
# JSON tab: new JSON-style quick actions
ttk.Button(
quick_cmd_frame, text="tgtset (cur)", command=lambda: self._on_send_tgtset()
).pack(side=tk.LEFT, padx=8)
# PPI widget and toggle removed from this debug window
json_tab,
text="Send Target (JSON)",
command=lambda: (self.send_mode_var.set("json"), self._on_send_target()),
).pack(side=tk.TOP, anchor="w", padx=4, pady=4)
ttk.Button(
json_tab,
text="Reset (JSON)",
command=lambda: (
self.send_mode_var.set("json"),
self._on_send_reset_button(),
),
).pack(side=tk.TOP, anchor="w", padx=4, pady=4)
# New: send a single JSON that zeroes all targets IDs 0..31
ttk.Button(
json_tab,
text="Reset IDs",
command=lambda: (self.send_mode_var.set("json"), self._on_send_reset_ids()),
).pack(side=tk.TOP, anchor="w", padx=4, pady=4)
# The explicit send-mode radio toggle is no longer necessary because
# the CMD / JSON tabs each provide send buttons for their respective
# payload types. Keep send_mode_var available for buttons to set.
def _create_notebook_tabs(self):
"""Create the notebook tabs used by the SFP Debug Window.
@ -868,6 +902,122 @@ class SfpDebugWindow(tk.Toplevel):
self._log_to_widget("ERROR: Failed to send reset.", "ERROR")
return False
def _on_send_reset_ids(self):
"""Build and send a single JSON payload that contains targets 0..31
with all numeric fields zeroed and flags off. This is a client-side
'hard reset' for the simulation when the server does not implement a
dedicated reset command.
"""
if not self.shared_communicator or not self.shared_communicator.is_open:
self._log_to_widget("ERROR: Cannot send reset IDs, not connected.", "ERROR")
messagebox.showerror(
"Connection Error", "Communicator is not connected.", parent=self
)
return False
try:
targets = []
for tid in range(32):
# Create a minimal waypoint with zeros
wp = Waypoint(
maneuver_type=ManeuverType.FLY_TO_POINT,
target_range_nm=0.0,
target_azimuth_deg=0.0,
target_altitude_ft=0.0,
target_velocity_fps=0.0,
target_heading_deg=0.0,
)
t = Target(target_id=tid, trajectory=[wp])
# Ensure all flags/fields are off/zero
t.active = False
t.traceable = False
t.restart = False
t.current_range_nm = 0.0
t.current_azimuth_deg = 0.0
t.current_velocity_fps = 0.0
t.current_heading_deg = 0.0
t.current_altitude_ft = 0.0
targets.append(t)
# We have a transport limit (MAX_BYTES). Build compact JSONs and
# split into the largest batches that fit the limit.
import json as _json
MAX_BYTES = 1020
def build_compact_payload(tlist):
s = command_builder.build_json_update(tlist)
try:
obj = _json.loads(s)
compact = _json.dumps(obj, separators=(",", ":"))
except Exception:
# Fallback: remove common whitespace
compact = s.replace("\n", "").replace("\r", "")
return compact
n = len(targets)
# Find the largest batch size that fits in MAX_BYTES for the first batch
batch_size = n
while batch_size > 0:
payload = build_compact_payload(targets[:batch_size])
if len(payload.encode("utf-8")) <= MAX_BYTES:
break
batch_size -= 1
if batch_size == 0:
self._log_to_widget(
"ERROR: Cannot fit even a single target into transport limit.",
"ERROR",
)
return False
# Send in batches of batch_size
i = 0
overall_ok = True
first_payload_saved = False
while i < n:
j = min(i + batch_size, n)
payload = build_compact_payload(targets[i:j])
if not payload.endswith("\n"):
payload = payload + "\n"
# Optionally save only the first payload for debugging
if (not first_payload_saved) and hasattr(
self.shared_communicator, "_save_json_payload_to_temp"
):
try:
self.shared_communicator._save_json_payload_to_temp(
payload, f"sfp_debug_reset_ids_part_{i}"
)
first_payload_saved = True
except Exception:
pass
self._log_to_widget(
f"Sending Reset IDs JSON payload part {i}-{j-1}...", "INFO"
)
ok = self.shared_communicator._send_single_command(payload)
overall_ok = overall_ok and bool(ok)
if not ok:
self._log_to_widget(
f"Failed to send Reset IDs payload part {i}-{j-1}.", "ERROR"
)
i = j
if overall_ok:
self._log_to_widget(
"Successfully sent all Reset IDs JSON payloads.", "INFO"
)
else:
self._log_to_widget(
"One or more Reset IDs JSON payloads failed.", "ERROR"
)
return overall_ok
except Exception:
self.logger.exception("Failed to build/send Reset IDs payload")
self._log_to_widget("ERROR: Failed to send Reset IDs.", "ERROR")
return False
def _on_send_simple_command(self, command_str: str):
if not self.shared_communicator or not self.shared_communicator.is_open:
self._log_to_widget("ERROR: Cannot send command, not connected.", "ERROR")
@ -891,86 +1041,19 @@ class SfpDebugWindow(tk.Toplevel):
return False
def _on_send_tgtset(self):
try:
target_id = self.tgt_id_var.get()
range_nm = self.tgt_range_var.get()
az_deg = self.tgt_az_var.get()
alt_ft = self.tgt_alt_var.get()
vel_kn = self.tgt_vel_var.get()
hdg_deg = self.tgt_hdg_var.get()
"""Send a tgtset for the current fields.
vel_fps = vel_kn * KNOTS_TO_FPS
is_active = self.tgt_active_var.get()
is_traceable = self.tgt_traceable_var.get()
updates = {
"range_nm": f"{range_nm:.2f}",
"azimuth_deg": f"{az_deg:.2f}",
"velocity_fps": f"{vel_fps:.2f}",
"heading_deg": f"{hdg_deg:.2f}",
"altitude_ft": f"{alt_ft:.2f}",
"active": is_active,
"traceable": is_traceable,
}
# Support JSON send mode: mirror the same data into a temp Target and send as JSON
if self.send_mode_var.get() == "json":
The UI refactor accidentally duplicated widget-creation code into this
method which referenced a `parent` variable that doesn't exist here.
Keep behaviour simple: delegate to the existing _on_send_target which
builds and sends the appropriate payload based on the selected mode.
"""
try:
initial_waypoint = Waypoint(
maneuver_type=ManeuverType.FLY_TO_POINT,
target_range_nm=range_nm,
target_azimuth_deg=az_deg,
target_altitude_ft=alt_ft,
target_velocity_fps=vel_fps,
target_heading_deg=hdg_deg,
)
temp_target = Target(
target_id=target_id, trajectory=[initial_waypoint]
)
temp_target.current_range_nm = range_nm
temp_target.current_azimuth_deg = az_deg
temp_target.current_velocity_fps = vel_fps
temp_target.current_heading_deg = hdg_deg
temp_target.current_altitude_ft = alt_ft
temp_target.active = is_active
temp_target.traceable = is_traceable
json_payload = command_builder.build_json_update([temp_target])
if hasattr(self.shared_communicator, "_save_json_payload_to_temp"):
try:
self.shared_communicator._save_json_payload_to_temp(
json_payload, "sfp_debug_send"
)
self._on_send_target()
return True
except Exception:
pass
if not json_payload.endswith("\n"):
json_payload = json_payload + "\n"
return self._on_send_simple_command(json_payload)
except Exception:
self.logger.exception(
"Failed to build/send JSON payload for selective tgtset"
)
self._log_to_widget(
"ERROR: Failed to build/send JSON payload.", "ERROR"
)
self.logger.exception("Failed while sending tgtset")
return False
else:
command_str = command_builder.build_tgtset_selective(target_id, updates)
# _on_send_simple_command will normalize/prefix as needed
return self._on_send_simple_command(command_str)
except (ValueError, tk.TclError) as e:
self._log_to_widget(f"ERROR: Invalid input for tgtset: {e}", "ERROR")
return False
def _on_close(self):
self.logger.info("SFP Debug Window closing.")
if self.shared_communicator:
self.shared_communicator.remove_connection_state_callback(
self._update_toggle_state
)
if self.payload_router:
self.payload_router.remove_ris_target_listener(self.update_ppi_targets)
self.destroy()
def _update_toggle_state(self, connected: bool):
try:

2
tests/gui/test_main_view_reset.py
View File

@ -52,6 +52,8 @@ def test_reset_uses_legacy_when_not_configured():
sent = comm.send_commands.call_args[0][0]
assert isinstance(sent, list)
assert sent == ["mex.t_rows=80\n", "tgtset /-s\n"]
import pytest
from target_simulator.gui.main_view import MainView
from target_simulator.core.models import Target