Annotator¶

def delete_current_image(self) -> None:
    """Delete current image with confirmation."""
    current_id = self.controller.get_current_image_id()
    if not current_id:
        messagebox.showwarning("Warning", "No image selected")
        return

    # Get image info for confirmation
    img_info = self.controller.get_image_info(current_id)
    filename = img_info.file_name if img_info else current_id

    # Confirmation dialog
    response = messagebox.askyesno(
        "Delete Image",
        f"Are you sure you want to delete this image?\n\n"
        f"Filename: {filename}\n"
        f"ID: {current_id}\n\n"
        "This will remove the image from the dataset.\n"
        "This action cannot be undone.",
        icon="warning",
    )

    if not response:
        return

    # Delete the image
    if self.controller.delete_current_image():
        # Get next image to display
        next_id = self.controller.get_next_image_after_delete()

        if next_id:
            # Load next image
            self.load_image(next_id)
            self.image_list_panel.select_image(next_id)
            self.update_counter()
            self.status_var.set(f"✓ Image deleted: {filename}")
        else:
            # No more images in this split
            self.canvas.clear()
            self.info_panel.clear()
            self.update_counter()
            self.status_var.set(f"✓ Image deleted: {filename} (no more images)")
            messagebox.showinfo(
                "Split Empty", f"No more images in the '{self.controller.current_split}' split."
            )

        # Update image list panel
        if self.controller.current_split:
            images = self.controller.get_images_in_split(self.controller.current_split)
            self.image_list_panel.set_images(images)

        # Mark workspace as modified
        self._update_window_title()

    else:
        messagebox.showerror("Error", "Failed to delete image")
        self.status_var.set("❌ Failed to delete image")

def import_dataset(self) -> None:
    """Import dataset from COCO, YOLO, or raw image directory.

    Shows ImportDialog for format selection and path input, then
    loads the dataset in a background thread with progress
    indication.
    """
    if not self._check_unsaved_changes():
        return

    dialog = ImportDialog(self.root)  # type: ignore
    result = dialog.show()

    if result:
        format_type = result["format"]
        path = result["path"]
        splits = result["splits"]
        initial_tags = result.get("tags", [])

        # Show loading dialog
        import threading

        from boxlab.annotator.dialogs import LoadingDialog

        loading = LoadingDialog(self.root, "Importing Dataset")  # type: ignore
        loading.update_message(f"Loading {format_type.upper()} dataset...")
        loading.update_status(f"Reading from {path}")

        # Use threading to avoid blocking
        error_container: list[None | BaseException] = [None]

        def do_import_thread() -> None:
            try:
                if format_type == "raw":
                    categories = result.get("categories", [])
                    self.controller.load_dataset(
                        path, format_type, splits, categories, initial_tags
                    )
                else:
                    self.controller.load_dataset(
                        path, format_type, splits, initial_tags=initial_tags
                    )

            except Exception as e:
                error_container[0] = e
                logger.error(f"Failed to load dataset: {e}", exc_info=True)

        def check_thread_done(thread: threading.Thread, loading_dialog: LoadingDialog) -> None:
            def _check_thread_done(*_args: t.Any) -> None:
                check_thread_done(thread, loading_dialog)

            if thread.is_alive():
                self.root.after(100, _check_thread_done, "fuck")
            else:
                loading_dialog.close()

                if error_container[0]:
                    messagebox.showerror(
                        "Error", f"Failed to load dataset:\n{error_container[0]}"
                    )
                    self.status_var.set("Ready")
                else:
                    loading_dialog.update_status("Updating UI...")

                    self.controller.workspace_path = None
                    self.controller.set_workspace_modified(True)
                    self._update_window_title()

                    self.update_after_load()

                    self.status_var.set(
                        f"✓ Loaded {self.controller.total_images()} images from {len(splits)} split(s)"
                    )

        load_thread = threading.Thread(target=do_import_thread, daemon=True)
        load_thread.start()

        def _check_thread_done(*_args: t.Any) -> None:
            check_thread_done(load_thread, loading)

        self.root.after(100, _check_thread_done, "fuck")

