"""
Helper functions for Mujoco experiments.

Multiprocessing seems to break if certain libraries are imported (gym or SB3 probably?) before the processes are spawned.
Some issue with pytorch, and apple's Metal/MPS. Behavior on my machine: either crashes the first time autograd is called (BC training),
or runs through 1st experiment, but when 2nd experiment spawns new processes, crashes w/ reference to Metal/MPS.
Some functions triggering this or otherwise causing problems are moved here, which seems to avoid the issue.

Trajectory: because pre-saved trajs I had were previously created with __main__ == experiment_runner_mujoco.py,
  which causes an error when executing via main_mujoco.py
GaussianPolicy: for safety, because of the torch defn's
Parser: because if it's defined in core/parser.py, importing it causes imports of all of `core` which contains problematic libraries.
"""

import numpy as np
from typing import List, Dict, Any, Optional, Tuple
import torch
from dataclasses import dataclass
import hashlib
import torch.nn as nn
import matplotlib.pyplot as plt

import argparse
import yaml
import os
import sys


# ====== Trajectory class & manipulation ======
@dataclass(slots=True)
class Trajectory:
    """Represents a trajectory with states, actions, rewards, and info dictionaries.

    For HalfCheetah the infos are important, b/c they contain the x-coordinate (that measures progress&reward)"""

    states: List[np.ndarray]
    actions: List[np.ndarray]
    rewards: List[float]  # assume scalar reward
    infos: List[Dict[str, Any]]
    true_obs: List[np.ndarray] = None  # in case obs are normalized (reacher, HC)
    # extras: Optional[np.ndarray] = None  # shape [T, K] or None

    def __post_init__(self):
        if self.true_obs is None:
            self.true_obs = self.states

    def __len__(self):
        return len(self.states)

    def __getitem__(self, idx):
        """Allow indexing like traj[0] -> (state, action, reward, info)"""
        return (
            self.states[idx],
            self.actions[idx],
            self.rewards[idx],
            self.infos[idx],
            self.true_obs[idx],
        )

    def to_sa_pairs(self) -> List[Tuple[np.ndarray, np.ndarray]]:
        """returns [(s0, a0), (s1, a1), ...]"""
        return list(zip(self.true_obs, self.actions))

    def total_return(self) -> float:
        return float(np.sum(self.rewards))

    def to_tensors(self):
        """returns tuple (states, actions), both tensors"""
        return (
            torch.tensor(self.true_obs, dtype=torch.float32),
            torch.tensor(self.actions, dtype=torch.float32),
        )

    # HalfCheetah/Walker-specific:
    def get_x_positions(self) -> np.ndarray:
        """Extract x_positions from info dictionaries."""
        return np.array([info["x_position"] for info in self.infos])

    def get_final_x_position(self) -> float:
        """Get the final x_position."""
        return self.infos[-1]["x_position"]

    def compute_discounted_cumulative_reward(self, discount_factor):
        return np.dot(self.rewards, np.power(discount_factor, np.arange(len(self.rewards))))

    def compute_cumulative_reward(self):
        return np.sum(self.rewards)


def extract_x_pos_from_info(info: Dict[str, Any]) -> Optional[np.ndarray]:
    # Keep this tiny and fixed-size; example key
    if "x_position" in info:
        return np.array([info["x_position"]], dtype=np.float32)
    return None


