text:
# -*- coding: utf-8 -*-
#
# Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#     http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
This module plots the results from the Active Learning Pipeline.
"""

import os
import json
import logging
from collections import Counter
from typing import Dict, List
import numpy as np
from ..components._util import _is_module_available, _get_module_or_attr
from .classifiers import MindMeldALClassifier
from ..path import (
    AL_ACCURACIES_PATH,
    AL_SELECTED_QUERIES_PATH,
)

logger = logging.getLogger(__name__)

FIRST_EPOCH = "0"
FIRST_ITERATION = "0"

class MissingDataError(Exception):
    pass

class PlotManager:
    """Handles plotting. Plots supported:
    At the domain, intent and entity level:
        1. Plot Single Epoch (Compares Strategies)
        2. Plot Average Across Epochs (Compares Strategies)
        3. Plot All Epochs (Compares Epochs)
    At the domain and intent level:
        1. Plot Stacked Bar (Compares Selection Distributions across Iterations)
    """

    def __init__(
        self,
        experiment_dir_path: str,
        aggregate_statistic: str,
        class_level_statistic: str,
    ):
        """
        Args:
            experiment_dir_path (str): Path to the experiment directory.
            aggregate_statistic (str): Aggregate statistic to record.
                (Options: "accuracy", "f1_weighted", "f1_macro", "f1_micro".)
            class_level_statistic (str): Class_level statistic to record.
                (Options: "f_beta", "percision", "recall")
        """
        self.experiment_dir_path = experiment_dir_path
        self.aggregate_statistic = MindMeldALClassifier._validate_aggregate_statistic(
            aggregate_statistic
        )
        self.class_level_statistic = (
            MindMeldALClassifier._validate_class_level_statistic(class_level_statistic)
        )
        self.accuracies_data = self.get_accuracies_json_data()
        self.queries_data = self.get_queries_json_data()

        if not _is_module_available("matplotlib"):
            raise ModuleNotFoundError(
                "Library not found: 'matplotlib'. Run 'pip install mindmeld[active_learning]' to install."
            )

        self.plt = _get_module_or_attr("matplotlib.pyplot")

    # Get JSON Data
    def get_accuracies_json_data(self) -> Dict:
        """Loads accuracies.json from the experiment directory path.
        Returns:
            data (dict): Data loaded from accuracies.json.
        """
        accuracies_json_path = AL_ACCURACIES_PATH.format(
            experiment_folder=self.experiment_dir_path
        )
        with open(accuracies_json_path, "r") as infile:
            return json.load(infile)

    def get_queries_json_data(self) -> Dict:
        """Loads selected_queries.json from the experiment directory path.
        selected_queries.json stores the queries selected by active learning
        at each iteration.
        Returns:
            data (dict): Data loaded from selected_queries.json.
        """
        selected_queries_json_path = AL_SELECTED_QUERIES_PATH.format(
            experiment_folder=self.experiment_dir_path
        )
        with open(selected_queries_json_path, "r") as infile:
            return json.load(infile)

    def queries_json_data_has_data(self) -> Dict:
        """Checks whether queries.json is empty {}.
        Returns:
            has_data (bool): Whether queries.json has data.
        """
        return self.get_queries_json_data() != {}

    # Plotting Meta Functions
    def generate_plots(self):
        """Entry point for generating plots as per the specifications provided in the config."""
        logger.info("Starting to generate plots")
        plot_functions = [
            "plot_single_epoch",
            "plot_avg_across_epochs",
            "plot_all_epochs",
        ]
        for plot_function in plot_functions:
            logger.info("Plotting: %s", plot_function)
            plot_function = getattr(PlotManager, plot_function)
            self.plotting_wrapper(plot_function)
        if self.queries_json_data_has_data():
            logger.info("Plotting: plot_stacked_bar")
            self.plot_stacked_bar()

    def plotting_wrapper(
        self,
        function,
        plot_domain: bool = True,
        plot_intents: bool = True,
        plot_entities: bool = False,
    ):
        """Plotting wrapper functions for plots that use data from accuracies.json
        Args:
            function (generator): plotting function.
            plot_domain (bool): Whether to generate plots at the domain level.
            plot_intents (bool): Whether to generate plots at the intent level.
            plot_entities (bool): Whether to generate plots at the entity level.
        """
        self._check_first_epoch_and_iter_exist()
        if not plot_domain:
            return
        function(self, y_keys=["overall"])
        for domain in self.get_domain_list():
            function(self, y_keys=[domain, "overall"])
            if not plot_intents:
                continue
            for intent in self.get_intent_list(domain):
                function(
                    self,
                    y_keys=[domain, intent, "overall"],
                    use_aggregate_statistic=False,
                )
                if not plot_entities:
                    continue
                y_keys = [domain, intent, "entities", "overall"]
                function(self, y_keys=y_keys)
                for entity in self.get_entity_list(domain, intent):
                    y_keys = [domain, intent, "entities", entity]
                    function(self, y_keys=y_keys)

    # Helper Methods
    @staticmethod
    def get_nested(data_dict: Dict, selected_keys: List):
        """Filter data from a nested dictionary selecting from a set of keys.
        Args:
            data_dict (dict): Dictionary containing data to filter
            selected_keys (list): List of keys used to filter the given dictionary.

        Returns:
            data_dict (dict): Dictionary containing the filtered nested data.
        """
        for selected_key in selected_keys:
            data_dict = data_dict[selected_key]
        return data_dict

    @staticmethod
    def get_across_iterations(epoch_dict: Dict, selected_keys: List):
        """Filter data across all iterations in a single epoch as specified by a set of keys.
        Args:
            epoch_dict (dict): Dict containing accuracies across iterations for a single epoch.
            selected_keys (list): List of keys used to filter the given dictionary.
        Return:
            data (list): List of the selected data across iterations.
        """
        return [
            PlotManager.get_nested(epoch_dict[str(i)], selected_keys)
            for i in range(len(epoch_dict))
        ]

    @property
    def strategies(self) -> List:
        """
        Returns:
            strategies (list): List of selection strategies for the given experiment.
        """
        strategies = list(self.accuracies_data)
        if len(strategies) == 0:
            raise MissingDataError("Did not find data in accuracies.json.")
        return strategies

    def _check_first_epoch_and_iter_exist(self):
        """Check whether data for the first iteration in the first epoch exists.
        Data from the first epoch and iteration is used to determine the domains, intents,
        and entities included in training.
        Raises:
            MissingDataError: Throws an error if the anticipated data is not found.
        """
        first_strategy = self.strategies[0]
        if FIRST_EPOCH not in self.accuracies_data[first_strategy]:
            raise MissingDataError("Did not find data for the first epoch.")
        if FIRST_ITERATION not in self.accuracies_data[first_strategy][FIRST_EPOCH]:
            raise MissingDataError(
                "Did not find data for the first iteration in the first epoch."
            )

    def get_domain_list(self) -> List:
        """Method to get a list of domains included in training from the first epoch and iteration.
        Returns:
            domain_list (list): List of domains for the current experiment.
        """
        first_strategy = self.strategies[0]
        domain_list = list(
            self.accuracies_data[first_strategy][FIRST_EPOCH][FIRST_ITERATION][
                "accuracies"
            ].keys()
        )
        # The 'overall' score across domains is removed as it is not a domain
        domain_list.remove("overall")
        return domain_list

    def get_intent_list(self, domain: str) -> List:
        """Method to get a list of intents included in training from the first epoch and iteration.
        Args:
            domain (str): The domain to get retrieve intents for.
        Returns:
            intent_list (list): List of intent for a given domain in the current experiment.
        """
        first_strategy = self.strategies[0]
        intent_list = list(
            self.accuracies_data[first_strategy][FIRST_EPOCH][FIRST_ITERATION][
                "accuracies"
            ][domain].keys()
        )
        # The 'overall' score across intents is removed as it is not an intent
        intent_list.remove("overall")
        return intent_list

    def get_entity_list(self, domain: str, intent: str) -> List:
        """Method to get a list of entities included in training from the first epoch and iteration.
        Args:
            domain (str): The domain containing the intent to retreive entities from.
            intent (str): The intent to retreive entities from.
        Returns:
            entity_list (list): List of entities in the given intent.
        """
        first_strategy = self.strategies[0]
        entity_list = list(
            self.accuracies_data[first_strategy][FIRST_EPOCH][FIRST_ITERATION][
                "accuracies"
            ][domain][intent]["entities"].keys()
        )
        # The 'overall' score across entities is removed as it is not an entity
        entity_list.remove("overall")
        return entity_list

    def create_plot_dir(self, y_keys: List):
        """Creates folders to support the generated path if they do not already exist.
        Args:
            y_keys (list): Keys to access the data from a epoch to be used as y values for plotting.
        """
        path_list = [self.experiment_dir_path, "plots"] + y_keys
        os.makedirs(os.path.join(*path_list), exist_ok=True)

    def get_img_path(self, y_keys: List, file_name: str):
        """
        Args:
            y_keys (list): Keys to access the data from a epoch to be used as y values for plotting.
            file_name (str): Name of the file to save.
        Returns:
            img_path (str): Path of the image to be saved.
        """
        path_list = [self.experiment_dir_path, "plots"] + y_keys + [f"{file_name}.png"]
        return os.path.join(*path_list)

    # Plotting Functions
    def plot_single_epoch(
        self,
        y_keys: List,
        epoch: int = 0,
        display: bool = False,
        save: bool = True,
        use_aggregate_statistic: bool = True,
    ):
        """Plot accuracies across a single epoch for each strategy.
        Args:
            y_keys (list): Keys to access the data from a epoch to be used as y values for plotting.
            epoch (int): The epoch to plot.
            display (bool): Whether to show the plot.
            save (bool): Whether to save the plot.
            use_aggregate_statistic (bool): If True, the aggregate_statistic will be used as the
                label for the y_axis. If False, the class_level_statistic will be used.
        """
        self.create_plot_dir(y_keys)
        for strategy in self.strategies:
            epoch_dict = self.accuracies_data[strategy][str(epoch)]
            x_values = PlotManager.get_across_iterations(epoch_dict, ["num_sampled"])
            y_values = PlotManager.get_across_iterations(
                epoch_dict, ["accuracies"] + y_keys
            )
            self.plt.plot(x_values, y_values)

        self.plt.xlabel("Number of selected queries")
        y_label = (
            self.aggregate_statistic
            if use_aggregate_statistic
            else self.class_level_statistic
        )
        self.plt.ylabel(y_label.capitalize())
        title = f"Epoch_{epoch}_Results_({'-'.join(y_keys)})"
        self.plt.title(title)
        self.plt.legend(self.strategies, loc="lower right")
        self.plt.grid()
        self.plt.tight_layout()
        fig = self.plt.gcf()
        if display:
            self.plt.show()
        if save:
            fig.savefig(self.get_img_path(y_keys, title))
            self.plt.clf()

    def plot_avg_across_epochs(
        self,
        y_keys: List,
        display: bool = False,
        save: bool = True,
        use_aggregate_statistic: bool = True,
    ):
        """Plot average accuracies across all epochs for each strategy.
        Args:
            y_keys (list): Keys to access the data from a epoch to be used as y values for plotting.
            display (bool): Whether to show the plot.
            save (bool): Whether to save the plot.
            use_aggregate_statistic (bool): If True, the aggregate_statistic will be used as the
                label for the y_axis. If False, the class_level_statistic will be used.
        """
        self.create_plot_dir(y_keys)
        for strategy in self.strategies:
            n_epochs = len(self.accuracies_data[strategy])
            all_y_values = []
            for epoch in range(n_epochs):
                epoch_dict = self.accuracies_data[strategy][str(epoch)]
                x_values = PlotManager.get_across_iterations(
                    epoch_dict, ["num_sampled"]
                )
                y_values = PlotManager.get_across_iterations(
                    epoch_dict, ["accuracies"] + y_keys
                )
                all_y_values.append(y_values)
            all_y_values = np.array(all_y_values)
            y_avg_values = all_y_values.mean(axis=0)
            self.plt.plot(x_values, y_avg_values)

        self.plt.xlabel("Number of selected queries")
        y_label = (
            self.aggregate_statistic
            if use_aggregate_statistic
            else self.class_level_statistic
        )
        self.plt.ylabel(y_label.capitalize())
        title = f"Avg_Across_Epochs_({'-'.join(y_keys)})"
        self.plt.title(title)
        self.plt.legend(self.strategies, loc="lower right")
        self.plt.grid()
        self.plt.tight_layout()
        fig = self.plt.gcf()
        if display:
            self.plt.show()
        if save:
            fig.savefig(self.get_img_path(y_keys, title))
            self.plt.clf()

    def plot_all_epochs(
        self,
        y_keys: List,
        display: bool = False,
        save: bool = True,
        use_aggregate_statistic: bool = True,
    ):
        """Plot all epochs. Creates a plot for each strategy.
        Args:
            y_keys (list): Keys to access the data from a epoch to be used as y values for plotting.
            display (bool): Whether to show the plot.
            save (bool): Whether to save the plot.
            use_aggregate_statistic (bool): If True, the aggregate_statistic will be used as the
                label for the y_axis. If False, the class_level_statistic will be used.
        """
        self.create_plot_dir(y_keys)
        for strategy in self.strategies:
            n_epochs = len(self.accuracies_data[strategy].keys())
            all_y_values = []
            for epoch in range(n_epochs):
                epoch_dict = self.accuracies_data[strategy][str(epoch)]
                x_values = PlotManager.get_across_iterations(
                    epoch_dict, ["num_sampled"]
                )
                y_values = PlotManager.get_across_iterations(
                    epoch_dict, ["accuracies"] + y_keys
                )
                self.plt.plot(x_values, y_values)
                all_y_values.append(y_values)
            max_len = max([len(i) for i in all_y_values])
            for y_values in all_y_values:
                y_values = np.pad(
                    y_values,
                    (0, max_len - len(y_values)),
                    "constant",
                    constant_values=(0, y_values[-1]),
                )
            y_avg_values = np.array(all_y_values).mean(axis=0)
            self.plt.plot(x_values, y_avg_values)

            self.plt.xlabel("Number of selected queries")
            y_label = (
                self.aggregate_statistic
                if use_aggregate_statistic
                else self.class_level_statistic
            )
            self.plt.ylabel(y_label.capitalize())
            title = f"{strategy}_All_Epochs_({'-'.join(y_keys)})"
            self.plt.title(title)
            self.plt.legend(
                ["epoch " + str(epoch) for epoch in range(n_epochs)] + ["avg"],
                loc="lower right",
            )
            self.plt.grid()
            self.plt.tight_layout()
            fig = self.plt.gcf()
            if display:
                self.plt.show()
            if save:
                fig.savefig(self.get_img_path(y_keys, title))
                self.plt.clf()

    @staticmethod
    def get_unique_labels(all_counters: List) -> List:
        """
        Args:
            all_counters (list): List of Counters, each counter represents a single iteration.
        Returns:
            unique_labels (list): A List of unique and sorted keys across all_counters.
        """
        unique_labels = []
        for counter in all_counters:
            unique_labels.extend(list(counter.keys()))
        unique_labels = list(set(unique_labels))
        unique_labels.sort()
        return unique_labels

    @staticmethod
    def get_label_set_counter(all_counters: List, unique_labels: List) -> Dict:
        """
        Args:
            all_counters (list): List of Counters, each counter represents a single iteration.
            unique_labels (list): A List of unique and sorted keys across all_counters.
        Returns:
            label_set_counter (dict): Each unique label is mapped to a list, the value at each
                index in the list corresponds to the count of the label in that iteration.
        """
        label_set_counter = {k: np.zeros(len(all_counters)) for k in unique_labels}
        for iteration, counter in enumerate(all_counters):
            for label in counter:
                label_set_counter[label][iteration] += counter[label]
        return label_set_counter

    # pylint: disable=W0613
    def plot_stacked_bar(
        self,
        epoch: int = 0,
        plot_domains: bool = True,
        plot_intents: bool = True,
        **kwargs,
    ):
        """Plots a stacked bar graph of selection distributions across iterations for an epoch.
        Args:
            epoch (int): The epoch to plot.
            plot_domain (bool): Whether to generate plots at the domain level.
            plot_intents (bool): Whether to generate plots at the intent level.
        """
        for strategy in self.queries_data:
            epoch_dict = self.queries_data[strategy][str(epoch)]
            domain_counters, intent_counters = [], []
            for iteration in epoch_dict:
                query_list = epoch_dict[iteration]
                if plot_domains:
                    domains = [q["domain"] for q in query_list]
                    domain_counters.append(Counter(domains))
                if plot_intents:
                    intents = [f"{q['intent']}|{q['domain']}" for q in query_list]
                    intent_counters.append(Counter(intents))

            for level, all_counters in zip(
                ["domain", "intent"], [domain_counters, intent_counters]
            ):
                if all_counters:
                    unique_labels = PlotManager.get_unique_labels(all_counters)
                    label_set_counter = PlotManager.get_label_set_counter(
                        all_counters, unique_labels
                    )
                    self._plot_stacked_bar(
                        num_iters=len(all_counters),
                        label_set_counter=label_set_counter,
                        strategy=strategy,
                        intent_level=(level == "intent"),
                    )

    def _plot_stacked_bar(
        self,
        num_iters: int,
        label_set_counter: Dict,
        strategy: str,
        intent_level: bool,
        display: bool = False,
        save: bool = True,
    ):
        """Helper function to plot a stacked bar graph.
        Args:
            num_iters (int): Number of iterations in the given epoch.
            label_set_counter (dict): Each unique label is mapped to a list, the value at the
                index in the list corresponds to the count of the label in that iteration.
            strategy (str): Selection strategy.
            intent_level (bool): Whether the plot is for intent or domain level distributions.
            display (bool): Whether to show the plot.
            save (bool): Whether to save the plot.
        """
        fig, ax = self.plt.subplots()
        total_bottom = np.zeros(num_iters)
        iterations = [str(i) for i in range(num_iters)]
        for label in label_set_counter:
            ax.bar(
                iterations, label_set_counter[label], bottom=total_bottom, label=label
            )
            total_bottom = np.add(total_bottom, label_set_counter[label])

        ax.set_ylabel("Counts")
        ax.set_xlabel("Iteration")
        level = "Intent" if intent_level else "Domain"
        title = f"{strategy}_{level}_Selection_Distribution_Per_Iteration"
        ax.set_title(title)
        self.plt.tight_layout()
        box = ax.get_position()
        ax.set_position([box.x0, box.y0, box.width * 2, box.height])
        ax.legend(loc="center left", bbox_to_anchor=(1, 0.5))

        if display:
            self.plt.show()
        if save:
            img_dir_path = os.path.join(*[self.experiment_dir_path, "plots", "overall"])
            os.makedirs(img_dir_path, exist_ok=True)
            img_path = os.path.join(img_dir_path, f"{title}.png")
            fig.savefig(img_path, bbox_inches="tight")
            self.plt.clf()


text:
# Standard library
import os
import abc
from collections import namedtuple

# Third-party
import numpy as np
from astropy import units as u
from astropy import constants
from astropy.convolution import convolve_fft
from fast_histogram import histogram2d

# Project
from ..util import check_units, check_random_state, data_dir
from ..filters import FilterSystem, get_filter_names
from ..source import Source
from ..log import logger

__all__ = ['IdealImager', 'ArtImager']

# return object for the ideal imager
IdealImage = namedtuple('IdealImage', 'image zpt bandpass')

# return object for the artificial imager
ArtImage = namedtuple(
    'ArtImage',
    'raw_counts src_counts sky_counts '
    'image var_image calibration '
    'zpt bandpass exptime'
)

mAB_0 = 48.6
def fnu_from_AB_mag(mag):
    """
    Convert AB magnitude into flux density fnu in cgs units.
    """
    fnu = 10.**((mag + mAB_0)/(-2.5))
    return fnu*u.erg/u.s/u.Hz/u.cm**2

class Imager(metaclass=abc.ABCMeta):
    """
    Base class for Imager objects.

    Parameters
    ----------
    phot_system : str or list-like
        Name of the photometric system(s).
    """

    def __init__(self, phot_system):
        self.phot_system = phot_system
        self.filters = get_filter_names(phot_system)

    def _check_source(self, source):
        """Verify that input object is a Source object."""
        if Source not in source.__class__.__mro__:
            raise Exception(f'{type(src)} is not a valid Source object')

    def _check_bandpass(self, bandpass):
        """Verify bandpass is the loaded filter list."""
        if bandpass not in self.filters:
            raise Exception(f'you do not seem to have {bandpass}-band mags')

    def inject(self, x, y, signal, xy_dim):
        """
        Inject sources into image.

        Parameters
        ----------
        x : `~numpy.ndarray`
            Source x coordinates.
        y : `~numpy.ndarray`
            Source y coordinates.
        signal : `~numpy.ndarray`
            The signal to inject into image (e.g., flux, counts, etc...).
        xy_dim : int or list-like
            The dimensions of mock image in xy units. If `int` is given, it is
            assumed to be both the x and y dimensions: (xy_dim, xy_dim).

        Returns
        -------
        image : `~numpy.ndarray`
            The mock image *before* psf convolution.
        """
        bins = tuple(np.asarray(xy_dim).astype(int))
        hist_range = [[0, bins[0] - 1], [0, bins[1]- 1]]
        image = histogram2d(x, y, bins=bins,
                            weights=signal, range=hist_range).T
        return image

    def apply_seeing(self, image, psf=None, boundary='wrap', **kwargs):
        """
        Convolve mock image with psf.

        Parameters
        ----------
        image : `~numpy.ndarray`
            The mock image.
        psf : `~numpy.ndarray` or None, optional
            The point-spread function. If None, will return the input image.
        boundary : {'fill', 'wrap'}, optional
            A flag indicating how to handle boundaries:

                * 'fill': set values outside the array boundary to fill_value
                  (default)
                * 'wrap': periodic boundary

            The `None` and 'extend' parameters are not supported for FFT-based
            convolution.

        Returns
        -------
        image : `~numpy.ndarray`
            The psf-convolved image. If `psf` is not `None`, the original image
            is returned.
        """
        if psf is not None:
            image = convolve_fft(image , psf,
                                 boundary=boundary,
                                 normalize_kernel=True)
        return image

    def observe(self, bandpass, psf, **kwargs):
        """Mock observe source."""
        return NotImplementedError()

class IdealImager(Imager):
    """Ideal imager for making noise-free images."""

    def observe(self, source, bandpass, psf=None, zpt=27, **kwargs):
        """
        Make ideal observation.

        Parameters
        ----------
        source : `~artpop.source.Source`
            Source to mock observer.
        bandpass : str
            Filter of observation. Must be a filter in the given
            photometric system(s).
        psf : `~numpy.ndarray` or None, optional
            The point-spread function. If None, will not psf-convolve image.
        zpt : float, optional
            The magnitude zero point of the mock image.

        .. note::
            The returned parameters are stored as attributes of a
            `~collections.namedtuple` object.

        Returns
        -------
        image : `~numpy.ndarray`
            The ideal mock image.
        bandpass : str
            Filter of the observation.
        zpt : float
            The magnitude zero point.
        """
        self._check_bandpass(bandpass)
        self._check_source(source)
        flux = 10**(0.4 * (zpt - source.mags[bandpass]))
        image = self.inject(source.x, source.y, flux, source.xy_dim)
        image = self.apply_seeing(image, psf, **kwargs)
        observation = IdealImage(image=image, bandpass=bandpass, zpt=zpt)
        return observation

class ArtImager(Imager):
    """
    Imager for making fully artificial images.

    Parameters
    ----------
    phot_system : str or list-like
        Name of the photometric system(s).
    diameter : float or `~astropy.units.Quantity`, optional
        Diameter of the telescope aperture. If a float is given, the units
        will be assumed to be meters.
    read_noise : float, optional
        RMS of Gaussian read noise. Set to zero by default.
    throughput : float, optional
        Throughput factor (e.g., to account for QE). Set to one by default.
        Note the filter response curves used by MIST already include
        atmospheric transmission if applicable.
    random_state : `None`, int, list of ints, or `~numpy.random.RandomState`
        If `None`, return the `~numpy.random.RandomState` singleton used by
        ``numpy.random``. If `int`, return a new `~numpy.random.RandomState`
        instance seeded with the `int`.  If `~numpy.random.RandomState`,
        return it. Otherwise raise ``ValueError``.
    """

    def __init__(self, phot_system, diameter=10, read_noise=0.0, throughput=1,
                 random_state=None, **kwargs):
        super(ArtImager, self).__init__(phot_system)
        self.throughput = throughput
        self.read_noise = read_noise
        self.diameter = check_units(diameter, 'm')
        self.rng = check_random_state(random_state)
        self.filter_system = FilterSystem(self.phot_system, **kwargs)

    @property
    def area(self):
        """Area of aperture."""
        r = self.diameter / 2
        return np.pi * r**2

    def mag_to_counts(self, mags, bandpass, exptime):
        """
        Convert magnitudes to counts.

        Parameters
        ----------
        mags : `~numpy.ndarray` or `~astropy.table.Column`
            AB magnitudes to be converted into counts.
        bandpass : str
            Filter of observation. Must be a filter in the given
            photometric system(s).
        exptime : float or `~astropy.units.Quantity`
            Exposure time. If float is given, the units are assumed to
            be `~astropy.units.second`.

        Returns
        -------
        counts : `~numpy.ndarray`
            The counts associated with each magnitude.
        """
        exptime = check_units(exptime, 's')
        fnu = fnu_from_AB_mag(mags)
        dlam = self.filter_system.dlam(bandpass)
        lam_eff = self.filter_system.lam_eff(bandpass)
        photon_flux = (dlam / lam_eff) * fnu / constants.h.to('erg s')
        counts = photon_flux * self.area.to('cm2') * exptime.to('s')
        counts = counts.decompose()
        assert counts.unit == u.dimensionless_unscaled
        counts *= self.throughput
        return counts.value

    def mu_to_counts_per_pixel(self, mu, bandpass, exptime, pixel_scale):
        """
        Convert a constant surface brightness into counts per pixel.

        Parameters
        ----------
        mu : float
            Surface brightness in units of `~astropy.units.mag` per square
            `~astropy.units.arcsec`.
        bandpass : str
            Filter of observation. Must be a filter in the given
            photometric system(s).
        exptime : float or `~astropy.units.Quantity`
            Exposure time. If float is given, the units are assumed to
            be `~astropy.units.second`.
        pixel_scale : `~astropy.units.Quantity`
            Pixel scale.

        Returns
        -------
        counts_per_pixel : float
            Counts per pixel associated with the given surface brightness.
        """
        exptime = check_units(exptime, 's')
        dlam = self.filter_system.dlam(bandpass)
        lam_eff = self.filter_system.lam_eff(bandpass)
        pixel_scale = pixel_scale.to('arcsec / pixel')
        E_lam = (constants.h * constants.c / lam_eff).decompose().to('erg')
        fnu_per_square_arcsec = fnu_from_AB_mag(mu) / u.arcsec**2
        flam_per_square_arcsec = fnu_per_square_arcsec *\
            constants.c.to('angstrom/s') / lam_eff**2
        flam_per_pixel = flam_per_square_arcsec * pixel_scale**2
        photon_flux_per_sq_pixel = (flam_per_pixel * dlam / E_lam).\
            decompose().to('1/(cm2*pix2*s)')
        counts_per_pixel = photon_flux_per_sq_pixel * exptime.to('s')
        counts_per_pixel *= self.area.to('cm2') * u.pixel**2
        assert counts_per_pixel.unit == u.dimensionless_unscaled
        counts_per_pixel *= self.throughput
        return counts_per_pixel.value

    def calibration(self, bandpass, exptime, zpt=27.0):
        """
        Calculate the calibration factor, which converts counts into
        calibrated flux units.

        Parameters
        ----------
        bandpass : str
            Filter of observation. Must be a filter in the given
            photometric system(s).
        exptime : float or `~astropy.units.Quantity`
            Exposure time. If float is given, the units are assumed to
            be `~astropy.units.second`.
        zpt : float, optional
            The magnitude zero point of the mock image.

        Returns
        -------
        cali_factor : float
            The calibration factor.
        """
        fs = self.filter_system
        dlam_over_lam = fs.dlam(bandpass) / fs.lam_eff(bandpass)
        lam_factor = (dlam_over_lam / constants.h.cgs).value
        cali_factor = 10**(0.4 * zpt) * 10**(0.4 * mAB_0) / lam_factor
        cali_factor /= exptime.to('s').value
        cali_factor /= self.area.to('cm2').value
        return cali_factor

    def observe(self, source, bandpass, exptime, psf=None, zpt=27.0,
                mu_sky=None, **kwargs):
        """
        Make artificial observation.

        .. note::
            The returned parameters are stored as attributes of a
            `~collections.namedtuple` object.

        Parameters
        ----------
        source : `~artpop.source.Source`
            Artificial source to be observed.
        bandpass : str
            Filter of observation. Must be a filter in the given
            photometric system(s).
        exptime : float or `~astropy.units.Quantity`
            Exposure time. If `float` is given, the units are assumed to
            be `~astropy.units.second`.
        psf : `~numpy.ndarray` or None, optional
            The point-spread function. If None, will not psf-convolve image.
        zpt : float, optional
            The magnitude zero point of the mock image.
        mu_sky : float or None, optional
            Constant surface brightness of the sky. If None is given, then no
            sky noise will be added.

        Returns
        -------
        raw_counts : `~numpy.ndarray`
            Raw count image, including Poisson noise.
        src_counts : `~numpy.ndarray`
            Source count image before Poission noise is added.
        sky_counts : float
            Counts per pixel from the sky.
        var_image : `~numpy.ndarray`
            Variance image.
        calibration : float
            Calibration factor that converts counts to calibrated flux units.
        bandpass : str
            Filter of the observation.
        zpt : float
            The magnitude zero point.
        exptime : `~astropy.units.Quantity`
            The exposure time of the mock observation.
        """
        self._check_bandpass(bandpass)
        self._check_source(source)
        exptime = check_units(exptime, 's')
        if mu_sky is not None:
            sky_counts = self.mu_to_counts_per_pixel(
                mu_sky, bandpass, exptime, source.pixel_scale)
        else:
            sky_counts = 0
        counts = self.mag_to_counts(source.mags[bandpass], bandpass, exptime)
        src_counts = self.inject(source.x, source.y, counts, source.xy_dim)
        src_counts = self.apply_seeing(src_counts, psf, **kwargs)
        if mu_sky is None:
            src_counts[src_counts < 0] = 0
        raw_counts = self.rng.poisson(src_counts + sky_counts)
        if self.read_noise > 0.0:
            rn = self.rng.normal(scale=self.read_noise, size=src_counts.shape)
            raw_counts = raw_counts + rn
        cali = self.calibration(bandpass, exptime, zpt)
        image_cali = (raw_counts - sky_counts) * cali
        var = raw_counts + self.read_noise**2
        observation = ArtImage(raw_counts=raw_counts,
                               src_counts=src_counts,
                               sky_counts=sky_counts,
                               image=image_cali,
                               var_image=var,
                               calibration=cali,
                               zpt=zpt, bandpass=bandpass,
                               exptime=exptime)

        return observation

ext: py

reason: import numpy,from astropy


text:
from readfiles import *
from sklearn.preprocessing import normalize
from numpy import *
fromyears =2016

def compute(vals):
    scorediff = (int(vals[3]) - int(vals[5]))
    return (scorediff),0.

def winner(t1,t2,X):
    score1 = X[keyToEnum[t1]][keyToEnum[t2]]
    score2 = X[keyToEnum[t2]][keyToEnum[t1]]
    if score1>score2:
        return t1
    return t2

teams = readteams()
incidencematrix = readresults(teams,fromyears,compute)
bracket = readbrackets()

mininc = incidencematrix.min()
X = incidencematrix #- mininc
diff = False
while not diff:
    tmp = normalize(dot(X,X))
    diff = allclose(tmp,X)
    X = tmp

rounds = 6
for i in range(rounds):
    print [teams[ids] for ids in bracket]
    newround = []
    print "---------------"
    for ididx in xrange(0,len(bracket),2):
        ids1,ids2 = bracket[ididx],bracket[ididx+1]
        wid = winner(ids1,ids2,X)
        newround.append(wid)
    bracket = newround
print [teams[ids] for ids in bracket]

reason: from numpy


text:
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pymsm.multi_state_competing_risks_model import PathObject
from pymsm.statistics import paths_to_timestep_matrix, get_state_timestep_probs
from lifelines import AalenJohansenFitter
from typing import List, Dict
import warnings

def competingrisks_stackplot(
    data: pd.DataFrame,
    duration_col: str,
    event_col: str,
    order_top: List = None,
    order_bottom: List = None,
    times: np.ndarray = None,
    state_labels: Dict = None,
    fontsize: int = 18,
    ax=None,
):
    """Plot a stackplot for a competing risks dataset.

    Args:
        data (pd.DataFrame): dataset to plot
        duration_col (str): duration column name
        event_col (str): event column name
        order_top (List, optional): Order of the states to plot from the top. Defaults to None.
        order_bottom (List, optional): Order of the states to plot from the bottom. Defaults to None.
        times (np.ndarray, optional): manual times for x axis. Defaults to None.
        state_labels (Dict, optional): _description_. Defaults to None.
        fontsize (int, optional): _description_. Defaults to 18.
        ax (_type_, optional): Matplotlib ax to plot to. Defaults to None.

    Returns:
        _type_: _description_
    """
    if times is None:
        times = np.sort(data[duration_col].unique())

    failure_types = np.sort(data[event_col].unique())
    failure_types = failure_types[failure_types != 0]

    if (order_top is None) & (order_bottom is None):
        order_bottom = failure_types
    if order_top is None:
        order_top = []
    if order_bottom is None:
        order_bottom = []

    if state_labels is None:
        state_labels = dict(zip(failure_types, [str(f) for f in failure_types]))

    if ax is None:
        fig, ax = plt.subplots(1, 1)
        ax.set_ylim(0, 1)
        ax.set_xlim(0, times[-1])
        ax.set_ylabel("Probability", fontsize=fontsize)
        ax.set_xlabel("t", fontsize=fontsize)
        # ax.set_title(f"Starting from {state_labels[origin_state]}", fontsize=fontsize + 4)

    cumulative_densities = dict(zip(failure_types, ([0] * len(failure_types))))
    for failure_type in failure_types:
        T = data[duration_col]
        E = data[event_col]
        with warnings.catch_warnings():  # ignore warnings related to AJ Fitter
            warnings.simplefilter("ignore")
            ajf = AalenJohansenFitter()
            ajf.fit(T, E, event_of_interest=failure_type)
            cumulative_densities[failure_type] = ajf.predict(times, interpolate=True)

    cifs_top = []
    for i, failure_type in enumerate(order_top):
        color = f"C{failure_type}"
        cif = 1 - cumulative_densities[failure_type].values
        if i == 0:
            ax.fill_between(times, 1, cif, alpha=0.8, color=color)
            ax.text(
                x=times[-1] * 1.02,
                y=(cif[-1] + (1 - cif[-1]) / 2),
                s=state_labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        else:
            cif = cif - cifs_top[i - 1]
            ax.fill_between(times, cifs_top[i - 1], cif, alpha=0.8, color=color)
            ax.text(
                x=times[-1] * 1.02,
                y=(cif[-1] + (cifs_top[i - 1][-1] - cif[-1]) / 2),
                s=state_labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        ax.plot(times, cif, color="k")
        cifs_top.append(cif)

    cifs_bottom = []
    for i, failure_type in enumerate(order_bottom):
        color = f"C{failure_type}"
        cif = cumulative_densities[failure_type].values
        if i == 0:
            ax.fill_between(times, 0, cif, alpha=0.8, color=color)
            ax.text(
                x=times[-1] * 1.02,
                y=((cif[-1]) / 2),
                s=state_labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        else:
            cif = cif + cifs_bottom[i - 1]
            ax.fill_between(
                times,
                cifs_bottom[i - 1],
                cif,
                alpha=0.8,
                label=state_labels[failure_type],
                color=color,
            )
            ax.text(
                x=times[-1] * 1.02,
                y=(cif[-1] + (cifs_bottom[i - 1][-1] - cif[-1]) / 2),
                s=state_labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        ax.plot(times, cif, color="k")
        cifs_bottom.append(cif)
    return ax

def stackplot_state_timesteps(
    state_timestep_probs: np.ndarray,
    order_top: List = [],
    order_bottom: List = [],
    times: np.ndarray = None,
    labels: Dict = None,
    fontsize: int = 18,
    ax=None,
):
    failure_types = list(state_timestep_probs.keys())
    if labels is None:
        labels = dict(zip(failure_types, [str(f) for f in failure_types]))

    if times is None:
        times = np.arange(len(state_timestep_probs[failure_types[0]]))

    if ax is None:
        fig, ax = plt.subplots(1, 1)
        ax.set_ylim(0, 1)
        ax.set_xlim(0, times[-1])
        ax.set_ylabel("Probability", fontsize=fontsize)
        ax.set_xlabel("t", fontsize=fontsize)

    cifs_top = []
    for i, failure_type in enumerate(order_top):
        color = f"C{failure_type}"
        cif = 1 - state_timestep_probs[failure_type]
        if i == 0:
            ax.fill_between(times, 1, cif, alpha=0.8, color=color)
            ax.text(
                x=times[-1] * 1.02,
                y=(cif[-1] + (1 - cif[-1]) / 2),
                s=labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        else:
            cif = cif - cifs_top[i - 1]
            ax.fill_between(times, cifs_top[i - 1], cif, alpha=0.8, color=color)
            ax.text(
                x=times[-1] * 1.02,
                y=(cif[-1] + (cifs_top[i - 1][-1] - cif[-1]) / 2),
                s=labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        ax.plot(times, cif, color="k")
        cifs_top.append(cif)

    cifs_bottom = []
    for i, failure_type in enumerate(order_bottom):
        color = f"C{failure_type}"
        cif = state_timestep_probs[failure_type]
        if i == 0:
            ax.fill_between(times, 0, cif, alpha=0.8, color=color)
            ax.text(
                x=times[-1] * 1.02,
                y=((cif[-1]) / 2),
                s=labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        else:
            cif = cif + cifs_bottom[i - 1]
            ax.fill_between(
                times,
                cifs_bottom[i - 1],
                cif,
                alpha=0.8,
                label=labels[failure_type],
                color=color,
            )
            ax.text(
                x=times[-1] * 1.02,
                y=(cif[-1] + (cifs_bottom[i - 1][-1] - cif[-1]) / 2),
                s=labels[failure_type],
                fontsize=fontsize,
                color=color,
            )
        ax.plot(times, cif, color="k")
        cifs_bottom.append(cif)

def stackplot_state_timesteps_from_paths(
    paths: List[PathObject],
    max_timestep: int,
    order_top: List,
    order_bottom: List,
    labels: Dict = None,
    ax=None,
):
    timestep_matrix = paths_to_timestep_matrix(paths, max_timestep)
    state_timestep_probs = get_state_timestep_probs(timestep_matrix)
    stackplot_state_timesteps(
        state_timestep_probs, order_top, order_bottom, labels=labels, ax=ax
    )

max_stars_repo_name: hrossman/pymsm

max_issues_repo_name: hrossman/pymsm

max_forks_repo_name: hrossman/pymsm

reason: import numpy


text:
import time

from metadrive import MetaDriveEnv

if __name__ == '__main__':
    env = MetaDriveEnv(dict(environment_num=1000, traffic_density=0.1, start_seed=5000))
    obs = env.reset()
    start = time.time()
    vc = []
    for s in range(1000):
        env.reset(force_seed=s + 5000)
        print("We have {} vehicles in seed {} map!".format(len(env.engine.traffic_manager.vehicles), s))
        vc.append(len(env.engine.traffic_manager.vehicles))
        if (s + 1) % 1 == 0:
            print(f"{s + 1} | Time Elapse: {time.time() - start}")
    import numpy as np
    print(np.mean(vc), np.min(vc), np.max(vc), np.std(vc))

max_stars_repo_name: liuzuxin/metadrive

max_issues_repo_name: liuzuxin/metadrive

max_forks_repo_name: liuzuxin/metadrive


text:
"""
interval.py
--------------