def export_dataset(self) -> None:
    """Export dataset to COCO or YOLO format.

    Shows ExportDialog for format and naming options, then exports
    the dataset. Generates audit report JSON if in audit mode.
    """
    if not self.controller.has_dataset():
        messagebox.showwarning("Warning", "No dataset loaded")
        return

    if self.controller.audit_mode:
        stats = self.controller.get_audit_statistics()
        if stats["pending"] > 0:
            response = messagebox.askyesno(
                "Pending Audits",
                f"There are {stats['pending']} images pending audit.\n\n"
                "Do you want to continue exporting?",
            )
            if not response:
                return

    dialog = ExportDialog(self.root)  # type: ignore
    result = dialog.show()

    if result:
        output_dir = result["output_dir"]
        format_type = result["format"]
        naming = result["naming"]
        split_ratio = result.get("split_ratio")

        self.status_var.set(f"Exporting to {format_type.upper()} format...")
        self.root.update()

        try:
            self.controller.export_dataset(output_dir, format_type, naming, split_ratio)

            self.status_var.set(f"✓ Exported to {output_dir}")

            if self.controller.audit_mode:
                import pathlib

                report_path = pathlib.Path(output_dir) / "audit_report.json"
                self.controller.generate_audit_report_json(str(report_path))

                messagebox.showinfo(
                    "Success",
                    f"Dataset exported successfully to:\n{output_dir}\n\n"
                    f"Audit report (JSON) saved to:\n{report_path}",
                )
            else:
                messagebox.showinfo(
                    "Success", f"Dataset exported successfully to:\n{output_dir}"
                )

        except Exception as e:
            messagebox.showerror("Error", f"Failed to export dataset:\n{e}")
            self.status_var.set("Ready")
            logger.error(f"Failed to export dataset: {e}", exc_info=True)

def load_workspace(self) -> None:
    """Load workspace from .cyw file.

    Prompts for workspace file, loads it in background thread, and
    restores the complete application state including annotations,
    tags, and audit status.
    """
    if not self._check_unsaved_changes():
        return

    filepath = filedialog.askopenfilename(
        filetypes=[("Boxlab Workspace", "*.cyw"), ("All Files", "*.*")],
        title="Open Workspace",
    )

    if filepath:
        import threading

        from boxlab.annotator.dialogs import LoadingDialog

        loading = LoadingDialog(self.root, "Loading Workspace")  # type: ignore
        loading.update_message("Loading workspace...")
        loading.update_status(f"Reading {filepath}")

        error_container: list[BaseException | None] = [None]

        def do_load_thread() -> None:
            try:
                self.controller.load_workspace(filepath)
            except Exception as e:
                error_container[0] = e
                logger.error(f"Failed to load workspace: {e}", exc_info=True)

        def check_thread_done(thread: threading.Thread, loading_dialog: LoadingDialog) -> None:
            def _check_thread_done(*_args: t.Any) -> None:
                check_thread_done(thread, loading_dialog)

            if thread.is_alive():
                self.root.after(100, _check_thread_done, "fuck")
            else:
                loading_dialog.close()

                if error_container[0]:
                    messagebox.showerror(
                        "Error", f"Failed to load workspace:\n{error_container[0]}"
                    )
                else:
                    self._update_window_title()
                    self.update_after_load()

                    current_id = self.controller.get_current_image_id()
                    if current_id:
                        self.load_image(current_id)

                    self.status_var.set("✓ Workspace loaded")
                    messagebox.showinfo("Success", f"Workspace loaded from:\n{filepath}")

        load_thread = threading.Thread(target=do_load_thread, daemon=True)
        load_thread.start()

        def _check_thread_done(*_args: t.Any) -> None:
            check_thread_done(load_thread, loading)

        self.root.after(100, _check_thread_done, "fuck")