# ====== Policy class, getting actions, rollouts ======
class GaussianPolicy(nn.Module):
    """Gaussian policy, with MLP.

    Note: to get an action, call the external function policy_action.
    Reason to not have .act() or .predict() is that we want the BRIDGE code
    to work with SB3 policies, with ours, and with obs from SB3 VecEnvs or gym envs
    """

    # init
    def __init__(self, state_dim, action_dim, hidden_dim, device="cpu"):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
        )
        self.mean = nn.Linear(hidden_dim, action_dim)
        self.log_std = nn.Parameter(torch.ones(action_dim) * -2)
        self.device = device
        # self._computing_hash = False  # add recursion guard

    # forward: used in training
    def forward(self, state):
        x = self.net(state)
        mean = self.mean(x)
        std = self.log_std.exp().expand_as(mean)
        return mean, std

    # predict: used in eval
    def predict(self, obs, deterministic=False):
        """predicts action given obs (only used in eval)"""
        single_obs = False
        if len(obs.shape) == 1:
            obs = obs.unsqueeze(0)
            single_obs = True

        obs = torch.tensor(obs, dtype=torch.float32).to(self.device)

        with torch.no_grad():
            mean, std = self(obs)[:2]  # forward pass
            if deterministic:
                action = mean
            else:
                action = torch.normal(mean, std)
            action = action.detach().cpu().numpy()
            if single_obs:
                action = action.squeeze(0)
        return action, None  # None for hidden state

    def get_param_hash(self, truncate_to=16):
        # if self._computing_hash:
        # return "hash-recursion-detected"
        # self._computing_hash = True
        # try:
        param_bytes = []
        # for param in self.parameters():
        # param_bytes.append(param.data.cpu().numpy().tobytes())
        for param in self.state_dict().values():
            param_bytes.append(param.cpu().numpy().tobytes())
        combined = b"".join(param_bytes)
        hash_obj = hashlib.sha256(combined)
        if truncate_to:
            return hash_obj.hexdigest()[:truncate_to]
        return hash_obj.hexdigest()
        # finally:
        # self._computing_hash = False

    def __hash__(self):
        return int(self.get_param_hash(), 16)  # truncates 16 chars (by default), of base 16

    def __eq__(self, other):
        if not isinstance(other, GaussianPolicy):
            return False
        return torch.allclose(
            torch.cat([p.flatten() for p in self.parameters()]),
            torch.cat([p.flatten() for p in other.parameters()]),
        )
        # self.get_param_hash() == other.get_param_hash()


def hash_policy(policy, truncate_to=16):
    """returns hash (str) for both GaussianPolicy and SB3 policies (e.g. PPO), by default truncates to 16 chars"""
    if hasattr(policy, "get_param_hash"):  # isinstance(policy, GaussianPolicy)
        hex_str = policy.get_param_hash(truncate_to)
        return int(hex_str, 16)
    elif hasattr(policy, "policy"):  # isinstance(policy, PPO)
        param_bytes = []
        for param in policy.policy.parameters():
            param_bytes.append(param.data.cpu().numpy().tobytes())
        combined = b"".join(param_bytes)
        hash_obj = hashlib.sha256(combined)
        hex_str = hash_obj.hexdigest()[:truncate_to]
        return int(hex_str, 16)
    else:
        raise ValueError(f"Policy type {type(policy)} not supported in hash_policy")


def pad_metrics_mujoco(metrics, params):
    """
    if loop terminated early in one seed, then some of the metrics don't have the right length.
    fix this by padding them at the end with the last known value of each metric
    """
    no_padding_needed = True
    for seed in range(len(metrics)):
        for metric in metrics[seed]:
            if len(metrics[seed][metric]) < params["N_iterations"]:
                padding_length = params["N_iterations"] - len(metrics[seed][metric])
                # Pad with the last known value
                last_value = metrics[seed][metric][-1] if metrics[seed][metric] else 0
                metrics[seed][metric].extend([last_value] * padding_length)
                if "plotting" in params["verbose"] or "full" in params["verbose"]:
                    print(f"metric {metric} padded {padding_length} for seed {seed}")
                no_padding_needed = False
    if no_padding_needed and ("plotting" in params["verbose"] or "full" in params["verbose"]):
        print("no padding needed")
    return metrics


def str_to_bool(v):
    """Convert string representations of truth to True or False."""
    if isinstance(v, bool):
        return v
    if v.lower() in ("yes", "true", "t", "y", "1"):
        return True
    elif v.lower() in ("no", "false", "f", "n", "0"):
        return False
    else:
        raise argparse.ArgumentTypeError(f"Boolean value expected, got: {v}")


def none_or_str(value):
    if value.lower() in ["none", "null"]:
        return None
    return value


