nullpol.analysis.likelihood¶

class Chi2TimeFrequencyLikelihood(TimeFrequencyLikelihood):
    """A time-frequency likelihood class that computes the likelihood of the total null energy.

    Args:
        interferometers (list): List of interferometers.
        wavelet_frequency_resolution (float): The frequency resolution of the wavelet transform.
        wavelet_nx (int): The number of points in the wavelet transform.
        polarization_modes (list): List of polarization modes.
        polarization_basis (list): List of polarization basis.
        time_frequency_filter (str): The time-frequency filter.
    """

    def __init__(
        self,
        interferometers,
        wavelet_frequency_resolution,
        wavelet_nx,
        polarization_modes,
        *args,
        polarization_basis=None,
        time_frequency_filter=None,
        **kwargs,
    ):
        """Initialize the instance."""
        super().__init__(
            *args,
            interferometers=interferometers,
            wavelet_frequency_resolution=wavelet_frequency_resolution,
            wavelet_nx=wavelet_nx,
            polarization_modes=polarization_modes,
            polarization_basis=polarization_basis,
            time_frequency_filter=time_frequency_filter,
            **kwargs,
        )

    @property
    def DoF(self):
        """Degree of freedom.

        Returns:
            int: Degree of freedom.
        """
        polarization_basis_sum = np.sum(self.antenna_pattern_processor.polarization_basis)

        if self.data_context.time_frequency_filter is None:
            raise ValueError("Time frequency filter is not available")

        time_frequency_filter_sum = np.sum(self.data_context.time_frequency_filter)
        return (len(self.interferometers) - polarization_basis_sum) * time_frequency_filter_sum

    def log_likelihood(self):
        """Compute the log likelihood using the projection approach."""
        # Compute null energy using the parameters
        null_energy = self.null_stream_calculator.compute_null_energy(self.parameters)

        return scipy.stats.chi2.logpdf(null_energy, df=self.DoF)

    def _calculate_noise_log_likelihood(self):
        """Calculate the noise log likelihood.

        Returns:
            float: noise log likelihood.
        """
        if self.data_context.whitened_frequency_domain_strain_array is None:
            raise ValueError("Whitened frequency domain strain array is not available")
        if self.data_context.time_frequency_filter is None:
            raise ValueError("Time frequency filter is not available")

        wavelet_domain_strain_array = np.array(
            [
                transform_wavelet_freq(
                    data=self.data_context.whitened_frequency_domain_strain_array[i],
                    sampling_frequency=self.data_context.sampling_frequency,
                    frequency_resolution=self.data_context.wavelet_frequency_resolution,
                    nx=self.data_context.wavelet_nx,
                )
                for i in range(len(self.interferometers))
            ]
        )
        E = np.sum(np.abs(wavelet_domain_strain_array * self.data_context.time_frequency_filter) ** 2)
        log_likelihood = scipy.stats.chi2.logpdf(
            E, df=len(self.interferometers) * np.sum(self.data_context.time_frequency_filter)
        )
        return log_likelihood

class TimeFrequencyLikelihood(Likelihood):
    """A time-frequency likelihood class with modular architecture.

    This class uses composition to separate concerns:
    - Data management and antenna pattern operations are handled by NullStreamCalculator
    - Null stream computations are handled by NullStreamCalculator

    For low-level data access, use: likelihood.null_stream_calculator.data_context.property_name
    For antenna pattern operations, use: likelihood.null_stream_calculator.antenna_pattern_processor.method_name
    For null stream computations, use: likelihood.null_stream_calculator.method_name
    For high-level likelihood operations, use the methods on this class directly.

    Args:
        interferometers (list): List of interferometers.
        wavelet_frequency_resolution (float): The frequency resolution of the wavelet transform.
        wavelet_nx (int): The number of points in the wavelet transform.
        polarization_modes (list): List of polarization modes.
        polarization_basis (list): List of polarization basis.
        time_frequency_filter (str): The time-frequency filter.
    """

    def __init__(
        self,
        interferometers,
        wavelet_frequency_resolution,
        wavelet_nx,
        polarization_modes,
        *args,  # pylint: disable=unused-argument
        polarization_basis=None,
        time_frequency_filter=None,
        **kwargs,  # pylint: disable=unused-argument
    ):
        """Initialize the instance."""
        super().__init__({})

        # Initialize null stream calculator with all components
        self.null_stream_calculator = NullStreamCalculator(
            interferometers=interferometers,
            wavelet_frequency_resolution=wavelet_frequency_resolution,
            wavelet_nx=wavelet_nx,
            polarization_modes=polarization_modes,
            polarization_basis=polarization_basis,
            time_frequency_filter=time_frequency_filter,
        )

        # Initialize the normalization constant
        self._noise_log_likelihood_value = None

        # Marginalization
        self._marginalized_parameters = []

    @property
    def data_context(self):
        """Access to the data context through null_stream_calculator.

        Returns:
            TimeFrequencyDataContext: The data context instance.
        """
        return self.null_stream_calculator.data_context

    @property
    def antenna_pattern_processor(self):
        """Access to the antenna pattern processor through null_stream_calculator.

        Returns:
            AntennaPatternProcessor: The antenna pattern processor instance.
        """
        return self.null_stream_calculator.antenna_pattern_processor

    @property
    def interferometers(self):
        """A list of interferometers.

        Returns:
            bilby.gw.detector.InterferometerList: A list of interferometers.
        """
        return self.null_stream_calculator.data_context.interferometers

    def log_likelihood(self):
        """Log likelihood.

        Raises:
            NotImplementedError: This should be implemented in a subclass.
        """
        raise NotImplementedError("The log_likelihood method must be implemented in a subclass.")

    def _calculate_noise_log_likelihood(self):
        """Calculate noise log likelihood.

        This should be implemented in a subclass.

        Raises:
            NotImplementedError: This should be implemented in a subclass.
        """
        raise NotImplementedError("The _calculate_noise_log_likelihood method must be implemented in a subclass.")

    def noise_log_likelihood(self):
        """Compute the noise log likelihood.

        Returns:
            float: The noise log likelihood.
        """
        if self._noise_log_likelihood_value is None:
            self._noise_log_likelihood_value = self._calculate_noise_log_likelihood()
        return self._noise_log_likelihood_value