def save_workspace(self) -> None:
    """Save current workspace to associated .cyw file.

    If no workspace path is set, delegates to save_workspace_as().
    """
    if not self.controller.has_dataset():
        messagebox.showwarning("Warning", "No dataset loaded")
        return

    if not self.controller.workspace_path:
        self.save_workspace_as()
        return

    try:
        self.controller.save_workspace()
        self._update_window_title()
        self.status_var.set("✓ Workspace saved")
    except Exception as e:
        messagebox.showerror("Error", f"Failed to save workspace:\n{e}")
        logger.error(f"Failed to save workspace: {e}", exc_info=True)

def save_workspace_as(self) -> None:
    """Save workspace to a new .cyw file.

    Prompts for file location and saves complete workspace state.
    """
    if not self.controller.has_dataset():
        messagebox.showwarning("Warning", "No dataset loaded")
        return

    filepath = filedialog.asksaveasfilename(
        defaultextension=".cyw",
        filetypes=[("Boxlab Workspace", "*.cyw"), ("All Files", "*.*")],
        title="Save Workspace As",
    )

    if filepath:
        try:
            self.controller.save_workspace(filepath)
            self._update_window_title()
            self.status_var.set(f"✓ Workspace saved: {filepath}")
            messagebox.showinfo("Success", f"Workspace saved to:\n{filepath}")
        except Exception as e:
            messagebox.showerror("Error", f"Failed to save workspace:\n{e}")
            logger.error(f"Failed to save workspace: {e}", exc_info=True)

def next_image(self) -> None:
    """Navigate to next image in current split."""
    next_id = self.controller.next_image()
    if next_id:
        self.load_image(next_id)
        self.image_list_panel.select_image(next_id)
        self.update_counter()

def prev_image(self) -> None:
    """Navigate to previous image in current split."""
    prev_id = self.controller.prev_image()
    if prev_id:
        self.load_image(prev_id)
        self.image_list_panel.select_image(prev_id)
        self.update_counter()

def load_image(self, image_id: str) -> None:
    """Load and display image with annotations.

    Args:
        image_id: ID of the image to load.
    """
    try:
        image_info = self.controller.get_image_info(image_id)
        annotations = self.controller.get_annotations(image_id)

        if image_info and image_info.path and image_info.path.exists():
            from PIL import Image

            img = Image.open(image_info.path)

            self.canvas.display_image(img, annotations, self.controller.get_categories())

            audit_status = None
            if self.controller.audit_mode:
                audit_status = self.controller.get_audit_status(image_id)
                comment = self.controller.get_audit_comment(image_id)
                self.control_panel.update_audit_status(audit_status)
                self.control_panel.set_audit_comment(comment)

            tags = self.controller.get_image_tags(image_id)
            self.info_panel.set_current_tags(tags)

            self.info_panel.update_image_info(
                image_info=image_info,
                annotations=annotations,
                source=self.controller.get_image_source(image_id),
                audit_status=audit_status,
            )

            self.status_var.set(f"🖼 {image_info.file_name}")

    except Exception as e:
        messagebox.showerror("Error", f"Failed to load image:\n{e}")
        logger.error(f"Failed to load image {image_id}: {e}", exc_info=True)

def on_image_selected(self, image_id: str) -> None:
    """Handle image selection from list panel.

    Args:
        image_id: ID of the selected image.
    """
    if self.controller.current_split:
        images = self.controller.image_ids_by_split.get(self.controller.current_split, [])
        try:
            self.controller.current_index = images.index(image_id)
            self.load_image(image_id)
            self.update_counter()
        except ValueError:
            pass

def on_split_changed(self, split: str) -> None:
    """Handle split selection change.

    Args:
        split: Name of the selected split (e.g., "train", "val", "test").
    """
    self.controller.set_current_split(split)

    images = self.controller.get_images_in_split(split)
    self.image_list_panel.set_images(images)

    if self.controller.audit_mode:
        split_audit_map = {}
        for img_id, _ in images:
            split_audit_map[img_id] = self.controller.get_audit_status(img_id)

        self.image_list_panel.set_audit_status_map(split_audit_map)

        audit_stats = self.controller.get_audit_statistics()
        self.info_panel.update_dataset_info(self.controller.get_dataset_info(), audit_stats)

    if images:
        self.load_image(images[0][0])

    self.update_counter()