def parse_args_mujoco(args=None, base_config=None):
    """Parse command line arguments and load configuration."""
    parser = argparse.ArgumentParser(description="Run preference-based RL experiments")

    # Config selection arguments
    parser.add_argument(
        "-cn",
        "--config-name",
        type=str,
        default=None,
        help="Config name to load (e.g., 'special/debug_starmdp' or 'exps/123')",
    )
    parser.add_argument(
        "-env",
        "--environment",
        type=str,
        default=None,
        choices=["Reacher-v5", "HalfCheetah-v5", "reacher", "halfcheetah"],
        dest="env_id",
        help="Environment name (gym ID) to load default config for. Only used if no config name specified. Options: Reacher-v5, HalfCheetah-v5",
    )

    # Broad experiment params
    parser.add_argument(
        "--seed",
        type=int,
        default=42,
        help="Seed for the experiment",
    )
    parser.add_argument(
        "--N_experiments",
        "-seeds",
        type=int,
        dest="N_experiments",
        help="Number of experiments (seeds)",
    )
    parser.add_argument(
        "--N_iterations", "-its", type=int, dest="N_iterations", help="Online iterations per seed"
    )
    parser.add_argument("--episode_length", type=int, help="Episode length")
    parser.add_argument(
        "--phi_name",
        "--embedding_name",
        type=str,
        choices=[
            "avg_sa",
            "avg_s",
            "last_s",
            "actionenergy",
            "psm",
            "halfcheetah_xpos",
            "reacher_perf",
            "ant_perf",
            "walker2d_perf",
            "hopper_perf",
            "humanoid_perf",
            "walker2d_extended",
            "hopper_extended",
            "humanoid_extended",
        ],
        dest="embedding_name",
        help="Embedding function",
    )
    parser.add_argument(
        "--N_offline_trajs",
        "--trajs",
        "-trajs",
        type=int,
        dest="N_offline_trajs",
        help="Number of offline trajectories",
    )
    parser.add_argument(
        "--fresh_offline_trajs", type=str_to_bool, help="Fresh offline trajectories"
    )
    parser.add_argument(
        "--initial_pos_noise",
        type=float,
        help="Initial position noise for HalfCheetah. Default is 0.1",
    )

    # Offline learning parameters
    parser.add_argument("--N_confset_size", type=int, help="Number of initial policies sampled")
    parser.add_argument(
        "--confset_base",
        "-conf_base",
        dest="confset_base",
        type=str,
        choices=["bcnoise", "bignoise", "random"],
        help="What the candidate set is made of. 'bcnoise' (BC, +noise), 'bignoise' (BC+10x noise, w/o BC), 'random' (random policies).",
    )
    parser.add_argument(
        "--confset_dilution",
        "-conf_dilution",
        "-dilute",
        "-dilution",
        dest="confset_dilution",
        type=none_or_str,
        default=None,
        help="What gets added to the confset_base to augment the candidate set. 'None', 'random' (add random policies), 'bignoise' (add BC+10x noise policies).",
    )
    parser.add_argument(
        "--N_confset_dilution",
        "-n_dilution",
        dest="N_confset_dilution",
        type=int,
        help="Number of policies to add to the confset_base.",
    )
    parser.add_argument(
        "--confset_noise",
        type=float,
        help="Noise added to BC policy to generate confset. If not provided, filled in w/ environment defaults: Reacher: 0.05, HalfCheetah: ???",
    )
    parser.add_argument("--n_bc_epochs", type=int, help="Number of BC epochs")
    parser.add_argument("--bc_loss", type=str, choices=["log-loss", "mse"], help="BC loss")
    parser.add_argument("--bc_print_evals", type=str_to_bool, help="Print BC evaluations")
    parser.add_argument(
        "--radius",
        "-radius",
        type=float,
        dest="radius",
        help="Radius for filtering offline confset: L2(embed(π_BC) - embed(π_candidate)) < radius. If unspecified, uses hardcoded defaults per embedding.",
    )
    parser.add_argument(
        "--expert_in_candidates",
        "-exp_in_cand",
        dest="expert_in_candidates",
        type=str_to_bool,
        help="Expert in search space",
    )
    parser.add_argument(
        "--expert_in_confset",
        "-exp_in_conf",
        dest="expert_in_confset",
        type=str_to_bool,
        help="Expert in confset",
    )
    parser.add_argument(
        "--expert_in_eval",
        "-exp_in_eval",
        dest="expert_in_eval",
        type=str_to_bool,
        help="Expert in eval",
    )
    parser.add_argument(
        "--which_eval_space",
        "-eval_space",
        dest="which_eval_space",
        type=str,
        choices=["candidates", "pi_zero", "pi_t"],
        help="Which eval space",
    )
    parser.add_argument(
        "--fresh_embeddings",
        type=str_to_bool,
        help="Force recalculation of embeddings",
    )

    # Online learning parameters
    parser.add_argument("--N_rollouts", type=int, help="Number of rollouts per iteration")
    parser.add_argument(
        "--filter_pi_t_yesno",
        "--filter_online",
        dest="filter_pi_t_yesno",
        type=str_to_bool,
        help="Filter Pi_t according to {pi in Pi_0 s.t. for all other pi': Δϕᵀw + γ * sqrt(Δϕᵀ V_inv Δϕ) >= 0}",
    )
    parser.add_argument(
        "--filter_pi_t_gamma",
        "--filter_online_gamma",
        "--filter_gamma",
        dest="filter_pi_t_gamma",
        type=float,
        help="Gamma for filtering Pi_t",
    )
    parser.add_argument(
        "--gamma_debug_mode",
        "--gamma_debug",
        dest="gamma_debug_mode",
        type=str_to_bool,
        default=False,
        help="Debug mode for gamma",
    )
    parser.add_argument("--W", type=int, help="W parameter")
    parser.add_argument(
        "--w_trainfunc",
        type=str,
        choices=["rebuttals", "mle"],
        help="Trainfunc for w",
    )
    parser.add_argument(
        "--w_regularization",
        type=none_or_str,
        default=None,
        help="Regularization for w (None or l2)",
    )
    parser.add_argument("--w_epochs", type=int, help="MLE epochs for w")
    parser.add_argument(
        "--w_initialization",
        "-w",
        "-w_init",
        dest="w_initialization",
        type=str,
        choices=["uniform", "zeros", "random"],
        help="Weight initialization method",
    )
    parser.add_argument(
        "--project_w",
        type=str_to_bool,
        help="Project w s.t. ||w|| <= W",
    )
    parser.add_argument(
        "--retrain_w_from_scratch",
        type=str_to_bool,
        help="Retrain w from scratch",
    )
    parser.add_argument("--w_sigmoid_slope", type=float, help="Sigmoid slope for weights")
    parser.add_argument(
        "--which_policy_selection",
        type=str,
        choices=["random", "ucb", "max_uncertainty"],
        help="Policy selection method",
    )
    parser.add_argument(
        "--ucb_beta",
        "--beta",
        "--ucb",
        dest="ucb_beta",
        type=float,
        help="Beta for UCB policy selection. Used when selecting policy pairs with UCB: formula is ucb_score = σ(Δϕᵀ w) + ucb_beta * sqrt(Δϕᵀ V_inv Δϕ)",
    )
    parser.add_argument(
        "--V_init", type=str, choices=["small", "bounds"], help="V initialization method"
    )
    parser.add_argument(
        "--n_embedding_samples",
        "-n_samples",
        type=int,
        dest="n_embedding_samples",
        help="Number of samples for estimating policy embedding",
    )

    # policy model params
    parser.add_argument(
        "--hidden_dim",
        type=int,
        help="Hidden dimension of policy model. SB3 defaults to 64 x2, halfcheetah: 256 x2",
    )

    # Verbosity and saving
    parser.add_argument(
        "--verbose",
        nargs="*",  # user can pass multiple options that are concatted into list
        help="Verbosity options",
    )
    parser.add_argument(
        "--run_baseline", "-baseline", dest="run_baseline", type=str_to_bool, help="Run baseline"
    )
    parser.add_argument(
        "--run_bridge", "-bridge", dest="run_bridge", type=str_to_bool, help="Run bridge"
    )
    parser.add_argument("--save_results", type=str_to_bool, help="Save results")
    parser.add_argument("--run_ID", "-id", dest="run_ID", type=str, help="ID")
    parser.add_argument(
        "--loaded_run_behaviour",
        "--loaded_run_behavior",  # for the americans
        "--load",
        "-load",
        type=str,
        dest="loaded_run_behaviour",
        choices=["continue", "redo", "overwrite"],
        help="Purpose of loaded run",
    )
    parser.add_argument(
        "--which_plot_subopt",
        "-plot",
        "--plot",
        type=str,
        dest="which_plot_subopt",
        nargs="*",
        choices=[
            "suboptimality_percent",
            "regret",
            "regret_indiv",
            "cumulative_regret",
            "raw_reward",
        ],
        help="Which plot to show",
    )
    parser.add_argument(
        "--baseline_or_bridge",
        type=str,
        choices=["baseline", "bridge"],
        help="Which method to run",
    )
    parser.add_argument(
        "--plot_scores",
        type=str_to_bool,
        help="Plot histogram of all candidates' scores at each loop iteration",
    )
    parser.add_argument(
        "--exclude_outliers",
        "--outliers",
        "-outliers",
        "--exclude",
        "-exclude",
        dest="exclude_outliers",
        type=str,
        help="Exclude outliers from the analysis",
        choices=[
            "worst_bcexpertdist",
            "worst_cumregret",
            "95conf_bcexpertdist",
            "95conf_cumregret",
        ],
    )
    parser.add_argument(
        "--use_wandb",
        "-wandb",
        dest="use_wandb",
        type=str_to_bool,
        help="Use wandb",
    )
    parser.add_argument(
        "--plot_slim",
        type=str_to_bool,
        help="Plot in slim mode",
    )
    parser.add_argument(
        "--plot_logy",
        type=str_to_bool,
        help="Plot in logy mode",
    )

    parsed_args = parser.parse_args(args)

    # Load configuration based on priority
    config_path = None

    # 1. Load user-specified config
    if parsed_args.config_name:
        config_path = os.path.join("configs", f"{parsed_args.config_name}.yaml")
        if not os.path.exists(config_path):
            raise FileNotFoundError(f"User-specified config file not found: {config_path}")
        with open(config_path, "r") as f:
            try:
                params = yaml.safe_load(f)
            except yaml.YAMLError as e:
                print(f"Error parsing YAML file {config_path}: {e}")
                raise
        print(f"Loaded config from: {config_path}")

    # 2. If no config specified, load default config corresponding to user-specified environment
    elif parsed_args.env_id:
        # Look for environment-specific default configs
        if parsed_args.env_id in ["HalfCheetah-v5", "halfcheetah"]:
            envname = "hc"
        elif parsed_args.env_id in ["Reacher-v5", "reacher"]:
            envname = "reacher"
        else:
            raise ValueError(f"Invalid environment ID: {parsed_args.env_id}")
        config_path = os.path.join("configs", "special", f"{envname}_default.yaml")
        if not os.path.exists(config_path):
            raise ValueError(f"Environment-specific config at {config_path} not found. Aborting.")
        with open(config_path, "r") as f:
            try:
                params = yaml.safe_load(f)
            except yaml.YAMLError as e:
                print(f"Error parsing YAML file {config_path}: {e}")
                raise
        print(f"Loaded config from: {config_path}")

    elif base_config is not None:
        params = base_config.copy()
        print("Using base config hardcoded in main_mujoco.py file")

    # 3. If neither config nor environment specified, load global default config
    else:
        config_path = os.path.join("configs", "special", "starmdp_default.yaml")
        if not os.path.exists(config_path):
            raise FileNotFoundError(f"Global default config at {config_path} not found. Aborting.")
        with open(config_path, "r") as f:
            params = yaml.safe_load(f)
        print(f"Loaded config from: {config_path}")

    assert params is not None

    # Override with command line arguments
    for key, value in vars(parsed_args).items():
        # Skip the config selection arguments
        if key in ["config_name", "environment"]:
            continue
        # Only override if the value was explicitly provided (not None)
        if value is not None:
            try:
                params[key] = value
                print(f"Overriding {key} with {value}")
            except Exception as e:
                print(f"Error overriding {key} with {value}: {e}")

    # handling of list arguments (in case they come as None)
    if "verbose" not in params or params["verbose"] is None:
        params["verbose"] = []

    if "which_plot_subopt" not in params or params["which_plot_subopt"] is None:
        params["which_plot_subopt"] = []

    if isinstance(params["which_plot_subopt"], str):
        params["which_plot_subopt"] = [params["which_plot_subopt"]]

    return params


