eryon_surface/network/

impl_network.rs

/*
    Appellation: impl_network <module>
    Contrib: @FL03
*/
use crate::model::SurfaceModel;
use crate::network::SurfaceNetwork;
use crate::points::{Point, PointKind};
use cnc::nn::{NeuralError, Predict, PredictWithConfidence, Train};
use cnc::params::Params;
use cnc::prelude::{Forward, Init, InitInplace};
use ndarray::ScalarOperand;
use num_traits::{Float, FromPrimitive, Num, NumAssign, Signed};
use rstmt::nrt::{LPR, Triad};

impl<A> SurfaceNetwork<A>
where
    A: FromPrimitive + ScalarOperand,
{
    /// Add a new critical point to configure part of the network's hidden layer
    pub fn add_critical_point(&mut self, kind: PointKind) -> &mut Self
    where
        A: Float,
    {
        // Create a new critical point from the current triad
        let critical_point = Point::from_triad(kind, &self.headspace);

        // Add to critical points
        self.points.push(critical_point);

        // Ensure model has enough hidden layers for all critical points
        let current_layers = self.model.features().layers();
        let required_layers = self.points.len().max(3);

        if current_layers < required_layers {
            // Update model features
            self.model.features_mut().set_layers(required_layers);

            // Add additional layers as needed
            let layer_dims = self.model.features().dim_hidden();
            let hidden = self.model.hidden_mut();

            // Initialize new layers - use existing or zero initialization
            while hidden.len() < required_layers {
                hidden.push(Params::zeros(layer_dims));
            }
        }

        self
    }
    #[doc(hidden)]
    #[deprecated(since = "0.0.2", note = "Use `train` instead")]
    /// perform a single backpropagation step on qualifing datasets
    pub fn backward<X, Y, Z>(&mut self, input: &X, expected: &Y) -> Result<Z, NeuralError>
    where
        A: Float + FromPrimitive + NumAssign + ScalarOperand + rustfft::FftNum,
        SurfaceModel<A>: Train<X, Y, Output = Z>,
    {
        self.model_mut().train(input, expected)
    }
    #[doc(hidden)]
    #[deprecated(since = "0.0.2", note = "Use `predict` instead")]
    /// forward propogate the given input through the network
    pub fn forward<X, Y>(&self, input: &X) -> cnc::Result<Y>
    where
        SurfaceModel<A>: cnc::Forward<X, Output = Y>,
    {
        self.model.forward(input)
    }
    /// forward propogate the given input through the network
    pub fn predict<X, Y>(&self, input: &X) -> Result<Y, NeuralError>
    where
        SurfaceModel<A>: Predict<X, Output = Y>,
    {
        Predict::predict(&self.model, input)
    }
    /// perform a single backpropagation step on qualifing datasets
    pub fn train<X, Y, Z>(&mut self, input: &X, expected: &Y) -> Result<Z, NeuralError>
    where
        A: Float + FromPrimitive + NumAssign + ScalarOperand + rustfft::FftNum,
        SurfaceModel<A>: Train<X, Y, Output = Z>,
    {
        self.model_mut().train(input, expected)
    }
    /// get the position of the centroid w.r.t the headspace
    pub fn get_centroid(&self) -> Option<[A; 2]>
    where
        A: Float + FromPrimitive,
    {
        self.headspace().centroid()
    }
    /// returns the position of a particular hidden layer w.r.t. the current headspace
    pub fn get_position(&self, cp_index: usize) -> Option<[A; 3]>
    where
        A: Copy + Num,
    {
        if cp_index >= self.points.len() {
            return None;
        }

        let cp = &self.points[cp_index];
        let notes = self.headspace.notes();
        let mut result = [A::zero(); 3];

        // Weight each note by its barycentric coordinate
        for i in 0..3 {
            let note_val = A::from_usize(notes[i]).unwrap();
            // Apply modulus to properly position in tonal space
            let octave_shift =
                A::from_isize(cp.modulus().into_inner()).unwrap() * A::from_isize(12).unwrap();
            result[i] = (note_val + octave_shift) * cp.coordinates[i];
        }

        Some(result)
    }
    /// initialize the network
    pub fn init(self) -> Self
    where
        SurfaceModel<A>: Init,
    {
        let model = self.model.init();
        Self { model, ..self }
    }
    /// initialize the network in-place
    pub fn init_inplace(&mut self) -> &mut Self
    where
        SurfaceModel<A>: InitInplace,
    {
        self.model_mut().init_inplace();
        self
    }
    /// returns the model's prediction with a confidence value
    pub fn predict_with_confidence<X, Y>(
        &self,
        input: &X,
    ) -> Result<(Y, <SurfaceModel<A> as PredictWithConfidence<X>>::Confidence), NeuralError>
    where
        SurfaceModel<A>: PredictWithConfidence<X, Output = Y>,
    {
        self.model().predict_with_confidence(input)
    }
    /// reconfigure the network with respect to the given triad; the method is typically called
    /// after applying a transformation to the headspace
    #[cfg_attr(
        feature = "tracing",
        tracing::instrument(skip_all, name = "reconfigure", target = "network", level = "trace")
    )]
    pub fn reconfigure(&mut self, next: Triad)
    where
        A: Float,
    {
        // Create new critical points for the new triad
        let new_critical_points = self
            .points
            .iter()
            .map(|cp| Point::from_triad(cp.kind, &next))
            .collect::<Vec<_>>();

        // update the network's critical points
        self.set_points(new_critical_points);
        // update the model's headspace
        self.headspace = next;
    }
    /// Transfer knowledge from another surface network
    pub fn transfer_knowledge_from(&mut self, other: &Self, adaptation_rate: A)
    where
        A: Copy + PartialOrd + Num + Signed,
    {
        // Only transfer if the networks have the same number of critical points
        if self.points.len() != other.points.len() {
            return;
        }

        // Map critical points between networks based on type
        let mut cp_mapping = Vec::new();

        for (i, cp) in self.points.iter().enumerate() {
            let mut best_match = 0;
            let mut best_score = A::zero();

            for (j, other_cp) in other.points.iter().enumerate() {
                // Match based on kind and coordinates
                if cp.kind() == other_cp.kind() {
                    let mut similarity = A::from_f32(0.5).unwrap(); // Base score for matching kind

                    // Add coordinate similarity
                    let coord_similarity = (0..3).fold(A::zero(), |acc, k| {
                        acc + (A::one() - (cp.coordinates[k] - other_cp.coordinates[k]).abs())
                    }) / A::from_usize(3).unwrap();

                    similarity = similarity + coord_similarity * A::from_f32(0.5).unwrap();

                    if similarity > best_score {
                        best_score = similarity;
                        best_match = j;
                    }
                }
            }

            cp_mapping.push((i, best_match, best_score));
        }

        // Transfer primary weights
        for (self_idx, other_idx, score) in &cp_mapping {
            if *score > A::from_f32(0.6).unwrap() {
                for j in 0..3 {
                    // Blend weights based on similarity score and adaptation rate
                    self.model.input_mut().weights_mut()[[*self_idx, j]] = self
                        .model
                        .input()
                        .weights()[[*self_idx, j]]
                        * (A::one() - adaptation_rate * *score)
                        + other.model.input().weights()[[*other_idx, j]] * adaptation_rate * *score;
                }
            }
        }

        // Transfer secondary weights
        for (self_idx, other_idx, score) in &cp_mapping {
            if *score > A::from_f32(0.6).unwrap() {
                // Blend weights
                self.model.output_mut().weights_mut()[[0, *self_idx]] = self
                    .model
                    .output()
                    .weights()[[0, *self_idx]]
                    * (A::one() - adaptation_rate * *score)
                    + other.model.output().weights()[[0, *other_idx]] * adaptation_rate * *score;
            }
        }

        // Transfer temporal context if both have surface layers
        if !self.model.hidden().is_empty() && !other.model.hidden().is_empty() {
            for (self_i, other_i, score_i) in &cp_mapping {
                for (self_j, other_j, score_j) in &cp_mapping {
                    if *score_i > A::from_f32(0.6).unwrap() && *score_j > A::from_f32(0.6).unwrap()
                    {
                        // Blend temporal weights
                        self.model.hidden_mut()[0].weights_mut()[[*self_i, *self_j]] =
                            self.model.hidden()[0].weights()[[*self_i, *self_j]]
                                * (A::one() - adaptation_rate * *score_i * *score_j)
                                + other.model.hidden()[0].weights()[[*other_i, *other_j]]
                                    * adaptation_rate
                                    * *score_i
                                    * *score_j;
                    }
                }
            }
        }
    }
    /// applies the given [LPR] transformation to the headspace of the network before applying
    /// the neccessary adjusments to the architecture.
    #[cfg_attr(
        feature = "tracing",
        tracing::instrument(skip_all, name = "transform", target = "network", level = "trace")
    )]
    pub fn transform(&mut self, transform: LPR) -> crate::SurfaceResult<()>
    where
        A: Float,
    {
        // apply transformation to the triad
        let new_triad = self.headspace().transform(transform);
        // Update network's triad and critical points
        self.reconfigure(new_triad);

        Ok(())
    }
    /// Walk the network through a series of transformations
    #[cfg_attr(
        feature = "tracing",
        tracing::instrument(skip_all, name = "walk", target = "network", level = "trace")
    )]
    pub fn walk<I>(&mut self, iter: I) -> crate::SurfaceResult<()>
    where
        A: Float,
        I: IntoIterator<Item = LPR>,
    {
        for transform in iter {
            self.transform(transform)?;
        }
        Ok(())
    }
}

impl<A, X, Y> Forward<X> for SurfaceNetwork<A>
where
    A: Float + FromPrimitive + NumAssign + ScalarOperand,
    SurfaceModel<A>: Forward<X, Output = Y>,
{
    type Output = Y;

    fn forward(&self, input: &X) -> cnc::Result<Self::Output> {
        <SurfaceModel<A> as Forward<X>>::forward(self.model(), input)
    }
}

impl<A, X, Y, Z> Train<X, Y> for SurfaceNetwork<A>
where
    A: Float + FromPrimitive + NumAssign + ScalarOperand,
    SurfaceModel<A>: Train<X, Y, Output = Z>,
{
    type Output = Z;

    fn train(&mut self, input: &X, expected: &Y) -> Result<Self::Output, cnc::nn::NeuralError> {
        <SurfaceModel<A> as Train<X, Y>>::train(self.model_mut(), input, expected)
    }
}