def on_annotations_changed(self) -> None:
    """Handle annotation modifications on canvas."""
    current_id = self.controller.get_current_image_id()
    if current_id:
        annotations = self.canvas.get_annotations()
        self.controller.update_annotations(current_id, annotations)

        image_info = self.controller.get_image_info(current_id)
        if image_info:
            self.info_panel.update_image_info(
                image_info=image_info,
                annotations=annotations,
                source=self.controller.get_image_source(current_id),
            )

    self._update_window_title()
    self.status_var.set("⚠️ Unsaved changes")

def on_edit_mode_changed(self, enabled: bool) -> None:
    """Handle edit mode toggle.

    Args:
        enabled: Whether edit mode is enabled.
    """
    self.canvas.set_edit_mode(enabled)
    mode_text = "ON" if enabled else "OFF"
    self.status_var.set(f"Edit mode: {mode_text}")

def on_category_changed(self, category_id: int | None) -> None:
    """Handle category selection change.

    Args:
        category_id: Selected category ID, or None.
    """
    self.canvas.set_current_category(category_id)

def on_tags_changed(self, tags: list[str]) -> None:
    """Handle tags change for current image.

    Args:
        tags: List of tag strings.
    """
    current_id = self.controller.get_current_image_id()

    if current_id:
        self.controller.set_image_tags(current_id, tags)

        self.info_panel.set_current_tags(tags)
        self._update_window_title()
        self.status_var.set("⚠️ Unsaved changes")

def on_new_tag_created(self, tag: str) -> None:
    """Handle new tag creation.

    Args:
        tag: New tag string.
    """
    self.controller.add_tag(tag)
    self.info_panel.set_available_tags(self.controller.available_tags)
    self._update_window_title()
    self.status_var.set(f"✓ New tag created: {tag}")

def on_audit_mode_changed(self, enabled: bool) -> None:
    """Handle audit mode toggle.

    Args:
        enabled: Whether audit mode is enabled.
    """
    self.controller.enable_audit_mode(enabled)

    self.image_list_panel.show_audit_filter(enabled)

    if enabled:
        complete_audit_map = {}
        for split_images in self.controller.image_ids_by_split.values():
            for img_id in split_images:
                complete_audit_map[img_id] = self.controller.get_audit_status(img_id)

        self.image_list_panel.set_audit_status_map(complete_audit_map)

        audit_stats = self.controller.get_audit_statistics()
        self.info_panel.update_dataset_info(self.controller.get_dataset_info(), audit_stats)

        current_id = self.controller.get_current_image_id()
        if current_id:
            status = self.controller.get_audit_status(current_id)
            self.control_panel.update_audit_status(status)

    mode_text = "ON" if enabled else "OFF"
    self.status_var.set(f"Audit mode: {mode_text}")

def approve_current(self) -> None:
    """Approve current image and move to next."""
    if not self.controller.audit_mode:
        return

    current_id = self.controller.get_current_image_id()
    if not current_id:
        return

    self.controller.set_audit_status(current_id, "approved")
    self.status_var.set("✓ Image approved")

    complete_audit_map = {}
    current_split_images = self.controller.get_images_in_split(
        self.controller.current_split or ""
    )
    for img_id, _ in current_split_images:
        complete_audit_map[img_id] = self.controller.get_audit_status(img_id)

    self.image_list_panel.set_audit_status_map(complete_audit_map)

    audit_stats = self.controller.get_audit_statistics()
    self.info_panel.update_dataset_info(self.controller.get_dataset_info(), audit_stats)

    next_id = self.controller.next_image()

    if next_id:
        self.load_image(next_id)
        self.image_list_panel.select_image(next_id)
        self.update_counter()

        status = self.controller.get_audit_status(next_id)
        self.control_panel.update_audit_status(status)