Deal with 1D intervals which are defined by:
  [start position, end position]
"""

import numpy as np

def check(a, b, digits):
    """
    Check input ranges, convert them to vector form,
    and get a fixed precision integer version of them.

    Parameters
    --------------
    a : (2, ) or (2, n) float
      Start and end of a 1D interval
    b : (2, ) or (2, n) float
      Start and end of a 1D interval
    digits : int
      How many digits to consider

    Returns
    --------------
    a : (2, n) float
      Ranges as vector
    b : (2, n) float
      Ranges as vector
    a_int : (2, n) int64
      Ranges rounded to digits, as vector
    b_int : (2, n) int64
      Ranges rounded to digits, as vector
    is_1D : bool
      If True, input was single pair of ranges
    """
    a = np.array(a, dtype=np.float64)
    b = np.array(b, dtype=np.float64)

    if a.shape != b.shape or a.shape[-1] != 2:
        raise ValueError('ranges must be identical and (2,)!')

    # if input was single interval reshape it here
    is_1D = False
    if len(a.shape) == 1:
        a = a.reshape((-1, 2))
        b = b.reshape((-1, 2))
        is_1D = True

    # make sure ranges are sorted
    a.sort(axis=1)
    b.sort(axis=1)

    # compare in fixed point as integers
    a_int = (a * 10**digits).round().astype(np.int64)
    b_int = (b * 10**digits).round().astype(np.int64)

    return a, b, a_int, b_int, is_1D

def intersection(a, b, digits=8):
    """
    Given a pair of ranges, merge them in to
    one range if they overlap at all

    Parameters
    --------------
    a : (2, ) float
      Start and end of a 1D interval
    b : (2, ) float
      Start and end of a 1D interval
    digits : int
      How many digits to consider

    Returns
    --------------
    intersects : bool or (n,) bool
      Indicates if the ranges overlap at all
    new_range : (2, ) or (2, 2) float
      The unioned range from the two inputs,
      or both of the original ranges if not overlapping
    """
    # check shape and convert
    a, b, a_int, b_int, is_1D = check(a, b, digits)

    # what are the starting and ending points of the overlap
    overlap = np.zeros(a.shape, dtype=np.float64)

    # A fully overlaps B
    current = np.logical_and(a_int[:, 0] <= b_int[:, 0],
                             a_int[:, 1] >= b_int[:, 1])
    overlap[current] = b[current]

    # B fully overlaps A
    current = np.logical_and(a_int[:, 0] >= b_int[:, 0],
                             a_int[:, 1] <= b_int[:, 1])
    overlap[current] = a[current]

    # A starts B ends
    # A:, 0   B:, 0     A:, 1        B:, 1
    current = np.logical_and(
        np.logical_and(a_int[:, 0] <= b_int[:, 0],
                       b_int[:, 0] < a_int[:, 1]),
        a_int[:, 1] < b_int[:, 1])
    overlap[current] = np.column_stack([b[current][:, 0],
                                        a[current][:, 1]])

    # B starts A ends
    # B:, 0  A:, 0    B:, 1  A:, 1
    current = np.logical_and(
        np.logical_and(b_int[:, 0] <= a_int[:, 0],
                       a_int[:, 0] < b_int[:, 1]),
        b_int[:, 1] < a_int[:, 1])
    overlap[current] = np.column_stack([a[current][:, 0],
                                        b[current][:, 1]])

    # is range overlapping at all
    intersects = overlap.ptp(axis=1) > 10**-digits

    if is_1D:
        return intersects[0], overlap[0]

    return intersects, overlap

max_stars_repo_name: hawkaa/trimesh

max_issues_repo_name: hawkaa/trimesh

max_forks_repo_name: hawkaa/trimesh


text:
import matplotlib.pyplot as plt
import random
import numpy as np
import math

from environments.environment import Environment
from environments.env_generation import SingleEnvironmentGenerator

class SimpleTurnEnvironment(Environment):
    '''
    Environment where the agent takes on a single pixel.
    The environment has three possible actions each turn.
        - turn left
        - go forward
        - turn right
    The agent will turn with 45 degree intervals. When the agent
    is turned, the agent will go to the diagonal adjacent pixel.
    The state that the agent perceives is turned according to its
    rotation.
    '''
    def __init__(self, generator):
        self.angle_count = 0
        super().__init__(generator)

    def get_nb_actions(self):
        return 3

    def reset(self):
        self.done = False
        self.env_info = self.generator.generate_environment(extra_spacing=True)
        self.current_pos = random.sample(self.env_info.start_positions, 1)[0]
        self.visited_tiles = np.zeros_like(self.env_info.get_obstacle_map())
        self.visited_tiles[self.current_pos] = 1
        self.nb_steps = 0
        self.total_reward = 0
        self.total_terrain_diff = 0

        self.angle_count = self.get_random_angle_count()

        return self.get_state()

    def get_state(self):
        dim_x, dim_y = self.get_dimension()
        x = np.linspace(0.5, dim_x - 0.5, dim_x) - (dim_x / 2)
        y = np.linspace(0.5, dim_y - 0.5, dim_y) - (dim_y / 2)

        yy, xx = np.meshgrid(x, y)
        angle = self.angle_count * (math.pi / 4)

        xx_rot = math.cos(angle) * xx - math.sin(angle) * yy + (dim_x / 2)
        yy_rot = math.sin(angle) * xx + math.cos(angle) * yy + (dim_y / 2)

        extra_spacing_x = (dim_x - 1) // 5
        extra_spacing_y = (dim_y - 1) // 5

        xx_idxs = np.floor(xx_rot).astype(int) + extra_spacing_x
        yy_idxs = np.floor(yy_rot).astype(int) + extra_spacing_y

        full_obstacle_map = self.env_info.get_obstacle_map(extra_spacing=True)
        curr_obstacle_map = full_obstacle_map[(xx_idxs, yy_idxs)]

        full_coverage_map = np.zeros((dim_x + 2 * extra_spacing_x, dim_y + 2 * extra_spacing_y))
        full_coverage_map[extra_spacing_x:-extra_spacing_x, extra_spacing_y:-extra_spacing_y] = self.visited_tiles
        curr_coverage_map = full_coverage_map[(xx_idxs, yy_idxs)]

        curr_pos_map = np.zeros((dim_x, dim_y))
        if not self.env_info.get_obstacle_map()[self.current_pos] == 1:
            curr_pos_x = self.current_pos[0] + 0.5 - (dim_x / 2)
            curr_pos_y = self.current_pos[1] + 0.5 - (dim_y / 2)
            rot_pos_x = curr_pos_x * math.cos(angle) + curr_pos_y * math.sin(angle)
            rot_pos_y = -curr_pos_x * math.sin(angle) + curr_pos_y * math.cos(angle)
            curr_pos_map[
                math.floor(rot_pos_x + dim_x / 2),
                math.floor(rot_pos_y + dim_y / 2)
            ] = 1.0

        if self.gives_terrain_info():
            full_terrain_map = self.env_info.get_terrain_map(extra_spacing=True)
            curr_terrain_map = full_terrain_map[(xx_idxs, yy_idxs)]

            return [curr_pos_map, curr_coverage_map, curr_obstacle_map, curr_terrain_map]

        return [curr_pos_map, curr_coverage_map, curr_obstacle_map]

    def get_random_angle_count(self):
        rand_value = np.random.randint(0, 8)
        valid = False

        while not valid:
            n_pos = SimpleTurnEnvironment.get_new_position(self.current_pos, rand_value)
            if self.env_info.get_obstacle_map()[n_pos] == 0:
                valid = True
            else:
                rand_value = np.random.randint(0, 8)

        return rand_value

    @staticmethod
    def get_new_position(curr_pos, angle_cnt):
        n_x = curr_pos[0]
        if 4 > angle_cnt > 0:
            n_x -= 1
        if 4 < angle_cnt < 8:
            n_x += 1

        n_y = curr_pos[1]
        turned_angle = (angle_cnt + 2) % 8
        if 4 > turned_angle > 0:
            n_y += 1
        if 4 < turned_angle < 8:
            n_y -= 1

        return (n_x, n_y)

    def get_extra_info(self):
        dim_x, dim_y = self.get_dimension()
        angle = self.angle_count * (math.pi / 4)

        curr_pos_x = self.current_pos[0] + 0.5 - (dim_x / 2)
        curr_pos_y = self.current_pos[1] + 0.5 - (dim_y / 2)

        rot_pos_x = curr_pos_x * math.cos(angle) + curr_pos_y * math.sin(angle)
        rot_pos_y = -curr_pos_x * math.sin(angle) + curr_pos_y * math.cos(angle)

        extra = super().get_extra_info()
        extra.update({
            'rotation': self.angle_count * math.pi / 4,
            'rot_pos': (math.floor(rot_pos_x + (dim_x / 2)),
                        math.floor(rot_pos_y + (dim_y / 2)))
        })

        return extra

    def step(self, action):
        n_position = self.current_pos
        if action == 0:
            self.angle_count = (self.angle_count + 1) % 8

        elif action == 1:
            n_position = self.get_new_position(self.current_pos, self.angle_count)

        elif action == 2:
            self.angle_count = (self.angle_count - 1) % 8

        self.nb_steps += 1

        reward = self.get_reward(n_position)
        self.total_reward += reward
        done = self.is_done(n_position)

        if self.env_info.get_obstacle_map()[n_position] == 0 and self.visited_tiles[n_position] == 0:
            self.visited_tiles[n_position] = 1

        self.current_pos = n_position

        info = {
            "reward": reward,
            "total reward": self.total_reward,
            "nb steps": self.nb_steps,
            "nb visited tiles": self.get_nb_visited_tiles(),
            "complete coverage": self.complete_coverage(),
        }
        if self.env_info.has_terrain_info():
            terrain_diff = abs(
                self.env_info.terrain_map[self.current_pos] - self.env_info.terrain_map[n_position]
            )
            self.total_terrain_diff += terrain_diff
            info["terrain diff"] = terrain_diff
            info["total terrain diff"] = self.total_terrain_diff
        info.update(self.get_extra_info())

        return self.get_state(), reward, done, info

if __name__ == "__main__":
    pass

max_stars_repo_name: JonasVervloet/RL-Coverage-Planner

max_issues_repo_name: JonasVervloet/RL-Coverage-Planner

max_forks_repo_name: JonasVervloet/RL-Coverage-Planner


text:
#!/usr/bin/env python3
# coding=UTF-8
#
# BSD 2-Clause License
#
# Copyright (c) 2019, Jason Leake
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice, this
#    list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Plot stats from ax3
#
from tkinter import filedialog
import argparse
import configparser
import csv
import matplotlib.dates as mdate
import matplotlib.pyplot as plt
import numpy as np
import os
import sys
import tkinter as tk

class PlotMinutes:
    """ Produce the minutes plots
    """

    def _plot(self, data, index, xlines, title, startEpoch,
              outputfile, showtime, ymin, ymax, grid):
        """ Generate the actual plot. Returns outputfile
        """
        fig, axis = plt.subplots()
        if ymax is not None or ymin is not None:
            plt.ylim(ymin, ymax)

        axis.grid(grid)
        axis.set_title(title)
        axis.set_ylabel("Acceleration (g)")

        if showtime:
            # Index 0 of data is epoch time
            seconds = mdate.epoch2num(data[:, [0]])
            dateFormat = "%H:%M"
            dateFormatter = mdate.DateFormatter(dateFormat)
            locator = mdate.HourLocator(interval=6)
            axis.xaxis.set_major_locator(locator)
            axis.xaxis.set_minor_locator(mdate.HourLocator())
            locator.MAXTICKS = 5000
            axis.xaxis.set_major_formatter(dateFormatter)

            axis.set_xlabel("Time")
            for line in xlines:
                # Value is minute, baselined to start of run
                epoch = float(line.strip()) * 60 + float(startEpoch)
                plt.axvline(mdate.epoch2num(epoch), c="red")

            plt.plot(seconds, data[:, [index]])
        else:
            for line in xlines:
                line = int(line.strip())
                plt.axvline(line, c="red")

            axis.set_xlabel("Minute")
            # Index 1 of data is minutes from start
            plt.plot(data[:, [1]], data[:, [index]])
        plt.savefig(outputfile)
        plt.draw()
        plt.close()
        return outputfile

    def fileTitles(self):
        """ Return list of file title components.  If empty, no plot should be produced
        """
        # Order of fields is MPRS
        return [ "",
                 "",
                 "",
                 "mean_x",
                 "peak_to_peak_x",
                 "rms_x",
                 "std_dev_x",
                 "mean_y",
                 "peak_to_peak_y",
                 "rms_y",
                 "std_dev_y",
                 "mean_z",
                 "peak_to_peak_z",
                 "rms_z",
                 "std_dev_z",
                 "mean_total",
                 "peak_to_peak_total",
                 "rms_total",
                 "std_dev_total"]

    def __call__(self, filename, controlFile=None, showtime=False,
                 ymin=None, ymax=None, selectedPlot=None, grid=False):
        """Process a CSV file of per-minute data as produced by ax3_stats.py.
        controlFile, if specified, is an INI file that controls the
        generation of short plots as well as the normal ones. It
        contains the following in the section [PLOT] -- start is the
        minute to start "short" plot, end -- is the minute to end the
        "short" plot

        """

        # e.g. fred/minutes_2020-10-20.csv gives path = fred, suffix =
        # 2020-10-20
        path, name = os.path.split(filename)
        prefix = os.path.splitext(name)[0]
        suffixIndex = len(prefix.split("_"))
        suffix = prefix.split("_")[suffixIndex - 1]
        if suffix is None:
            suffix = ""

        outputFiles = []

        config = configparser.ConfigParser()
        if controlFile is not None:
            if os.path.exists(controlFile):
                config.read(controlFile)
            else:
                print(f"Control file {controlFile} not found", file=sys.stderr)
                sys.exit(1)

        # Numpy CSV reader
        data = np.genfromtxt(filename, delimiter=",")

        # Baselined flag is in the last field of the CSV file.  Only look
        # at the first row since the value is the same throughout the file
        baselined = data[0][19] != 0

        # These are the fields in the CSV file, except for the last,
        # is_baselined one.  We don't get as far as processing that
        # because we stop when we have exhausted the fileTitle list
        # below

        # Order of fields is MPRS
        title = [ "epoch",
                  "minute",
                  "size",
                  "Mean of x acceleration over minute",
                  "peak to peak of x acceleration over minute",
                  "RMS of x acceleration over minute",
                  "std dev of x acceleration over minute",
                  "Mean of y acceleration over minute",
                  "peak to peak of y acceleration over minute",
                  "RMS of y acceleration over minute",
                  "std dev of y acceleration over minute",
                  "Mean of z acceleration over minute",
                  "peak to peak of z acceleration over minute",
                  "RMS of z acceleration over minute",
                  "std dev of z acceleration over minute",
                  "Mean of tot acceleration over minute",
                  "peak to peak of tot acceleration over minute",
                  "RMS of tot acceleration over minute",
                  "std dev of tot acceleration over minute"]

        fileTitle = self.fileTitles()

        # All the interesting stuff happens in the first few mins in
        # some datasets.  This is the code for producing plots which
        # are limited and annotated by the optional config file
        if config.has_section("PLOT"):
            if config.has_option("PLOT", "start"):
                start = int(config.get("PLOT", "start"))
            else:
                start = 0

            if config.has_option("PLOT", "end"):
                end = int(config.get("PLOT", "end"))
            else:
                start = len(data) - 1

            if config.has_option("PLOT", "xlines"):
                xlines = config.get("PLOT","xlines")
            else:
                xlines = None
            # xlinesList is the list of X values that have vertical
            # red lines displayed
            xlinesList = xlines.split(',')
            # The subset of points that are to be plotted
            truncated = np.array(data[start:end, :])

            for index in range(3, len(title)):
                if selectedPlot is not None:
                    if selectedPlot != fileTitle[index]:
                        continue

                plotTitle = title[index]
                if baselined:
                    plotTitle = "Baselined " + plotTitle

                outputFile = self.makeOutFile(baselined,
                                              path,
                                              "subsection_" +
                                              fileTitle[index] + "_" + suffix)

                outputFiles.append(self._plot(truncated, index, xlinesList,
                                              plotTitle,
                                              data[0, 0], # epoch time of start of file
                                              outputFile,
                                              showtime, ymin, ymax, grid))
        else:
            # Create an empty array so that nothing is plotted
            # in the next block for the full dataset
            xlinesList = []

        # For the full dataset
        for index in range(3, len(title)):
            if selectedPlot is not None:
                if selectedPlot != fileTitle[index]:
                    continue

            plotTitle = title[index]
            if baselined:
                plotTitle = "Baselined " + plotTitle

            outputFile = self.makeOutFile(baselined,
                                          path,
                                          "plot_" +
                                          fileTitle[index] + "_" + suffix)

            outputFiles.append(self._plot(data, index, xlinesList,
                                          plotTitle, data[0, 0],
                                          outputFile,
                                          showtime, ymin, ymax, grid))

    def makeOutFile(self, baselined, path, filename):
        """ Make output filename """
        if baselined:
            filename = "baselined_" + filename
        fullPath = os.path.join(path, filename + ".pdf")
        print("Output file is", fullPath)
        return fullPath

def main():
    controlFile = None
    if len(sys.argv) < 2:
        root = tk.Tk()
        root.withdraw()
        filePath = filedialog.askopenfilename(
            filetypes = [("Comma separated file (CSV) format",".csv")])
        controlFile = None
        showtime = None
        ymin = None
        ymax = None
        selectedPlot = None
        grid = None
    else:
        parser = argparse.ArgumentParser(description=
                                         "Plot statistics for accelerometer file")
        parser.add_argument("filename", help="Input filename")
        parser.add_argument("--controlfile", nargs="?", help="INI file to control plotting")
        parser.add_argument("--select", nargs="?", help="INI file to control plotting")
        parser.add_argument("--showtime", help="X axis is actual time", action="store_true")
        parser.add_argument("--ymin", help="Y axis minimum", type=float, default=None)
        parser.add_argument("--ymax", help="Y axis maximum", type=float, default=None)
        parser.add_argument("--grid", help="Add grid to plot", action="store_true")
        args = parser.parse_args()
        filePath = args.filename
        controlFile = args.controlfile
        showtime = args.showtime
        ymin = args.ymin
        ymax = args.ymax
        selectedPlot = args.select
        grid = args.grid

    plotMinutes = PlotMinutes()
    plotMinutes(filePath, controlFile, showtime,
                ymin, ymax, selectedPlot, grid)

if __name__ == "__main__":
    main()

max_stars_repo_name: isopleth/kinetics

max_issues_repo_name: isopleth/kinetics

max_forks_repo_name: isopleth/kinetics


text:
import pytest
from pathlib import Path
import numpy as np
import os
import scipy.io as sio
from layout_data.utils.convert import mat2h5

@pytest.fixture(scope="session")
def prepare_data_path(tmpdir_factory):
    tmpdir = Path(tmpdir_factory.mktemp("data"))

    num = 20
    shape = (200, 200)
    for subdir in ["train", "test"]:
        # subdir_path = os.path.join(tmpdir, subdir)
        subdir_path = tmpdir / subdir
        os.mkdir(subdir_path)
        for i in range(num):
            u = np.random.randn(*shape)
            F = np.random.randn(*shape)

            # path = os.path.join(subdir_path, f'{i}.mat')
            path = subdir_path / f"{i}.mat"
            sio.savemat(str(path), {"u": u, "F": F})

    return tmpdir, num, shape

@pytest.fixture(scope="session")
def prepare_datah5_path(prepare_data_path):
    path, num, shape = prepare_data_path
    path = path / "train"
    fn = "h5.h5"
    path_h5 = path / fn
    mat2h5(path, path_h5)
    return path, fn, num, shape

max_stars_repo_name: zweien/layout-data

max_issues_repo_name: zweien/layout-data

max_forks_repo_name: zweien/layout-data

reason: import numpy,import scipy


text:
"""
Utilities for .jpg and .jpeg data
"""

#  Copyright 2019 Gabriele Valvano
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

from PIL import Image
import numpy as np

def get_jpg_image(filename):
    """
    Loads JPEG image into 3D Numpy array of shape (width, height, channels).

    Args:
        filename (str): file name
    """
    with Image.open(filename) as image:
        im_arr = np.array(image)
    return im_arr

def save_jpg_image(array, filename):
    """
    Saves JPEG image from array 3D Numpy array of shape (width, height, channels).

    Args:
        array (np.array): array to save
        filename (str): file name

    """
    img = Image.fromarray(array)
    img.save(filename)

max_stars_repo_name: gvalvano/idas

max_forks_repo_name: gvalvano/idas


text:
from typing import List, Optional

import numpy as np
from scipy.special import logsumexp

from naive_bayes.distributions import Bernoulli, Categorical, Normal
from naive_bayes.models.abstract import AbstractModel

# TODO: add str parametrization for ExtendedNaiveBayes
class ExtendedNaiveBayes(AbstractModel):
    """
    Extended (allow different distributions for each feature) Naive Bayes model.
    """

    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
        """
        Method to fit the model.

        :param np.ndarray X: training data.
        :param np.ndarray y: target values.
        """

        self._check_input_data(X=X, y=y)

        # priors
        _, counts = np.unique(y, return_counts=True)
        self.priors = counts / counts.sum()

        # distributions
        for feature in range(len(self.distributions)):
            self.distributions[feature].fit(X[:, feature], y)  # type: ignore

    def predict_log_proba(self, X: np.ndarray) -> np.ndarray:
        """
        Method to compute class log probabilities.

        :param np.ndarray X: training data.
        :return: class log probabilities.
        :rtype: np.ndarray
        """

        self._check_input_data(X=X)

        n_samples = X.shape[0]

        log_prob_y = np.log(self.priors)
        log_prob_xy = np.repeat(log_prob_y[np.newaxis, :], repeats=n_samples, axis=0)

        for feature in range(len(self.distributions)):
            log_prob_xy += self.distributions[feature].predict_log_proba(X[:, feature])  # type: ignore

        log_prob_x = logsumexp(log_prob_xy, axis=1)
        log_prob_y_x = log_prob_xy - log_prob_x[:, np.newaxis]

        return log_prob_y_x

    def sample(self, n_samples: int, random_state: Optional[int] = None) -> np.ndarray:
        """
        Generate samples from fitted data.

        :param int n_samples: number of samples.
        :param Optional[int] random_state: random number generator seed.
        :return: samples.
        :rtype: np.ndarray
        """

        n_features = len(self.distributions)
        n_classes = len(self.priors)

        samples = np.zeros((n_samples, n_features, n_classes))
        for feature in range(len(self.distributions)):
            samples[:, feature, :] = self.distributions[feature].sample(  # type: ignore
                n_samples=n_samples, random_state=random_state
            )

        return samples

class GaussianNaiveBayes(ExtendedNaiveBayes):
    """
    Naive Bayes model with normal distributed features.
    """

    def __init__(self, n_features: int) -> None:
        """
        Init model with {n_features} normal distributed features.

        :param int n_features: number of features.
        """

        super().__init__(distributions=[Normal() for _ in range(n_features)])

class BernoulliNaiveBayes(ExtendedNaiveBayes):
    """
    Naive Bayes model with bernoulli distributed features.
    """

    def __init__(self, n_features: int) -> None:
        """
        Init model with {n_features} bernoulli distributed features.

        :param int n_features: number of features.
        """

        super().__init__(distributions=[Bernoulli() for _ in range(n_features)])

class CategoricalNaiveBayes(ExtendedNaiveBayes):
    """
    Naive Bayes model with categorical distributed features.
    """

    def __init__(self, n_features: int, n_categories: List[int]) -> None:
        """
        Init model with {n_features} categorical distributed features.

        :param int n_features: number of features.
        :param List[int] n_categories: number of categories for each feature.
        """

        assert (
            len(n_categories) == n_features
        ), "length of n_categories should be equal n_features."

        super().__init__(
            distributions=[Categorical(n_categories[i]) for i in range(n_features)]
        )

# TODO: add MultinomialNaiveBayes

max_forks_repo_name: dayyass/extended-naive-bayes

reason: import numpy,from scipy


text:
from unittest import mock

import numpy as np
import pytest
from requests.exceptions import HTTPError
import responses

from tests.conftest import SAMPLE_DATA_DIR
from wildebeest.load_funcs.image import load_image_from_disk, load_image_from_url

SAMPLE_DATA_BASE_URL = (
    'https://github.com/ShopRunner/wildebeest/raw/main/tests/sample_data'
)

@responses.activate
def test_load_error_406():
    filename = 'wildebeest_gray.jpg'
    url = f'{SAMPLE_DATA_BASE_URL}/{filename}'

    responses.add(responses.GET, url, status=406)

    with pytest.raises(HTTPError):
        with mock.patch('requests.Response'):
            result = load_image_from_url(inpath=url)

        assert result.status_code == 406

@pytest.fixture
def wildebeest_gray_local():
    filename = 'wildebeest_gray.jpg'
    path = SAMPLE_DATA_DIR / filename
    return load_image_from_disk(path)

@pytest.fixture
def wildebeest_gray_remote():
    filename = 'wildebeest_gray.jpg'
    url = f'{SAMPLE_DATA_BASE_URL}/{filename}'
    return load_image_from_url(url)

@pytest.fixture
def wildebeest_rgb_local():
    filename = 'wildebeest_rgb.jpg'
    path = SAMPLE_DATA_DIR / filename
    return load_image_from_disk(path)

@pytest.fixture
def wildebeest_rgb_remote():
    filename = 'wildebeest_rgb.jpg'
    url = f'{SAMPLE_DATA_BASE_URL}/{filename}'
    return load_image_from_url(url)

@pytest.fixture
def wildebeest_rgba_local():
    filename = 'wildebeest_rgba.png'
    path = SAMPLE_DATA_DIR / filename
    return load_image_from_disk(path)

@pytest.fixture
def wildebeest_rgba_remote():
    filename = 'wildebeest_rgba.png'
    url = f'{SAMPLE_DATA_BASE_URL}/{filename}'
    return load_image_from_url(url)

def test_load_image_from_disk_grayscale_shape(wildebeest_gray_local):
    image = wildebeest_gray_local

    expected_shape = (34, 25)
    actual_shape = image.shape

    assert actual_shape == expected_shape

def test_load_image_from_disk_rgb_shape(wildebeest_rgb_local):
    image = wildebeest_rgb_local

    expected_shape = (32, 32, 3)
    actual_shape = image.shape

    assert actual_shape == expected_shape

def test_load_image_from_disk_rgba_shape(wildebeest_rgba_local):
    image = wildebeest_rgba_local

    expected_shape = (32, 32, 4)
    actual_shape = image.shape

    assert actual_shape == expected_shape

def test_load_image_from_disk_red():
    filename = 'red.png'
    path = SAMPLE_DATA_DIR / filename

    actual_image = load_image_from_disk(path)[:, :, :3]
    expected_image = np.zeros((32, 32, 3), dtype='uint8')
    expected_image[:, :, 0] = 255

    np.testing.assert_equal(actual_image, expected_image)

def test_load_image_from_disk_green():
    filename = 'green.png'
    path = SAMPLE_DATA_DIR / filename

    actual_image = load_image_from_disk(path)[:, :, :3]
    expected_image = np.zeros((32, 32, 3), dtype='uint8')
    expected_image[:, :, 1] = 255

    np.testing.assert_equal(actual_image, expected_image)

def test_load_image_from_disk_blue():
    filename = 'blue.png'
    path = SAMPLE_DATA_DIR / filename

    actual_image = load_image_from_disk(path)[:, :, :3]
    expected_image = np.zeros((32, 32, 3), dtype='uint8')
    expected_image[:, :, 2] = 255

    np.testing.assert_equal(actual_image, expected_image)

def test_load_image_from_url_gray(wildebeest_gray_local, wildebeest_gray_remote):
    np.testing.assert_equal(wildebeest_gray_remote, wildebeest_gray_local)

def test_load_image_from_url_rgb(wildebeest_rgb_local, wildebeest_rgb_remote):
    np.testing.assert_equal(wildebeest_rgb_local, wildebeest_rgb_remote)

def test_load_image_from_url_rgba(wildebeest_rgba_local, wildebeest_rgba_remote):
    np.testing.assert_equal(wildebeest_rgba_local, wildebeest_rgba_remote)

max_stars_repo_name: gsganden/wildebeest

max_issues_repo_name: gsganden/wildebeest

max_forks_repo_name: gsganden/wildebeest


text:
"""
Copyright (c) 2020 CRISP