def print_config_mujoco(params):
    """Print the current configuration in a readable format."""
    print("=" * 60)
    print(f"EXPERIMENT CONFIGURATION of {params['run_ID']}")
    print("=" * 60)

    print(f"\nEnvironment Settings:")
    print(f"  Environment: {params['env_id']}")
    print(f"  Episode length: {params['episode_length']}")
    print(f"  Embedding function: {params['embedding_name']}")

    print(f"\nExperiment Settings:")
    print(f"  No. of seeds: {params['N_experiments']}")
    print(f"  Iterations per seed: {params['N_iterations']}")
    print(f"  No. queried preferences per iteration: {params['N_rollouts']}")
    print(f"  Offline trajectories: {params['N_offline_trajs']}")
    print(f"  Offline confset size: {params['N_confset_size']}")
    print(f"  Offline confset noise: {params['confset_noise']}")

    print(f"\nTraining Settings:")
    print(f"  PPO | arch: [{params['hidden_dim']} x2]")
    print(f"  BC  | epochs: {params['n_bc_epochs']}")
    print(f"  BC  | loss: {params['bc_loss']}")
    print(f"  w   | epochs: {params['w_epochs']}")
    print(f"  w   | trainfunc: {params['w_trainfunc']}")
    print(f"  w   | regularization: {params['w_regularization']}")
    print(f"  w   | initialization: {params['w_initialization']}")
    print(f"  w   | sigmoid slope: {params['w_sigmoid_slope']}")
    print(f"  w   | bound W: {params['W']}")
    print(f"  w   | project: {params['project_w']}")
    print(f"  w   | retrain from scratch: {params['retrain_w_from_scratch']}")

    print(f"\nOnline Learning Settings:")
    print(f"  V   | initialization: {params['V_init']}")
    print(f"  ϕ(π)| samples: {params['n_embedding_samples']}")
    print(f"  policy selection: {params['which_policy_selection']}")

    if params["verbose"]:
        print(f"\nVerbosity: {', '.join(params['verbose'])}")

    print(f"\nRun Settings:")
    print(f"  Run baseline: {params['run_baseline']}")
    print(f"  Run bridge: {params['run_bridge']}")
    print(f"  Save results: {params['save_results']}")
    print(f"  Run ID: {params['run_ID']}")
    print(f"  Loaded run behaviour: {params['loaded_run_behaviour']}")
    print(f"  Regret plot: {params['which_plot_subopt']}")
    print(f"  Plot scores: {params['plot_scores']}")