def reject_current(self) -> None:
    """Reject current image and move to next."""
    if not self.controller.audit_mode:
        return

    current_id = self.controller.get_current_image_id()
    if not current_id:
        return

    self.controller.set_audit_status(current_id, "rejected")
    self.status_var.set("✗ Image rejected")

    complete_audit_map = {}
    current_split_images = self.controller.get_images_in_split(
        self.controller.current_split or ""
    )
    for img_id, _ in current_split_images:
        complete_audit_map[img_id] = self.controller.get_audit_status(img_id)

    self.image_list_panel.set_audit_status_map(complete_audit_map)

    audit_stats = self.controller.get_audit_statistics()
    self.info_panel.update_dataset_info(self.controller.get_dataset_info(), audit_stats)

    next_id = self.controller.next_image()

    if next_id:
        self.load_image(next_id)
        self.image_list_panel.select_image(next_id)
        self.update_counter()

        status = self.controller.get_audit_status(next_id)
        self.control_panel.update_audit_status(status)

def on_audit_comment_changed(self, comment: str) -> None:
    """Handle audit comment change.

    Args:
        comment: Comment text.
    """
    current_id = self.controller.get_current_image_id()
    if current_id:
        self.controller.set_audit_comment(current_id, comment)
        self._update_window_title()