Generates interpolated dictionary

:author: Andrew H. Song
"""

import numpy as np
import pickle
import scipy

def generate_interpolated_Dictionary(d, numOfsubgrids, normalize=1, kind='cubic'):
	"""
	Generates interpolated dictionary given the original dictionary and number of subgrids
	(TODO): Incorporate with generate_sinc_Dictionary

	Inputs
	======
	numOfsubgrids: integer
		Specifies number of subgrids. For example if 10, divides the original sampling grid into 10 equal partitions
	kind: 'cubic' (default), 'linear', or 'sinc'
		Interpolator function

	Outputs
	=======
	d_interpolated: 2-D array
		Interpolated dictionary
	interpolator: Dictionary
		Dictionary of interpolator functions

	"""

	interpolator = {}
	if numOfsubgrids<=1:
		return d, interpolator

	print("Interpolating with 1/{} sub-grid".format(numOfsubgrids))

	interval = 1/numOfsubgrids
	delay_arr = np.arange(interval, 1, interval)

	numOfsamples, numOfelements = d.shape
	assert((0 not in delay_arr) and (1 not in delay_arr)), "Only non-integer delays are allowed"
	assert np.mod(numOfsamples,2)==1, "The filter length must be odd."

	numOfdelays = len(delay_arr)

	# Extra element is for the original template
	d_interpolated = np.zeros((numOfsamples, numOfelements*(numOfdelays+1)))
	d_interpolated[:, np.arange(numOfelements)*(numOfdelays + 1)] = d

	# The first interpolator should be a shifted delta function (to produce the original element)
	delta_fn = np.zeros(numOfsamples)
	delta_fn[int(numOfsamples/2)] = 1
	interpolator[0] = delta_fn

	for didx, delay in enumerate(delay_arr,1):
		if kind == 'cubic':
			x_interp = np.linspace(-2,2,5,endpoint=True) - delay
			f_interp = []
			for idx in x_interp:
				if abs(idx)>=2:
					f_interp.append(0)
				elif 1<=abs(idx) and abs(idx)<2:
					f_new = -0.5*np.power(abs(idx),3) + 2.5*np.power(abs(idx),2) - 4*abs(idx)+2
					f_interp.append(f_new)
				else:
					f_new = 1.5*np.power(abs(idx),3) - 2.5*np.power(abs(idx),2) + 1
					f_interp.append(f_new)
		elif kind == 'linear':
			x_interp = np.linspace(-1,1,3,endpoint=True) - delay
			f_interp = []
			for idx in x_interp:
				if abs(idx)>=1:
					f_interp.append(0)
				else:
					f_new = 1 - abs(idx)
					f_interp.append(f_new)
		elif kind == 'sinc':
			if np.mod(numOfsamples, 2)==0:
				x = np.arange(numOfsamples) - int(numOfsamples/2)+1
			else:
				x = np.arange(numOfsamples) - int(numOfsamples/2)

			f_interp = np.sinc(x-delay)
		else:
			raise NotImplementedError("This interpolator is not implemented!")

		interpolator[didx] = f_interp

		for fidx in np.arange(numOfelements):
			elem = d[:,fidx]
			d_interpolated[:, fidx*(numOfdelays+1)+didx] = scipy.signal.convolve(elem, f_interp, mode='same')

	if normalize:
		d_interpolated = d_interpolated/np.linalg.norm(d_interpolated, axis=0)

	return d_interpolated, interpolator

def generate_interpolated_Dictionary_2D(d, x_delay_arr, y_delay_arr, normalize=1, kind='cubic'):
	"""
	Interpolation of the dictionary for 2D.
	For now, doesn't return the interpolators
	x_delay_arr and y_delay_arr need to have 0 in them.
	"""

	from scipy.interpolate import griddata

	assert np.array_equal(x_delay_arr, y_delay_arr), "Delays in both directions should be same"

	if len(x_delay_arr)==0:
		return d

	numOfsamples, _, numOfelements = d.shape
	assert np.mod(numOfsamples,2)==1, "The filter length must be odd."
	numOfdelays = len(x_delay_arr)

	# Extra element is for the original template
	d_interpolated = np.zeros((numOfsamples, numOfsamples, numOfelements*numOfdelays**2))

	grid_original = [(i,j) for i in np.arange(numOfsamples) for j in np.arange(numOfsamples)]

	for x_delay_idx, x_delay in enumerate(x_delay_arr):
		for y_delay_idx, y_delay in enumerate(y_delay_arr):

			index = x_delay_idx * numOfdelays + y_delay_idx
			if x_delay_idx ==0:
				x_grid = np.arange(numOfsamples)
			else:
				x_grid = (1-x_delay) + np.arange(numOfsamples-1)
				x_grid = np.insert(x_grid,0,0)

			if y_delay_idx == 0:
				y_grid = np.arange(numOfsamples)
			else:
				y_grid = (1-y_delay) + np.arange(numOfsamples-1)
				y_grid = np.insert(y_grid,0,0)

			grid_new = [(i,j) for i in x_grid for j in y_grid]

			for fidx in range(numOfelements):
				itp = griddata(grid_original, d[..., fidx].reshape(1,-1).flatten(), grid_new, method='cubic')
				d_interpolated[..., fidx * numOfdelays**2 + index] = itp.reshape(numOfsamples, numOfsamples)

	if normalize:
		for idx in range(numOfelements*(numOfdelays+1)):
			d_interpolated[..., idx] = d_interpolated[..., idx]/np.linalg.norm(d_interpolated[..., idx])

	return d_interpolated

max_stars_repo_name: DominicTanzillo/srcdl

max_issues_repo_name: DominicTanzillo/srcdl

max_forks_repo_name: DominicTanzillo/srcdl

reason: import numpy,import scipy,from scipy


text:
from collections import Iterable
from numbers import Integral

import numba
import numpy as np

from ..compatibility import range, zip_longest
from ..slicing import normalize_index
from ..butils import _zero_of_dtype
from ..coo.indexing import _prune_indices, _mask, _compute_mask

def getitem(x, index):
    """
    This function implements the indexing functionality for BCOO.

    The overall algorithm has three steps:

    1. Normalize the index to canonical form. Function: normalize_index
    2. Get the mask, which is a list of integers corresponding to
       the indices in coords/data for the output data. Function: _mask
    3. Transform the coordinates to what they will be in the output.

    Parameters
    ----------
    x : BCOO
        The array to apply the indexing operation on.
    index : {tuple, str}
        The index into the array.
    """
    from .bcore import BCOO

    # If string, this is an index into an np.void Custom dtype.
    if isinstance(index, str):
        return getblock(x, index)

    # Otherwise, convert into a tuple.
    if not isinstance(index, tuple):
        index = (index,)

    # Check if the last index is an ellipsis.
    last_ellipsis = len(index) > 0 and index[-1] is Ellipsis

    # Normalize the index into canonical form.
    index = normalize_index(index, x.shape)

    # zip_longest so things like x[..., None] are picked up.
    if len(index) != 0 and all(ind == slice(0, dim, 1)
                               for ind, dim in zip_longest(index, x.outer_shape)):
        return x

    if x._offset is None:
        x_offset = (0,) * x.ndim
    else:
        x_offset = x._offset
    if x._last_block_shape is None:
        x_last_block_shape = x.block_shape
    else:
        x_last_block_shape = x._last_block_shape

    # Get the mask
    mask = _outer_mask(x.coords, index, x.shape, x.block_shape, x_offset)

    # Get the length of the mask
    if isinstance(mask, slice):
        n = len(range(mask.start, mask.stop, mask.step))
    else:
        n = len(mask)

    coords = []
    shape = []
    block_mask = []
    block_shape = []
    offset = []
    last_block_shape = []
    i = 0
    for ind in index:
        if isinstance(ind, Integral):
            block_mask.append((ind + x_offset[i]) % x.block_shape[i])
            i += 1
        # Add to the shape and transform the coords in the case of a slice.
        elif isinstance(ind, slice):
            if ind.step < 0:
                raise NotImplementedError('Negative step is not supported')
            elif ind.step > 1:
                raise NotImplementedError('Non-continous step is not supported')
            shape.append(len(range(ind.start, ind.stop, ind.step)))
            block_mask.append(slice(None,None))
            block_shape.append(x.block_shape[i])
            offset.append((ind.start + x_offset[i]) % x.block_shape[i])
            last_block_shape.append((ind.stop - 1 + x_offset[i]) % x.block_shape[i] + 1)
            start = (x_offset[i] + ind.start) // x.block_shape[i]
            coords.append((x.coords[i, mask].astype(int) - start))
            i += 1
        elif isinstance(ind, Iterable):
            raise NotImplementedError('Advanced indexing is not yet supported.')
        # Add a dimension for None.
        elif ind is None:
            coords.append(np.zeros(n, dtype=x.coords.dtype))
            shape.append(1)
            block_mask.append(np.newaxis)
            block_shape.append(1)
            offset.append(0)
            last_block_shape.append(1)

    # Join all the transformed coords.
    if coords:
        coords = np.stack(coords, axis=0)
    else:
        # If index result is a scalar, return a 0-d BCOO or
        # a block depending on whether the last index is an ellipsis.
        if last_ellipsis:
            coords = np.empty((0, n), dtype=np.uint8)
        else:  # scalar
            if n != 0:
                return x.data[(mask,)+tuple(block_mask)][0]
            else:
                return _zero_of_dtype(x.dtype)[()]

    block_shape = tuple(block_shape)
    data = x.data[(mask,)+tuple(block_mask)]

    b = BCOO(coords, data, shape=shape, block_shape=block_shape,
             has_duplicates=False, sorted=True)

    if any(i != 0 for i in offset):
        b._offset = tuple(offset)
        raise NotImplementedError("Offset is not supported.")

    if tuple(last_block_shape) != b.block_shape:
        b._last_block_shape = tuple(last_block_shape)
        raise NotImplementedError("Boundary is not supported.")

    return b

def setitem(x, index, value):
    raise NotImplementedError

def getblock(x, index):
    """
    This function implements the functionality to index BCOO blocks.

    Parameters
    ----------
    x : BCOO
        The array to apply the indexing operation on.
    index : {tuple, str}
        The index into the array.
    """
    from .bcore import BCOO

    # If string, this is an index into an np.void Custom dtype.
    if isinstance(index, str):
        data = x.data[index]
        dtype_shape = x.data.dtype[index].shape
        coords = x.coords
        shape = x.shape
        block_shape = x.block_shape

        if dtype_shape:
            loc0 = np.zeros(coords.shape[1], dtype=coords.dtype)
            coords = np.vstack([coords, (loc0,)*len(dtype_shape)])
            shape = shape + dtype_shape
            block_shape = block_shape + dtype_shape

        b = BCOO(coords, data,
                 shape=shape, block_shape=block_shape,
                 has_duplicates=False,
                 sorted=True)

        if dtype_shape:
            if x._offset is not None:
                b._offset = x._offset + (0,)*len(dtype_shape)
            if x._last_block_shape is not None:
                b._last_block_shape = x._last_block_shape + dtype_shape

        return b

    # Otherwise, convert into a tuple.
    if not isinstance(index, tuple):
        index = (index,)

    # Check if the last index is an ellipsis.
    last_ellipsis = len(index) > 0 and index[-1] is Ellipsis

    # Normalize the index into canonical form.
    index = normalize_index(index, x.outer_shape)

    # zip_longest so things like x[..., None] are picked up.
    if len(index) != 0 and all(ind == slice(0, dim, 1)
                               for ind, dim in zip_longest(index, x.outer_shape)):
        return x

    # Get the mask
    mask = _mask(x.coords, index, x.outer_shape)

    # Get the length of the mask
    if isinstance(mask, slice):
        n = len(range(mask.start, mask.stop, mask.step))
    else:
        n = len(mask)

    coords = []
    outer_shape = []
    block_shape = []
    i = 0
    for ind in index:
        if isinstance(ind, Integral):
            outer_shape.append(1)
            block_shape.append(x.block_shape[i])
            coords.append(np.zeros(n, dtype=x.coords.dtype))
            i += 1
        # Add to the shape and transform the coords in the case of a slice.
        elif isinstance(ind, slice):
            outer_shape.append(len(range(ind.start, ind.stop, ind.step)))
            block_shape.append(x.block_shape[i])
            coords.append((x.coords[i, mask].astype(int) - ind.start) // ind.step)
            i += 1
        elif isinstance(ind, Iterable):
            raise NotImplementedError('Advanced indexing is not yet supported.')
        # Add a dimension for None.
        elif ind is None:
            coords.append(np.zeros(n, dtype=x.coords.dtype))
            outer_shape.append(1)
            block_shape.append(1)

    # Join all the transformed coords.
    if coords:
        coords = np.stack(coords, axis=0)
    else:
        # If index result is a scalar, return a 0-d BCOO or
        # a block depending on whether the last index is an ellipsis.
        if last_ellipsis:
            #raise RuntimeError("bsparse does not support 0-D BCOO")
            coords = np.empty((0, n), dtype=np.uint8)
        else:  # scalar
            if n != 0:
                return x.data[mask][0]
            else:
                return np.zeros((x.block_shape), dtype=x.dtype)

    block_shape = tuple(block_shape)
    shape = tuple([x*y for x, y in zip(outer_shape, block_shape)])
    data = x.data[mask].reshape((-1,) + block_shape)

    return BCOO(coords, data, shape=shape, block_shape=block_shape,
                has_duplicates=False,
                sorted=True)

def setblock(x, index, value):
    raise NotImplementedError

def _outer_mask(coords, indices, shape, block_shape, offset):
    indices = _prune_indices(indices, shape)

    ind_ar = np.empty((len(indices), 3), dtype=np.intp)

    for i, idx in enumerate(indices):
        if isinstance(idx, slice):
            if idx.step != 1:
                raise NotImplementedError('Non-continous step is not supported')
            start = (offset[i] + idx.start) // block_shape[i]
            stop = (offset[i] + idx.stop + block_shape[i]-1) // block_shape[i]
            ind_ar[i] = [start, stop, 1]
        else:  # idx is an integer
            start = (offset[i] + idx) // block_shape[i]
            ind_ar[i] = [start, start + 1, 1]

    mask, is_slice = _compute_mask(coords, ind_ar)

    if is_slice:
        return slice(mask[0], mask[1], 1)
    else:
        return mask

max_stars_repo_name: zhcui/sparse

max_issues_repo_name: zhcui/sparse

max_forks_repo_name: zhcui/sparse

reason: import numpy,import numba


text:
# -------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
# -------------------------------------------------------------------------------------------
import numpy as np
import pytest
from staticchar import preprocessing as pp

def test_background_estimate():
    data = np.concatenate([np.arange(10, 20), np.arange(5, 10)])
    assert pp.background_estimate(data, pp.BackgroundChoices.Minimum) == 5
    assert pp.background_estimate(data, pp.BackgroundChoices.FirstMeasurement) == 10
    with pytest.raises(ValueError):
        pp.background_estimate(data, None)  # type: ignore # auto

max_stars_repo_name: BrunoKM/station-b-libraries

max_issues_repo_name: BrunoKM/station-b-libraries

max_forks_repo_name: BrunoKM/station-b-libraries


text:
# -*- coding: utf-8 -*-

from __future__ import division

import numpy as np
from scipy import optimize

from pygsp import utils
from . import Filter  # prevent circular import in Python < 3.5

class Abspline(Filter):
    r"""Design an A B cubic spline wavelet filter bank.

    Parameters
    ----------
    G : graph
    Nf : int
        Number of filters from 0 to lmax (default = 6)
    lpfactor : float
        Low-pass factor lmin=lmax/lpfactor will be used to determine scales,
        the scaling function will be created to fill the lowpass gap.
        (default = 20)
    scales : ndarray
        Vector of scales to be used.
        By default, initialized with :func:`pygsp.utils.compute_log_scales`.

    Examples
    --------

    Filter bank's representation in Fourier and time (ring graph) domains.

    >>> import matplotlib.pyplot as plt
    >>> G = graphs.Ring(N=20)
    >>> G.estimate_lmax()
    >>> G.set_coordinates('line1D')
    >>> g = filters.Abspline(G)
    >>> s = g.localize(G.N // 2)
    >>> fig, axes = plt.subplots(1, 2)
    >>> _ = g.plot(ax=axes[0])
    >>> _ = G.plot(s, ax=axes[1])

    """

    def __init__(self, G, Nf=6, lpfactor=20, scales=None):

        def kernel_abspline3(x, alpha, beta, t1, t2):
            M = np.array([[1, t1, t1**2, t1**3],
                          [1, t2, t2**2, t2**3],
                          [0, 1, 2*t1, 3*t1**2],
                          [0, 1, 2*t2, 3*t2**2]])
            v = np.array([1, 1, t1**(-alpha) * alpha * t1**(alpha - 1),
                          -beta*t2**(- beta - 1) * t2**beta])
            a = np.linalg.solve(M, v)

            r1 = x <= t1
            r2 = (x >= t1)*(x < t2)
            r3 = (x >= t2)

            if isinstance(x, np.float64):

                if r1:
                    r = x[r1]**alpha * t1**(-alpha)
                if r2:
                    r = a[0] + a[1] * x + a[2] * x**2 + a[3] * x**3
                if r3:
                    r = x[r3]**(-beta) * t2**beta

            else:
                r = np.zeros(x.shape)

                x2 = x[r2]

                r[r1] = x[r1]**alpha * t1**(-alpha)
                r[r2] = a[0] + a[1] * x2 + a[2] * x2**2 + a[3] * x2**3
                r[r3] = x[r3]**(-beta) * t2 ** beta

            return r

        self.lpfactor = lpfactor

        lmin = G.lmax / lpfactor

        if scales is None:
            scales = utils.compute_log_scales(lmin, G.lmax, Nf - 1)
        self.scales = scales

        gb = lambda x: kernel_abspline3(x, 2, 2, 1, 2)
        gl = lambda x: np.exp(-np.power(x, 4))

        lminfac = .4 * lmin

        g = [lambda x: 1.2 * np.exp(-1) * gl(x / lminfac)]
        for i in range(0, Nf - 1):
            g.append(lambda x, i=i: gb(self.scales[i] * x))

        f = lambda x: -gb(x)
        xstar = optimize.minimize_scalar(f, bounds=(1, 2),
                                         method='bounded')
        gamma_l = -f(xstar.x)
        lminfac = .6 * lmin
        g[0] = lambda x: gamma_l * gl(x / lminfac)

        super(Abspline, self).__init__(G, g)

    def _get_extra_repr(self):
        return dict(lpfactor='{:.2f}'.format(self.lpfactor))

max_stars_repo_name: jafluri/pygsp

max_issues_repo_name: jafluri/pygsp

max_forks_repo_name: jafluri/pygsp


text:
from z3 import Implies

import numpy as np
from z3 import Implies

from quavl.lib.expressions.qbit import Qbits
from quavl.lib.models.circuit import Circuit, Method
from quavl.lib.operations.gates import Rx, Rz, V, CNOT, V_dag

def repair_toffoli():
    # Initialize circuit
    a, b, c = Qbits(['a', 'b', 'c'])
    n = 3

    circuit = Circuit([a, b, c],
                      [
                          V(a).controlled_by(c),
                          V(a).controlled_by(b),
                          # CNOT(b, c),
                          V_dag(a).controlled_by(b),
                          CNOT(b, c)
                      ])

    initial_values = [{(1, 0), (0, 1)} for _ in range(n)]

    circuit.initialize_qbits(initial_values)

    # Build specification
    final_qbits = circuit.get_final_qbits()

    circuit.solver.add(Implies(b.alpha.r == 1 and c.alpha.r == 1,
                               final_qbits[0].beta.r == a.alpha.r))

    # Prove and repair
    circuit.prove(method=Method.qbit_sequence_model,
                  dump_smt_encoding=True,
                  dump_solver_output=True,
                  synthesize_repair=True)

def prove_repaired_toffoli():
    rep_theta_0 = [-0.10000000000000001, -0.099999999999999992]
    rep_phi_0 = [0.099999999999999992, 0.10000000000000001]

    rep_theta_1 = [-0.10000000000000001, -0.099999999999999992]
    rep_phi_1 = [0.099999999999999992, 0.10000000000000001]

    rep_theta_2 = [-0.10000000000000001, -0.099999999999999992]
    rep_phi_2 = [0.099999999999999992, 0.10000000000000001]

    # Initialize circuit
    a, b, c = Qbits(['a', 'b', 'c'])
    n = 3

    circuit = Circuit([a, b, c],
                      [
                          V(a).controlled_by(c),
                          V(a).controlled_by(b),
                          # CNOT(b, c),
                          V_dag(a).controlled_by(b),
                          CNOT(b, c),
                          Rx(a, np.mean(rep_theta_0)),
                          Rz(a, np.mean(rep_phi_0)),
                          Rx(b, np.mean(rep_theta_1)),
                          Rz(b, np.mean(rep_phi_1)),
                          Rx(c, np.mean(rep_theta_2)),
                          Rz(c, np.mean(rep_phi_2))
                      ])

    initial_values = [{(1, 0), (0, 1)} for _ in range(n)]

    circuit.initialize_qbits(initial_values)

    # Build specification
    final_qbits = circuit.get_final_qbits()

    circuit.solver.add(Implies(b.alpha.r == 1 and c.alpha.r == 1,
                               final_qbits[0].beta.r == a.alpha.r))

    # Prove and repair
    circuit.prove(method=Method.qbit_sequence_model,
                  dump_smt_encoding=True,
                  dump_solver_output=True,
                  synthesize_repair=True)

if __name__ == "__main__":
    repair_toffoli()
    prove_repaired_toffoli()

max_stars_repo_name: fabianbauermarquart/quavl

max_issues_repo_name: fabianbauermarquart/quavl

max_forks_repo_name: fabianbauermarquart/quavl