Skip to content

hover.core.neural

  • Neural network components.

    torch-based template classes for implementing neural nets that work the most smoothly with hover.

    BaseVectorNet (Loggable)

    Abstract transfer learning model defining common signatures.

    Intended to define crucial interactions with built-in recipes like hover.recipes.active_learning().

    Source code in hover/core/neural.py
    class BaseVectorNet(Loggable):
    
        """
        ???+ note "Abstract transfer learning model defining common signatures."
    
            Intended to define crucial interactions with built-in recipes like `hover.recipes.active_learning()`.
        """
    
        @abstractmethod
        def predict_proba(self, inps):
            pass
    
        @abstractmethod
        def manifold_trajectory(
            self, inps, method=None, reducer_kwargs=None, spline_kwargs=None
        ):
            pass
    
        @abstractmethod
        def prepare_loader(self, dataset, key, **kwargs):
            pass
    
        @abstractmethod
        def train(self, train_loader, dev_loader=None, epochs=None, **kwargs):
            pass
    

    VectorNet (BaseVectorNet)

    Simple transfer learning model: a user-supplied vectorizer followed by a neural net.

    This is a parent class whose children may use different training schemes.

    Coupled with:

    • hover.utils.torch_helper.VectorDataset
    Source code in hover/core/neural.py
    class VectorNet(BaseVectorNet):
    
        """
        ???+ note "Simple transfer learning model: a user-supplied vectorizer followed by a neural net."
            This is a parent class whose children may use different training schemes.
    
            Coupled with:
    
            -   `hover.utils.torch_helper.VectorDataset`
        """
    
        DEFAULT_OPTIM_CLS = torch.optim.Adam
        DEFAULT_OPTIM_LOGLR = 2.0
        DEFAULT_OPTIM_KWARGS = {"lr": 0.1**DEFAULT_OPTIM_LOGLR, "betas": (0.9, 0.999)}
    
        def __init__(
            self,
            vectorizer,
            architecture,
            state_dict_path,
            labels,
            backup_state_dict=True,
            optimizer_cls=None,
            optimizer_kwargs=None,
            verbose=0,
            example_input="",
        ):
            """
            ???+ note "Create the `VectorNet`, loading parameters if available."
    
                | Param             | Type       | Description                          |
                | :---------------- | :--------- | :----------------------------------- |
                | `vectorizer`      | `callable` | the feature -> vector function       |
                | `architecture`    | `class`    | a `torch.nn.Module` child class      |
                | `state_dict_path` | `str`      | path to a (could-be-empty) `torch` state dict |
                | `labels`          | `list`     | list of `str` classification labels  |
                | `backup_state_dict` | `bool`   | whether to backup the loaded state dict |
                | `optimizer_cls`   | `subclass of torch.optim.Optimizer` | pytorch optimizer class |
                | `optimizer_kwargs`  | `dict`   | pytorch optimizer kwargs             |
                | `verbose`         | `int`      | logging verbosity level              |
                | `example_input`   | any        | example input to the vectorizer      |
            """
    
            assert isinstance(
                verbose, int
            ), f"Expected verbose as int, got {type(verbose)} {verbose}"
            self.verbose = verbose
            self.vectorizer = vectorizer
            self.example_input = example_input
            self.architecture = architecture
            self.setup_label_conversion(labels)
            self._dynamic_params = {}
    
            # set a path to store updated parameters
            self.nn_update_path = state_dict_path
    
            if backup_state_dict and os.path.isfile(state_dict_path):
                state_dict_backup_path = f"{state_dict_path}.{current_time('%Y%m%d%H%M%S')}"
                copyfile(state_dict_path, state_dict_backup_path)
    
            # initialize an optimizer object and a dict to hold dynamic parameters
            optimizer_cls = optimizer_cls or self.__class__.DEFAULT_OPTIM_CLS
            optimizer_kwargs = (
                optimizer_kwargs or self.__class__.DEFAULT_OPTIM_KWARGS.copy()
            )
    
            def callback_reset_nn_optimizer():
                """
                Callback function which has access to optimizer init settings.
                """
                self.nn_optimizer = optimizer_cls(self.nn.parameters())
                assert isinstance(
                    self.nn_optimizer, torch.optim.Optimizer
                ), f"Expected an optimizer, got {type(self.nn_optimizer)}"
                self._dynamic_params["optimizer"] = optimizer_kwargs
    
            self._callback_reset_nn_optimizer = callback_reset_nn_optimizer
            self.setup_nn(use_existing_state_dict=True)
            self._setup_widgets()
    
        def auto_adjust_setup(self, labels, auto_skip=True):
            """
            ???+ note "Auto-(re)create label encoder/decoder and neural net."
    
                Intended to be called in and out of the constructor.
    
                | Param             | Type       | Description                          |
                | :---------------- | :--------- | :----------------------------------- |
                | `labels`          | `list`     | list of `str` classification labels  |
                | `auto_skip`       | `bool`     | skip when labels did not change      |
            """
            # sanity check and skip
            assert isinstance(labels, list), f"Expected a list of labels, got {labels}"
            # if the sequence of labels matches label encoder exactly, skip
            label_match_flag = labels == sorted(
                self.label_encoder.keys(), key=lambda k: self.label_encoder[k]
            )
            if auto_skip and label_match_flag:
                return
    
            self.setup_label_conversion(labels)
            self.setup_nn(use_existing_state_dict=False)
    
            self._good(f"adjusted to new list of labels: {labels}")
    
        def setup_label_conversion(self, labels):
            """
            ???+ note "Set up label encoder/decoder and number of classes."
    
                | Param             | Type       | Description                          |
                | :---------------- | :--------- | :----------------------------------- |
                | `labels`          | `list`     | list of `str` classification labels  |
            """
            self.label_encoder = {_label: i for i, _label in enumerate(labels)}
            self.label_decoder = {i: _label for i, _label in enumerate(labels)}
            self.num_classes = len(self.label_encoder)
    
        def setup_nn(self, use_existing_state_dict=True):
            """
            ???+ note "Set up neural network and optimizers."
    
                Intended to be called in and out of the constructor.
    
                -   will try to load parameters from state dict by default
                -   option to override and discard previous state dict
                    -   often used when the classification targets have changed
    
                | Param                     | Type       | Description                          |
                | :------------------------ | :--------- | :----------------------------------- |
                | `labels`                  | `list`     | list of `str` classification labels  |
                | `use_existing_state_dict` | `bool`     | whether to use existing state dict   |
            """
            # set up vectorizer and the neural network with appropriate dimensions
            vec_dim = self.vectorizer(self.example_input).shape[0]
            self.nn = self.architecture(vec_dim, self.num_classes)
            self._callback_reset_nn_optimizer()
    
            state_dict_exists = os.path.isfile(self.nn_update_path)
            # if state dict exists, load it (when consistent) or overwrite
            if state_dict_exists:
                if use_existing_state_dict:
                    self.load(self.nn_update_path)
                else:
                    self.save(self.nn_update_path)
    
            self._good(f"reset neural net: in {vec_dim} out {self.num_classes}.")
    
        def load(self, load_path=None):
            """
            ???+ note "Load neural net parameters if possible."
    
                Can be directed to a custom state dict.
    
                | Param       | Type       | Description                  |
                | :---------- | :--------- | :--------------------------- |
                | `load_path` | `str`      | path to a `torch` state dict |
            """
            load_path = load_path or self.nn_update_path
            # if the architecture cannot match the state dict, skip the load and warn
            try:
                self.nn.load_state_dict(torch.load(load_path))
                self._info(f"loaded state dict {load_path}.")
            except Exception as e:
                self._warn(f"load VectorNet state path failed with {type(e)}: {e}")
    
        @classmethod
        def from_module(cls, model_module, labels, **kwargs):
            """
            ???+ note "Create a VectorNet model from a loadable module."
    
                | Param          | Type       | Description                          |
                | :------------- | :--------- | :----------------------------------- |
                | `model_module` | `module` or `str` | (path to) a local Python workspace module which contains a get_vectorizer() callable, get_architecture() callable, and a get_state_dict_path() callable |
                | `labels`       | `list`     | list of `str` classification labels  |
                | `**kwargs`     |      | forwarded to `self.__init__()` constructor |
            """
            if isinstance(model_module, str):
                from importlib import import_module
    
                model_module = import_module(model_module)
    
            # Load the model by retrieving the inp-to-vec function, architecture, and state dict
            model = cls(
                model_module.get_vectorizer(),
                model_module.get_architecture(),
                model_module.get_state_dict_path(),
                labels,
                **kwargs,
            )
    
            return model
    
        def save(self, save_path=None):
            """
            ???+ note "Save the current state dict with authorization to overwrite."
                | Param       | Type  | Description                           |
                | :---------- | :---- | :------------------------------------ |
                | `save_path` | `str` | option alternative path to state dict |
            """
            save_path = save_path or self.nn_update_path
            torch.save(self.nn.state_dict(), save_path)
            verb = "overwrote" if os.path.isfile(save_path) else "saved"
            self._info(f"{verb} state dict {save_path}.")
    
        def _setup_widgets(self):
            """
            ???+ note "Bokeh widgets for changing hyperparameters through user interaction."
            """
            self.epochs_slider = Slider(start=1, end=50, value=1, step=1, title="# epochs")
            self.loglr_slider = Slider(
                title="learning rate",
                start=1.0,
                end=7.0,
                value=self.__class__.DEFAULT_OPTIM_LOGLR,
                step=0.1,
                format=CustomJSTickFormatter(code="return Math.pow(0.1, tick).toFixed(8)"),
            )
    
            def update_lr(attr, old, new):
                self._dynamic_params["optimizer"]["lr"] = 0.1**self.loglr_slider.value
    
            self.loglr_slider.on_change("value", update_lr)
    
        def _layout_widgets(self):
            """
            ???+ note "Layout of widgets when plotted."
            """
            from bokeh.layouts import row
    
            return row(self.epochs_slider, self.loglr_slider)
    
        def view(self):
            """
            ???+ note "Overall layout when plotted."
            """
            return self._layout_widgets()
    
        def adjust_optimizer_params(self):
            """
            ???+ note "Dynamically change parameters of the neural net optimizer."
    
                - Intended to be polymorphic in child classes and to be called per epoch.
            """
            for _group in self.nn_optimizer.param_groups:
                _group.update(self._dynamic_params["optimizer"])
    
        def predict_proba(self, inps):
            """
            ???+ note "End-to-end single/multi-piece prediction from inp to class probabilities."
                | Param  | Type    | Description                          |
                | :----- | :------ | :----------------------------------- |
                | `inps` | dynamic | (a list of) input features to vectorize |
            """
            # if the input is a single piece of inp, cast it to a list
            FLAG_SINGLE = not isinstance(inps, list)
            if FLAG_SINGLE:
                inps = [inps]
    
            # the actual prediction
            self.nn.eval()
            vectors = torch.Tensor(np.array([self.vectorizer(_inp) for _inp in inps]))
            logits = self.nn(vectors)
            probs = F.softmax(logits, dim=-1).detach().numpy()
    
            # inverse-cast if applicable
            if FLAG_SINGLE:
                probs = probs[0]
    
            return probs
    
        def manifold_trajectory(
            self, inps, method=None, reducer_kwargs=None, spline_kwargs=None
        ):
            """
            ???+ note "Compute a propagation trajectory of the dataset manifold through the neural net."
    
                1. vectorize inps
                2. forward propagate, keeping intermediates
                3. fit intermediates to N-D manifolds
                4. fit manifolds using Procrustes shape analysis
                5. fit shapes to trajectory splines
    
                | Param    | Type    | Description                          |
                | :------- | :------ | :----------------------------------- |
                | `inps`   | dynamic | (a list of) input features to vectorize |
                | `method` | `str`   | reduction method: `"umap"` or `"ivis"`  |
                | `reducer_kwargs` | | kwargs to forward to dimensionality reduction |
                | `spline_kwargs` | | kwargs to forward to spline calculation |
            """
            from hover.core.representation.manifold import LayerwiseManifold
            from hover.core.representation.trajectory import manifold_spline
    
            if method is None:
                method = hover.config["data.embedding"]["default_reduction_method"]
    
            reducer_kwargs = reducer_kwargs or {}
            spline_kwargs = spline_kwargs or {}
    
            # step 1 & 2
            vectors = torch.Tensor(np.array([self.vectorizer(_inp) for _inp in inps]))
            self.nn.eval()
            intermediates = self.nn.eval_per_layer(vectors)
            intermediates = [_tensor.detach().numpy() for _tensor in intermediates]
    
            # step 3 & 4
            LM = LayerwiseManifold(intermediates)
            LM.unfold(method=method, **reducer_kwargs)
            seq_arr, disparities = LM.procrustes()
            seq_arr = np.array(seq_arr)
    
            # step 5
            traj_arr = manifold_spline(np.array(seq_arr), **spline_kwargs)
    
            return traj_arr, seq_arr, disparities
    
        def prepare_loader(self, dataset, key, **kwargs):
            """
            ???+ note "Create dataloader from `SupervisableDataset` with implied vectorizer(s)."
    
                | Param      | Type  | Description                |
                | :--------- | :---- | :------------------------- |
                | `dataset`  | `hover.core.dataset.SupervisableDataset` | the dataset to load |
                | `key`      | `str` | "train", "dev", or "test"  |
                | `**kwargs` | | forwarded to `dataset.loader()`  |
            """
            return dataset.loader(key, self.vectorizer, **kwargs)
    
        def train(self, train_loader, dev_loader=None, epochs=None):
            """
            ???+ note "Train the neural network part of the VecNet."
    
                - intended to be coupled with self.train_batch().
    
                | Param          | Type         | Description                |
                | :------------- | :----------- | :------------------------- |
                | `train_loader` | `torch.utils.data.DataLoader` | train set |
                | `dev_loader`   | `torch.utils.data.DataLoader` | dev set   |
                | `epochs`       | `int`        | number of epochs to train  |
            """
            epochs = epochs or self.epochs_slider.value
    
            train_info = []
            for epoch_idx in range(epochs):
                self._dynamic_params["epoch"] = epoch_idx + 1
                self.train_epoch(train_loader)
                if dev_loader is not None:
                    dev_loader = train_loader
                acc, conf_mat = self.evaluate(dev_loader)
                train_info.append({"accuracy": acc, "confusion_matrix": conf_mat})
            return train_info
    
        def train_epoch(self, train_loader, *args, **kwargs):
            """
            ???+ note "Train the neural network for one epoch."
    
                - Supports flexible args and kwargs for child classes that may implement self.train() and self.train_batch() differently.
    
                | Param          | Type         | Description                |
                | :------------- | :----------- | :------------------------- |
                | `train_loader` | `torch.utils.data.DataLoader` | train set |
                | `*args`        | | arguments to forward to `train_batch`   |
                | `**kwargs`     | | kwargs to forward to `train_batch`      |
            """
            self.adjust_optimizer_params()
            for batch_idx, (loaded_input, loaded_output, _) in enumerate(train_loader):
                self._dynamic_params["batch"] = batch_idx + 1
                self.train_batch(loaded_input, loaded_output, *args, **kwargs)
    
        def train_batch(self, loaded_input, loaded_output):
            """
            ???+ note "Train the neural network for one batch."
    
                | Param           | Type           | Description           |
                | :-------------- | :------------- | :-------------------- |
                | `loaded_input`  | `torch.Tensor` | input tensor          |
                | `loaded_output` | `torch.Tensor` | output tensor         |
            """
            self.nn.train()
            input_tensor = loaded_input.float()
            output_tensor = loaded_output.float()
    
            # compute logits
            logits = self.nn(input_tensor)
            loss = F.cross_entropy(logits, output_tensor)
    
            self.nn_optimizer.zero_grad()
            loss.backward()
            self.nn_optimizer.step()
    
            if self.verbose > 0:
                log_info = dict(self._dynamic_params)
                log_info["performance"] = "Loss {0:.3f}".format(loss)
                self._print(
                    "{0: <80}".format(
                        "Train: Epoch {epoch} Batch {batch} {performance}".format(
                            **log_info
                        )
                    ),
                    end="\r",
                )
    
        def evaluate(self, dev_loader):
            """
            ???+ note "Evaluate the VecNet against a dev set."
    
                | Param        | Type         | Description                |
                | :----------- | :----------- | :------------------------- |
                | `dev_loader` | `torch.utils.data.DataLoader` | dev set   |
            """
            self.nn.eval()
            true = []
            pred = []
            for loaded_input, loaded_output, _idx in dev_loader:
                _input_tensor = loaded_input.float()
                _output_tensor = loaded_output.float()
    
                _logits = self.nn(_input_tensor)
                _true_batch = _output_tensor.argmax(dim=1).detach().numpy()
                _pred_batch = F.softmax(_logits, dim=1).argmax(dim=1).detach().numpy()
                true.append(_true_batch)
                pred.append(_pred_batch)
            true = np.concatenate(true)
            pred = np.concatenate(pred)
            accuracy = classification_accuracy(true, pred)
            conf_mat = confusion_matrix(true, pred)
    
            if self.verbose >= 0:
                log_info = dict(self._dynamic_params)
                log_info["performance"] = "Acc {0:.3f}".format(accuracy)
                self._info(
                    "{0: <80}".format(
                        "Eval: Epoch {epoch} {performance}".format(**log_info)
                    )
                )
    
            return accuracy, conf_mat
    

    DEFAULT_OPTIM_CLS (Optimizer)

    Implements Adam algorithm.

    .. math:: \begin{aligned} &\rule{110mm}{0.4pt} \ &\textbf{input} : \gamma \text{ (lr)}, \beta_1, \beta_2 \text{ (betas)},\theta_0 \text{ (params)},f(\theta) \text{ (objective)} \ &\hspace{13mm} \lambda \text{ (weight decay)}, : amsgrad \ &\textbf{initialize} : m_0 \leftarrow 0 \text{ ( first moment)}, v_0\leftarrow 0 \text{ (second moment)},: \widehat{v_0}^{max}\leftarrow 0\[-1.ex] &\rule{110mm}{0.4pt} \ &\textbf{for} : t=1 : \textbf{to} : \ldots : \textbf{do} \ &\hspace{5mm}g_t \leftarrow \nabla_{\theta} f_t (\theta_{t-1}) \ &\hspace{5mm}\textbf{if} : \lambda \neq 0 \ &\hspace{10mm} g_t \leftarrow g_t + \lambda \theta_{t-1} \ &\hspace{5mm}m_t \leftarrow \beta_1 m_{t-1} + (1 - \beta_1) g_t \ &\hspace{5mm}v_t \leftarrow \beta_2 v_{t-1} + (1-\beta_2) g^2_t \ &\hspace{5mm}\widehat{m_t} \leftarrow m_t/\big(1-\beta_1^t \big) \ &\hspace{5mm}\widehat{v_t} \leftarrow v_t/\big(1-\beta_2^t \big) \ &\hspace{5mm}\textbf{if} : amsgrad \ &\hspace{10mm}\widehat{v_t}^{max} \leftarrow \mathrm{max}(\widehat{v_t}^{max}, \widehat{v_t}) \ &\hspace{10mm}\theta_t \leftarrow \theta_{t-1} - \gamma \widehat{m_t}/ \big(\sqrt{\widehat{v_t}^{max}} + \epsilon \big) \ &\hspace{5mm}\textbf{else} \ &\hspace{10mm}\theta_t \leftarrow \theta_{t-1} - \gamma \widehat{m_t}/ \big(\sqrt{\widehat{v_t}} + \epsilon \big) \ &\rule{110mm}{0.4pt} \[-1.ex] &\bf{return} : \theta_t \[-1.ex] &\rule{110mm}{0.4pt} \[-1.ex] \end{aligned}

    For further details regarding the algorithm we refer to Adam: A Method for Stochastic Optimization_.

    Parameters:

    Name Type Description Default
    params iterable

    iterable of parameters to optimize or dicts defining parameter groups

    required
    lr float

    learning rate (default: 1e-3)

    0.001
    betas Tuple[float, float]

    coefficients used for computing running averages of gradient and its square (default: (0.9, 0.999))

    (0.9, 0.999)
    eps float

    term added to the denominator to improve numerical stability (default: 1e-8)

    1e-08
    weight_decay float

    weight decay (L2 penalty) (default: 0)

    0
    amsgrad boolean

    whether to use the AMSGrad variant of this algorithm from the paper On the Convergence of Adam and Beyond_ (default: False)

    False

    .. _Adam: A Method for Stochastic Optimization: https://arxiv.org/abs/1412.6980 .. _On the Convergence of Adam and Beyond: https://openreview.net/forum?id=ryQu7f-RZ

    Source code in hover/core/neural.py
    class Adam(Optimizer):
        r"""Implements Adam algorithm.
    
        .. math::
           \begin{aligned}
                &\rule{110mm}{0.4pt}                                                                 \\
                &\textbf{input}      : \gamma \text{ (lr)}, \beta_1, \beta_2
                    \text{ (betas)},\theta_0 \text{ (params)},f(\theta) \text{ (objective)}          \\
                &\hspace{13mm}      \lambda \text{ (weight decay)},  \: amsgrad                      \\
                &\textbf{initialize} :  m_0 \leftarrow 0 \text{ ( first moment)},
                    v_0\leftarrow 0 \text{ (second moment)},\: \widehat{v_0}^{max}\leftarrow 0\\[-1.ex]
                &\rule{110mm}{0.4pt}                                                                 \\
                &\textbf{for} \: t=1 \: \textbf{to} \: \ldots \: \textbf{do}                         \\
                &\hspace{5mm}g_t           \leftarrow   \nabla_{\theta} f_t (\theta_{t-1})           \\
                &\hspace{5mm}\textbf{if} \: \lambda \neq 0                                           \\
                &\hspace{10mm} g_t \leftarrow g_t + \lambda  \theta_{t-1}                            \\
                &\hspace{5mm}m_t           \leftarrow   \beta_1 m_{t-1} + (1 - \beta_1) g_t          \\
                &\hspace{5mm}v_t           \leftarrow   \beta_2 v_{t-1} + (1-\beta_2) g^2_t          \\
                &\hspace{5mm}\widehat{m_t} \leftarrow   m_t/\big(1-\beta_1^t \big)                   \\
                &\hspace{5mm}\widehat{v_t} \leftarrow   v_t/\big(1-\beta_2^t \big)                   \\
                &\hspace{5mm}\textbf{if} \: amsgrad                                                  \\
                &\hspace{10mm}\widehat{v_t}^{max} \leftarrow \mathrm{max}(\widehat{v_t}^{max},
                    \widehat{v_t})                                                                   \\
                &\hspace{10mm}\theta_t \leftarrow \theta_{t-1} - \gamma \widehat{m_t}/
                    \big(\sqrt{\widehat{v_t}^{max}} + \epsilon \big)                                 \\
                &\hspace{5mm}\textbf{else}                                                           \\
                &\hspace{10mm}\theta_t \leftarrow \theta_{t-1} - \gamma \widehat{m_t}/
                    \big(\sqrt{\widehat{v_t}} + \epsilon \big)                                       \\
                &\rule{110mm}{0.4pt}                                                          \\[-1.ex]
                &\bf{return} \:  \theta_t                                                     \\[-1.ex]
                &\rule{110mm}{0.4pt}                                                          \\[-1.ex]
           \end{aligned}
    
        For further details regarding the algorithm we refer to `Adam: A Method for Stochastic Optimization`_.
    
        Args:
            params (iterable): iterable of parameters to optimize or dicts defining
                parameter groups
            lr (float, optional): learning rate (default: 1e-3)
            betas (Tuple[float, float], optional): coefficients used for computing
                running averages of gradient and its square (default: (0.9, 0.999))
            eps (float, optional): term added to the denominator to improve
                numerical stability (default: 1e-8)
            weight_decay (float, optional): weight decay (L2 penalty) (default: 0)
            amsgrad (boolean, optional): whether to use the AMSGrad variant of this
                algorithm from the paper `On the Convergence of Adam and Beyond`_
                (default: False)
    
        .. _Adam\: A Method for Stochastic Optimization:
            https://arxiv.org/abs/1412.6980
        .. _On the Convergence of Adam and Beyond:
            https://openreview.net/forum?id=ryQu7f-RZ
        """
    
        def __init__(self, params, lr=1e-3, betas=(0.9, 0.999), eps=1e-8,
                     weight_decay=0, amsgrad=False):
            if not 0.0 <= lr:
                raise ValueError("Invalid learning rate: {}".format(lr))
            if not 0.0 <= eps:
                raise ValueError("Invalid epsilon value: {}".format(eps))
            if not 0.0 <= betas[0] < 1.0:
                raise ValueError("Invalid beta parameter at index 0: {}".format(betas[0]))
            if not 0.0 <= betas[1] < 1.0:
                raise ValueError("Invalid beta parameter at index 1: {}".format(betas[1]))
            if not 0.0 <= weight_decay:
                raise ValueError("Invalid weight_decay value: {}".format(weight_decay))
            defaults = dict(lr=lr, betas=betas, eps=eps,
                            weight_decay=weight_decay, amsgrad=amsgrad)
            super(Adam, self).__init__(params, defaults)
    
        def __setstate__(self, state):
            super(Adam, self).__setstate__(state)
            for group in self.param_groups:
                group.setdefault('amsgrad', False)
    
        @torch.no_grad()
        def step(self, closure=None):
            """Performs a single optimization step.
    
            Args:
                closure (callable, optional): A closure that reevaluates the model
                    and returns the loss.
            """
            loss = None
            if closure is not None:
                with torch.enable_grad():
                    loss = closure()
    
            for group in self.param_groups:
                params_with_grad = []
                grads = []
                exp_avgs = []
                exp_avg_sqs = []
                max_exp_avg_sqs = []
                state_steps = []
                beta1, beta2 = group['betas']
    
                for p in group['params']:
                    if p.grad is not None:
                        params_with_grad.append(p)
                        if p.grad.is_sparse:
                            raise RuntimeError('Adam does not support sparse gradients, please consider SparseAdam instead')
                        grads.append(p.grad)
    
                        state = self.state[p]
                        # Lazy state initialization
                        if len(state) == 0:
                            state['step'] = 0
                            # Exponential moving average of gradient values
                            state['exp_avg'] = torch.zeros_like(p, memory_format=torch.preserve_format)
                            # Exponential moving average of squared gradient values
                            state['exp_avg_sq'] = torch.zeros_like(p, memory_format=torch.preserve_format)
                            if group['amsgrad']:
                                # Maintains max of all exp. moving avg. of sq. grad. values
                                state['max_exp_avg_sq'] = torch.zeros_like(p, memory_format=torch.preserve_format)
    
                        exp_avgs.append(state['exp_avg'])
                        exp_avg_sqs.append(state['exp_avg_sq'])
    
                        if group['amsgrad']:
                            max_exp_avg_sqs.append(state['max_exp_avg_sq'])
    
                        # update the steps for each param group update
                        state['step'] += 1
                        # record the step after step update
                        state_steps.append(state['step'])
    
                F.adam(params_with_grad,
                       grads,
                       exp_avgs,
                       exp_avg_sqs,
                       max_exp_avg_sqs,
                       state_steps,
                       amsgrad=group['amsgrad'],
                       beta1=beta1,
                       beta2=beta2,
                       lr=group['lr'],
                       weight_decay=group['weight_decay'],
                       eps=group['eps'])
            return loss
    

    step(self, closure=None)

    Performs a single optimization step.

    Parameters:

    Name Type Description Default
    closure callable

    A closure that reevaluates the model and returns the loss.

    None
    Source code in hover/core/neural.py
    @torch.no_grad()
    def step(self, closure=None):
        """Performs a single optimization step.
    
        Args:
            closure (callable, optional): A closure that reevaluates the model
                and returns the loss.
        """
        loss = None
        if closure is not None:
            with torch.enable_grad():
                loss = closure()
    
        for group in self.param_groups:
            params_with_grad = []
            grads = []
            exp_avgs = []
            exp_avg_sqs = []
            max_exp_avg_sqs = []
            state_steps = []
            beta1, beta2 = group['betas']
    
            for p in group['params']:
                if p.grad is not None:
                    params_with_grad.append(p)
                    if p.grad.is_sparse:
                        raise RuntimeError('Adam does not support sparse gradients, please consider SparseAdam instead')
                    grads.append(p.grad)
    
                    state = self.state[p]
                    # Lazy state initialization
                    if len(state) == 0:
                        state['step'] = 0
                        # Exponential moving average of gradient values
                        state['exp_avg'] = torch.zeros_like(p, memory_format=torch.preserve_format)
                        # Exponential moving average of squared gradient values
                        state['exp_avg_sq'] = torch.zeros_like(p, memory_format=torch.preserve_format)
                        if group['amsgrad']:
                            # Maintains max of all exp. moving avg. of sq. grad. values
                            state['max_exp_avg_sq'] = torch.zeros_like(p, memory_format=torch.preserve_format)
    
                    exp_avgs.append(state['exp_avg'])
                    exp_avg_sqs.append(state['exp_avg_sq'])
    
                    if group['amsgrad']:
                        max_exp_avg_sqs.append(state['max_exp_avg_sq'])
    
                    # update the steps for each param group update
                    state['step'] += 1
                    # record the step after step update
                    state_steps.append(state['step'])
    
            F.adam(params_with_grad,
                   grads,
                   exp_avgs,
                   exp_avg_sqs,
                   max_exp_avg_sqs,
                   state_steps,
                   amsgrad=group['amsgrad'],
                   beta1=beta1,
                   beta2=beta2,
                   lr=group['lr'],
                   weight_decay=group['weight_decay'],
                   eps=group['eps'])
        return loss
    

    __init__(self, vectorizer, architecture, state_dict_path, labels, backup_state_dict=True, optimizer_cls=None, optimizer_kwargs=None, verbose=0, example_input='') special

    Create the VectorNet, loading parameters if available.
    Param Type Description
    vectorizer callable the feature -> vector function
    architecture class a torch.nn.Module child class
    state_dict_path str path to a (could-be-empty) torch state dict
    labels list list of str classification labels
    backup_state_dict bool whether to backup the loaded state dict
    optimizer_cls subclass of torch.optim.Optimizer pytorch optimizer class
    optimizer_kwargs dict pytorch optimizer kwargs
    verbose int logging verbosity level
    example_input any example input to the vectorizer
    Source code in hover/core/neural.py
    def __init__(
        self,
        vectorizer,
        architecture,
        state_dict_path,
        labels,
        backup_state_dict=True,
        optimizer_cls=None,
        optimizer_kwargs=None,
        verbose=0,
        example_input="",
    ):
        """
        ???+ note "Create the `VectorNet`, loading parameters if available."
    
            | Param             | Type       | Description                          |
            | :---------------- | :--------- | :----------------------------------- |
            | `vectorizer`      | `callable` | the feature -> vector function       |
            | `architecture`    | `class`    | a `torch.nn.Module` child class      |
            | `state_dict_path` | `str`      | path to a (could-be-empty) `torch` state dict |
            | `labels`          | `list`     | list of `str` classification labels  |
            | `backup_state_dict` | `bool`   | whether to backup the loaded state dict |
            | `optimizer_cls`   | `subclass of torch.optim.Optimizer` | pytorch optimizer class |
            | `optimizer_kwargs`  | `dict`   | pytorch optimizer kwargs             |
            | `verbose`         | `int`      | logging verbosity level              |
            | `example_input`   | any        | example input to the vectorizer      |
        """
    
        assert isinstance(
            verbose, int
        ), f"Expected verbose as int, got {type(verbose)} {verbose}"
        self.verbose = verbose
        self.vectorizer = vectorizer
        self.example_input = example_input
        self.architecture = architecture
        self.setup_label_conversion(labels)
        self._dynamic_params = {}
    
        # set a path to store updated parameters
        self.nn_update_path = state_dict_path
    
        if backup_state_dict and os.path.isfile(state_dict_path):
            state_dict_backup_path = f"{state_dict_path}.{current_time('%Y%m%d%H%M%S')}"
            copyfile(state_dict_path, state_dict_backup_path)
    
        # initialize an optimizer object and a dict to hold dynamic parameters
        optimizer_cls = optimizer_cls or self.__class__.DEFAULT_OPTIM_CLS
        optimizer_kwargs = (
            optimizer_kwargs or self.__class__.DEFAULT_OPTIM_KWARGS.copy()
        )
    
        def callback_reset_nn_optimizer():
            """
            Callback function which has access to optimizer init settings.
            """
            self.nn_optimizer = optimizer_cls(self.nn.parameters())
            assert isinstance(
                self.nn_optimizer, torch.optim.Optimizer
            ), f"Expected an optimizer, got {type(self.nn_optimizer)}"
            self._dynamic_params["optimizer"] = optimizer_kwargs
    
        self._callback_reset_nn_optimizer = callback_reset_nn_optimizer
        self.setup_nn(use_existing_state_dict=True)
        self._setup_widgets()
    

    adjust_optimizer_params(self)

    Dynamically change parameters of the neural net optimizer.
    • Intended to be polymorphic in child classes and to be called per epoch.
    Source code in hover/core/neural.py
    def adjust_optimizer_params(self):
        """
        ???+ note "Dynamically change parameters of the neural net optimizer."
    
            - Intended to be polymorphic in child classes and to be called per epoch.
        """
        for _group in self.nn_optimizer.param_groups:
            _group.update(self._dynamic_params["optimizer"])
    

    auto_adjust_setup(self, labels, auto_skip=True)

    Auto-(re)create label encoder/decoder and neural net.

    Intended to be called in and out of the constructor.

    Param Type Description
    labels list list of str classification labels
    auto_skip bool skip when labels did not change
    Source code in hover/core/neural.py
    def auto_adjust_setup(self, labels, auto_skip=True):
        """
        ???+ note "Auto-(re)create label encoder/decoder and neural net."
    
            Intended to be called in and out of the constructor.
    
            | Param             | Type       | Description                          |
            | :---------------- | :--------- | :----------------------------------- |
            | `labels`          | `list`     | list of `str` classification labels  |
            | `auto_skip`       | `bool`     | skip when labels did not change      |
        """
        # sanity check and skip
        assert isinstance(labels, list), f"Expected a list of labels, got {labels}"
        # if the sequence of labels matches label encoder exactly, skip
        label_match_flag = labels == sorted(
            self.label_encoder.keys(), key=lambda k: self.label_encoder[k]
        )
        if auto_skip and label_match_flag:
            return
    
        self.setup_label_conversion(labels)
        self.setup_nn(use_existing_state_dict=False)
    
        self._good(f"adjusted to new list of labels: {labels}")
    

    evaluate(self, dev_loader)

    Evaluate the VecNet against a dev set.
    Param Type Description
    dev_loader torch.utils.data.DataLoader dev set
    Source code in hover/core/neural.py
    def evaluate(self, dev_loader):
        """
        ???+ note "Evaluate the VecNet against a dev set."
    
            | Param        | Type         | Description                |
            | :----------- | :----------- | :------------------------- |
            | `dev_loader` | `torch.utils.data.DataLoader` | dev set   |
        """
        self.nn.eval()
        true = []
        pred = []
        for loaded_input, loaded_output, _idx in dev_loader:
            _input_tensor = loaded_input.float()
            _output_tensor = loaded_output.float()
    
            _logits = self.nn(_input_tensor)
            _true_batch = _output_tensor.argmax(dim=1).detach().numpy()
            _pred_batch = F.softmax(_logits, dim=1).argmax(dim=1).detach().numpy()
            true.append(_true_batch)
            pred.append(_pred_batch)
        true = np.concatenate(true)
        pred = np.concatenate(pred)
        accuracy = classification_accuracy(true, pred)
        conf_mat = confusion_matrix(true, pred)
    
        if self.verbose >= 0:
            log_info = dict(self._dynamic_params)
            log_info["performance"] = "Acc {0:.3f}".format(accuracy)
            self._info(
                "{0: <80}".format(
                    "Eval: Epoch {epoch} {performance}".format(**log_info)
                )
            )
    
        return accuracy, conf_mat
    

    from_module(model_module, labels, **kwargs) classmethod

    Create a VectorNet model from a loadable module.
    Param Type Description
    model_module module or str (path to) a local Python workspace module which contains a get_vectorizer() callable, get_architecture() callable, and a get_state_dict_path() callable
    labels list list of str classification labels
    **kwargs forwarded to self.__init__() constructor
    Source code in hover/core/neural.py
    @classmethod
    def from_module(cls, model_module, labels, **kwargs):
        """
        ???+ note "Create a VectorNet model from a loadable module."
    
            | Param          | Type       | Description                          |
            | :------------- | :--------- | :----------------------------------- |
            | `model_module` | `module` or `str` | (path to) a local Python workspace module which contains a get_vectorizer() callable, get_architecture() callable, and a get_state_dict_path() callable |
            | `labels`       | `list`     | list of `str` classification labels  |
            | `**kwargs`     |      | forwarded to `self.__init__()` constructor |
        """
        if isinstance(model_module, str):
            from importlib import import_module
    
            model_module = import_module(model_module)
    
        # Load the model by retrieving the inp-to-vec function, architecture, and state dict
        model = cls(
            model_module.get_vectorizer(),
            model_module.get_architecture(),
            model_module.get_state_dict_path(),
            labels,
            **kwargs,
        )
    
        return model
    

    load(self, load_path=None)

    Load neural net parameters if possible.

    Can be directed to a custom state dict.

    Param Type Description
    load_path str path to a torch state dict
    Source code in hover/core/neural.py
    def load(self, load_path=None):
        """
        ???+ note "Load neural net parameters if possible."
    
            Can be directed to a custom state dict.
    
            | Param       | Type       | Description                  |
            | :---------- | :--------- | :--------------------------- |
            | `load_path` | `str`      | path to a `torch` state dict |
        """
        load_path = load_path or self.nn_update_path
        # if the architecture cannot match the state dict, skip the load and warn
        try:
            self.nn.load_state_dict(torch.load(load_path))
            self._info(f"loaded state dict {load_path}.")
        except Exception as e:
            self._warn(f"load VectorNet state path failed with {type(e)}: {e}")
    

    manifold_trajectory(self, inps, method=None, reducer_kwargs=None, spline_kwargs=None)

    Compute a propagation trajectory of the dataset manifold through the neural net.
    1. vectorize inps
    2. forward propagate, keeping intermediates
    3. fit intermediates to N-D manifolds
    4. fit manifolds using Procrustes shape analysis
    5. fit shapes to trajectory splines
    Param Type Description
    inps dynamic (a list of) input features to vectorize
    method str reduction method: "umap" or "ivis"
    reducer_kwargs kwargs to forward to dimensionality reduction
    spline_kwargs kwargs to forward to spline calculation
    Source code in hover/core/neural.py
    def manifold_trajectory(
        self, inps, method=None, reducer_kwargs=None, spline_kwargs=None
    ):
        """
        ???+ note "Compute a propagation trajectory of the dataset manifold through the neural net."
    
            1. vectorize inps
            2. forward propagate, keeping intermediates
            3. fit intermediates to N-D manifolds
            4. fit manifolds using Procrustes shape analysis
            5. fit shapes to trajectory splines
    
            | Param    | Type    | Description                          |
            | :------- | :------ | :----------------------------------- |
            | `inps`   | dynamic | (a list of) input features to vectorize |
            | `method` | `str`   | reduction method: `"umap"` or `"ivis"`  |
            | `reducer_kwargs` | | kwargs to forward to dimensionality reduction |
            | `spline_kwargs` | | kwargs to forward to spline calculation |
        """
        from hover.core.representation.manifold import LayerwiseManifold
        from hover.core.representation.trajectory import manifold_spline
    
        if method is None:
            method = hover.config["data.embedding"]["default_reduction_method"]
    
        reducer_kwargs = reducer_kwargs or {}
        spline_kwargs = spline_kwargs or {}
    
        # step 1 & 2
        vectors = torch.Tensor(np.array([self.vectorizer(_inp) for _inp in inps]))
        self.nn.eval()
        intermediates = self.nn.eval_per_layer(vectors)
        intermediates = [_tensor.detach().numpy() for _tensor in intermediates]
    
        # step 3 & 4
        LM = LayerwiseManifold(intermediates)
        LM.unfold(method=method, **reducer_kwargs)
        seq_arr, disparities = LM.procrustes()
        seq_arr = np.array(seq_arr)
    
        # step 5
        traj_arr = manifold_spline(np.array(seq_arr), **spline_kwargs)
    
        return traj_arr, seq_arr, disparities
    

    predict_proba(self, inps)

    End-to-end single/multi-piece prediction from inp to class probabilities.
    Param Type Description
    inps dynamic (a list of) input features to vectorize
    Source code in hover/core/neural.py
    def predict_proba(self, inps):
        """
        ???+ note "End-to-end single/multi-piece prediction from inp to class probabilities."
            | Param  | Type    | Description                          |
            | :----- | :------ | :----------------------------------- |
            | `inps` | dynamic | (a list of) input features to vectorize |
        """
        # if the input is a single piece of inp, cast it to a list
        FLAG_SINGLE = not isinstance(inps, list)
        if FLAG_SINGLE:
            inps = [inps]
    
        # the actual prediction
        self.nn.eval()
        vectors = torch.Tensor(np.array([self.vectorizer(_inp) for _inp in inps]))
        logits = self.nn(vectors)
        probs = F.softmax(logits, dim=-1).detach().numpy()
    
        # inverse-cast if applicable
        if FLAG_SINGLE:
            probs = probs[0]
    
        return probs
    

    prepare_loader(self, dataset, key, **kwargs)

    Create dataloader from SupervisableDataset with implied vectorizer(s).
    Param Type Description
    dataset hover.core.dataset.SupervisableDataset the dataset to load
    key str "train", "dev", or "test"
    **kwargs forwarded to dataset.loader()
    Source code in hover/core/neural.py
    def prepare_loader(self, dataset, key, **kwargs):
        """
        ???+ note "Create dataloader from `SupervisableDataset` with implied vectorizer(s)."
    
            | Param      | Type  | Description                |
            | :--------- | :---- | :------------------------- |
            | `dataset`  | `hover.core.dataset.SupervisableDataset` | the dataset to load |
            | `key`      | `str` | "train", "dev", or "test"  |
            | `**kwargs` | | forwarded to `dataset.loader()`  |
        """
        return dataset.loader(key, self.vectorizer, **kwargs)
    

    save(self, save_path=None)

    Save the current state dict with authorization to overwrite.
    Param Type Description
    save_path str option alternative path to state dict
    Source code in hover/core/neural.py
    def save(self, save_path=None):
        """
        ???+ note "Save the current state dict with authorization to overwrite."
            | Param       | Type  | Description                           |
            | :---------- | :---- | :------------------------------------ |
            | `save_path` | `str` | option alternative path to state dict |
        """
        save_path = save_path or self.nn_update_path
        torch.save(self.nn.state_dict(), save_path)
        verb = "overwrote" if os.path.isfile(save_path) else "saved"
        self._info(f"{verb} state dict {save_path}.")
    

    setup_label_conversion(self, labels)

    Set up label encoder/decoder and number of classes.
    Param Type Description
    labels list list of str classification labels
    Source code in hover/core/neural.py
    def setup_label_conversion(self, labels):
        """
        ???+ note "Set up label encoder/decoder and number of classes."
    
            | Param             | Type       | Description                          |
            | :---------------- | :--------- | :----------------------------------- |
            | `labels`          | `list`     | list of `str` classification labels  |
        """
        self.label_encoder = {_label: i for i, _label in enumerate(labels)}
        self.label_decoder = {i: _label for i, _label in enumerate(labels)}
        self.num_classes = len(self.label_encoder)
    

    setup_nn(self, use_existing_state_dict=True)

    Set up neural network and optimizers.

    Intended to be called in and out of the constructor.

    • will try to load parameters from state dict by default
    • option to override and discard previous state dict
      • often used when the classification targets have changed
    Param Type Description
    labels list list of str classification labels
    use_existing_state_dict bool whether to use existing state dict
    Source code in hover/core/neural.py
    def setup_nn(self, use_existing_state_dict=True):
        """
        ???+ note "Set up neural network and optimizers."
    
            Intended to be called in and out of the constructor.
    
            -   will try to load parameters from state dict by default
            -   option to override and discard previous state dict
                -   often used when the classification targets have changed
    
            | Param                     | Type       | Description                          |
            | :------------------------ | :--------- | :----------------------------------- |
            | `labels`                  | `list`     | list of `str` classification labels  |
            | `use_existing_state_dict` | `bool`     | whether to use existing state dict   |
        """
        # set up vectorizer and the neural network with appropriate dimensions
        vec_dim = self.vectorizer(self.example_input).shape[0]
        self.nn = self.architecture(vec_dim, self.num_classes)
        self._callback_reset_nn_optimizer()
    
        state_dict_exists = os.path.isfile(self.nn_update_path)
        # if state dict exists, load it (when consistent) or overwrite
        if state_dict_exists:
            if use_existing_state_dict:
                self.load(self.nn_update_path)
            else:
                self.save(self.nn_update_path)
    
        self._good(f"reset neural net: in {vec_dim} out {self.num_classes}.")
    

    train(self, train_loader, dev_loader=None, epochs=None)

    Train the neural network part of the VecNet.
    • intended to be coupled with self.train_batch().
    Param Type Description
    train_loader torch.utils.data.DataLoader train set
    dev_loader torch.utils.data.DataLoader dev set
    epochs int number of epochs to train
    Source code in hover/core/neural.py
    def train(self, train_loader, dev_loader=None, epochs=None):
        """
        ???+ note "Train the neural network part of the VecNet."
    
            - intended to be coupled with self.train_batch().
    
            | Param          | Type         | Description                |
            | :------------- | :----------- | :------------------------- |
            | `train_loader` | `torch.utils.data.DataLoader` | train set |
            | `dev_loader`   | `torch.utils.data.DataLoader` | dev set   |
            | `epochs`       | `int`        | number of epochs to train  |
        """
        epochs = epochs or self.epochs_slider.value
    
        train_info = []
        for epoch_idx in range(epochs):
            self._dynamic_params["epoch"] = epoch_idx + 1
            self.train_epoch(train_loader)
            if dev_loader is not None:
                dev_loader = train_loader
            acc, conf_mat = self.evaluate(dev_loader)
            train_info.append({"accuracy": acc, "confusion_matrix": conf_mat})
        return train_info
    

    train_batch(self, loaded_input, loaded_output)

    Train the neural network for one batch.
    Param Type Description
    loaded_input torch.Tensor input tensor
    loaded_output torch.Tensor output tensor
    Source code in hover/core/neural.py
    def train_batch(self, loaded_input, loaded_output):
        """
        ???+ note "Train the neural network for one batch."
    
            | Param           | Type           | Description           |
            | :-------------- | :------------- | :-------------------- |
            | `loaded_input`  | `torch.Tensor` | input tensor          |
            | `loaded_output` | `torch.Tensor` | output tensor         |
        """
        self.nn.train()
        input_tensor = loaded_input.float()
        output_tensor = loaded_output.float()
    
        # compute logits
        logits = self.nn(input_tensor)
        loss = F.cross_entropy(logits, output_tensor)
    
        self.nn_optimizer.zero_grad()
        loss.backward()
        self.nn_optimizer.step()
    
        if self.verbose > 0:
            log_info = dict(self._dynamic_params)
            log_info["performance"] = "Loss {0:.3f}".format(loss)
            self._print(
                "{0: <80}".format(
                    "Train: Epoch {epoch} Batch {batch} {performance}".format(
                        **log_info
                    )
                ),
                end="\r",
            )
    

    train_epoch(self, train_loader, *args, **kwargs)

    Train the neural network for one epoch.
    • Supports flexible args and kwargs for child classes that may implement self.train() and self.train_batch() differently.
    Param Type Description
    train_loader torch.utils.data.DataLoader train set
    *args arguments to forward to train_batch
    **kwargs kwargs to forward to train_batch
    Source code in hover/core/neural.py
    def train_epoch(self, train_loader, *args, **kwargs):
        """
        ???+ note "Train the neural network for one epoch."
    
            - Supports flexible args and kwargs for child classes that may implement self.train() and self.train_batch() differently.
    
            | Param          | Type         | Description                |
            | :------------- | :----------- | :------------------------- |
            | `train_loader` | `torch.utils.data.DataLoader` | train set |
            | `*args`        | | arguments to forward to `train_batch`   |
            | `**kwargs`     | | kwargs to forward to `train_batch`      |
        """
        self.adjust_optimizer_params()
        for batch_idx, (loaded_input, loaded_output, _) in enumerate(train_loader):
            self._dynamic_params["batch"] = batch_idx + 1
            self.train_batch(loaded_input, loaded_output, *args, **kwargs)
    

    view(self)

    Overall layout when plotted.
    Source code in hover/core/neural.py
    def view(self):
        """
        ???+ note "Overall layout when plotted."
        """
        return self._layout_widgets()