def show_audit_report(self) -> None:
    """Show audit report in a text dialog."""
    if not self.controller.audit_mode:
        messagebox.showinfo("Info", "Audit mode is not enabled")
        return

    report = self.controller.generate_audit_report()

    dialog = tk.Toplevel(self.root)
    dialog.title("Audit Report")
    dialog.geometry("600x400")
    dialog.transient(self.root)  # type: ignore

    dialog.update_idletasks()
    x = (dialog.winfo_screenwidth() // 2) - (600 // 2)
    y = (dialog.winfo_screenheight() // 2) - (400 // 2)
    dialog.geometry(f"+{x}+{y}")

    text_frame = ttk.Frame(dialog, padding=10)
    text_frame.pack(fill="both", expand=True)

    text = tk.Text(text_frame, wrap="word", font=("Consolas", 10))
    scrollbar = ttk.Scrollbar(text_frame, command=text.yview)
    text.config(yscrollcommand=scrollbar.set)

    text.pack(side="left", fill="both", expand=True)
    scrollbar.pack(side="right", fill="y")

    text.insert("1.0", report)
    text.config(state="disabled")

    ttk.Button(dialog, text="Close", command=dialog.destroy, width=12).pack(pady=10)

def export_audit_report(self) -> None:
    """Export audit report to JSON file."""
    if not self.controller.audit_mode:
        messagebox.showinfo("Info", "Audit mode is not enabled")
        return

    filepath = filedialog.asksaveasfilename(
        defaultextension=".json",
        filetypes=[("JSON Files", "*.json"), ("All Files", "*.*")],
        title="Export Audit Report (JSON)",
    )

    if filepath:
        try:
            self.controller.generate_audit_report_json(filepath)
            self.status_var.set(f"✓ Audit report exported: {filepath}")
            messagebox.showinfo("Success", f"Audit report (JSON) exported to:\n{filepath}")
        except Exception as e:
            messagebox.showerror("Error", f"Failed to export report:\n{e}")
            logger.error(f"Failed to export report: {e}", exc_info=True)

def open_current_dashboard(self) -> None:
    """Open Streamlit dashboard for current audit state."""
    if not self.controller.audit_mode:
        messagebox.showinfo("Info", "Audit mode is not enabled")
        return

    import subprocess
    import tempfile
    import threading

    # Export current audit state to temp file
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".json",
        delete=False,
        prefix="boxlab_audit_",
    ) as temp_file:
        temp_path = temp_file.name

    try:
        self.controller.generate_audit_report_json(temp_path)

        # Get dashboard path
        dashboard_path = pathlib.Path(__file__).parent / "streamlit" / "audit.py"

        if not dashboard_path.exists():
            messagebox.showerror("Error", f"Dashboard not found: {dashboard_path}")
            return

        # Show dialog with server info
        dialog = tk.Toplevel(self.root)
        dialog.title("Audit Dashboard Server")
        dialog.geometry("500x300")
        dialog.transient(self.root)

        # Center dialog
        dialog.update_idletasks()
        x = (dialog.winfo_screenwidth() // 2) - 250
        y = (dialog.winfo_screenheight() // 2) - 150
        dialog.geometry(f"+{x}+{y}")

        main_frame = ttk.Frame(dialog, padding=20)
        main_frame.pack(fill="both", expand=True)

        ttk.Label(
            main_frame, text="🚀 Starting Audit Dashboard...", font=("Segoe UI", 12, "bold")
        ).pack(pady=(0, 10))

        status_var = tk.StringVar(value="Initializing server...")
        status_label = ttk.Label(main_frame, textvariable=status_var)
        status_label.pack(pady=10)

        # Server info
        info_frame = ttk.LabelFrame(main_frame, text="Server Information", padding=10)
        info_frame.pack(fill="both", expand=True, pady=10)

        url_var = tk.StringVar(value="Starting...")
        ttk.Label(info_frame, text="Local URL:", font=("Segoe UI", 9, "bold")).pack(anchor="w")
        url_label = ttk.Label(info_frame, textvariable=url_var, foreground="blue")
        url_label.pack(anchor="w", pady=(0, 10))

        network_var = tk.StringVar(value="Starting...")
        ttk.Label(info_frame, text="Network URL:", font=("Segoe UI", 9, "bold")).pack(
            anchor="w"
        )
        network_label = ttk.Label(info_frame, textvariable=network_var, foreground="blue")
        network_label.pack(anchor="w")

        ttk.Label(
            info_frame,
            text="💡 Share the Network URL with your team to access the dashboard",
            font=("Segoe UI", 8),
            foreground="gray",
        ).pack(pady=(10, 0))

        # Buttons
        button_frame = ttk.Frame(main_frame)
        button_frame.pack(pady=(10, 0))

        process_container: dict[str, subprocess.Popen[str] | None] = {"process": None}

        def stop_server() -> None:
            if process_container["process"]:
                process_container["process"].terminate()
            dialog.destroy()

        def open_browser() -> None:
            import webbrowser

            url = url_var.get()
            if url.startswith("http"):
                webbrowser.open(url)

        open_btn = ttk.Button(button_frame, text="Open in Browser", command=open_browser)
        open_btn.pack(side="left", padx=5)
        open_btn.config(state="disabled")

        ttk.Button(button_frame, text="Stop Server", command=stop_server).pack(
            side="left", padx=5
        )

        dialog.protocol("WM_DELETE_WINDOW", stop_server)

        # Start server in background
        def start_server() -> None:
            try:
                # Start streamlit server
                cmd = [
                    sys.executable,
                    "-m",
                    "streamlit",
                    "run",
                    str(dashboard_path),
                    "--server.headless=true",
                    "--server.port=8501",
                    "--",
                    temp_path,
                ]

                process = subprocess.Popen(
                    cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1
                )

                process_container["process"] = process

                # Parse output for URLs
                import socket

                hostname = socket.gethostname()
                local_ip = socket.gethostbyname(hostname)

                local_url = "http://localhost:8501"
                network_url = f"http://{local_ip}:8501"

                # Update UI
                dialog.after(2000, status_var.set, "✅ Server is running!")
                dialog.after(2000, url_var.set, local_url)
                dialog.after(2000, network_var.set, network_url)
                dialog.after(2000, lambda _: open_btn.config(state="normal"), "fuck")

                # Auto-open browser
                import webbrowser

                dialog.after(3000, webbrowser.open, local_url, 0, True)

            except Exception as e:
                dialog.after(0, status_var.set, f"❌ Error: {e}")
                logger.error(f"Failed to start dashboard: {e}", exc_info=True)

        # Start server thread
        server_thread = threading.Thread(target=start_server, daemon=True)
        server_thread.start()

    except Exception as e:
        messagebox.showerror("Error", f"Failed to start dashboard:\n{e}")
        logger.error(f"Failed to start dashboard: {e}", exc_info=True)