# ====== LOGGING ======
# TODO: add a seed prefix for subprocesses. either a class that inherits, or via self.seed={None, 1,2,3...}
class Tee:
    """Tee class for logging to both file and console."""

    def __init__(self, file_path, mode="w"):
        self.terminal = sys.stdout
        self.log_file = open(file_path, mode)

    def write(self, message):
        try:
            self.terminal.write(message)
            self.log_file.write(message)
            self.log_file.flush()  # Ensure immediate writing to file
        except Exception:
            pass  # errors shouldnt crash experiment

    def flush(self):
        self.terminal.flush()
        self.log_file.flush()

    def close(self):
        self.log_file.close()


def plot_dists_rewards(dists_x, rewards, epsilon, title=None):
    fig_hist = plt.figure(figsize=(8, 6))
    plt.hist(dists_x)
    plt.axvline(epsilon, color="red", linestyle="--", label=f"epsilon = {epsilon}")
    plt.show()
    plt.close()

    num_accepted = 0
    for dist in dists_x:
        if dist < epsilon:
            num_accepted += 1

    basetitle = "Distance to BC Policy vs Average Reward"

    if title:
        basetitle += f"\n{title}"
    fig_scatter = plt.figure(figsize=(8, 6))
    plt.scatter(dists_x, rewards, alpha=0.6)
    plt.scatter(dists_x[0], rewards[0], color="green", s=100, alpha=0.8, label="BC Policy")
    plt.scatter(dists_x[-1], rewards[-1], color="red", s=100, alpha=0.8, label="Expert Policy")
    plt.xlabel("Distance to BC Policy")
    plt.ylabel("Average Reward")
    plt.title(basetitle)
    plt.axvline(epsilon, color="red", linestyle="--", label=f"epsilon = {epsilon}")
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.show()
    plt.close()

    print(
        f"acc.: {num_accepted}/{len(dists_x)}, skipped: {len(dists_x) - num_accepted}/{len(dists_x)}"
    )
    return fig_hist, fig_scatter
