text:
# Authors: Alexandre Gramfort <alexandre.gramfort@telecom-paristech.fr>
#          Matti Hamalainen <msh@nmr.mgh.harvard.edu>
#          Denis Engemann <denis.engemann@gmail.com>
#          Andrew Dykstra <andrew.r.dykstra@gmail.com>
#
# License: BSD (3-clause)

from copy import deepcopy
import numpy as np
import warnings

from .baseline import rescale
from .channels import ContainsMixin, PickDropChannelsMixin
from .filter import resample, detrend
from .fixes import in1d
from .utils import (_check_pandas_installed, check_fname, logger, verbose,
                    deprecated, object_hash)
from .viz import plot_evoked, plot_evoked_topomap, _mutable_defaults
from .viz import plot_evoked_field
from .viz import plot_evoked_image
from .externals.six import string_types

from .io.constants import FIFF
from .io.open import fiff_open
from .io.tag import read_tag
from .io.tree import dir_tree_find
from .io.pick import channel_type, pick_types
from .io.meas_info import read_meas_info, write_meas_info
from .io.proj import ProjMixin
from .io.write import (start_file, start_block, end_file, end_block,
                       write_int, write_string, write_float_matrix,
                       write_id)

aspect_dict = {'average': FIFF.FIFFV_ASPECT_AVERAGE,
               'standard_error': FIFF.FIFFV_ASPECT_STD_ERR}
aspect_rev = {str(FIFF.FIFFV_ASPECT_AVERAGE): 'average',
              str(FIFF.FIFFV_ASPECT_STD_ERR): 'standard_error'}

class Evoked(ProjMixin, ContainsMixin, PickDropChannelsMixin):
    """Evoked data

    Parameters
    ----------
    fname : string
        Name of evoked/average FIF file to load.
        If None no data is loaded.
    condition : int, or str
        Dataset ID number (int) or comment/name (str). Optional if there is
        only one data set in file.
    baseline : tuple or list of length 2, or None
        The time interval to apply rescaling / baseline correction.
        If None do not apply it. If baseline is (a, b)
        the interval is between "a (s)" and "b (s)".
        If a is None the beginning of the data is used
        and if b is None then b is set to the end of the interval.
        If baseline is equal ot (None, None) all the time
        interval is used. If None, no correction is applied.
    proj : bool, optional
        Apply SSP projection vectors
    kind : str
        Either 'average' or 'standard_error'. The type of data to read.
        Only used if 'condition' is a str.
    verbose : bool, str, int, or None
        If not None, override default verbose level (see mne.verbose).

    Attributes
    ----------
    info : dict
        Measurement info.
    ch_names : list of string
        List of channels' names.
    nave : int
        Number of averaged epochs.
    kind : str
        Type of data, either average or standard_error.
    first : int
        First time sample.
    last : int
        Last time sample.
    comment : string
        Comment on dataset. Can be the condition.
    times : array
        Array of time instants in seconds.
    data : array of shape (n_channels, n_times)
        Evoked response.
    verbose : bool, str, int, or None.
        See above.
    """
    @verbose
    def __init__(self, fname, condition=None, baseline=None, proj=True,
                 kind='average', verbose=None, setno=None):

        if fname is None:
            raise ValueError('No evoked filename specified')

        if condition is None and setno is not None:
            condition = setno
            msg = ("'setno' will be deprecated in 0.9. Use 'condition' "
                   "instead.")
            warnings.warn(msg, DeprecationWarning)

        self.verbose = verbose
        logger.info('Reading %s ...' % fname)
        fid, tree, _ = fiff_open(fname)
        if not isinstance(proj, bool):
            raise ValueError(r"'proj' must be 'True' or 'False'")

        #   Read the measurement info
        info, meas = read_meas_info(fid, tree)
        info['filename'] = fname

        #   Locate the data of interest
        processed = dir_tree_find(meas, FIFF.FIFFB_PROCESSED_DATA)
        if len(processed) == 0:
            fid.close()
            raise ValueError('Could not find processed data')

        evoked_node = dir_tree_find(meas, FIFF.FIFFB_EVOKED)
        if len(evoked_node) == 0:
            fid.close()
            raise ValueError('Could not find evoked data')

        # find string-based entry
        if isinstance(condition, string_types):
            if not kind in aspect_dict.keys():
                fid.close()
                raise ValueError('kind must be "average" or '
                                 '"standard_error"')

            comments, aspect_kinds, t = _get_entries(fid, evoked_node)
            goods = np.logical_and(in1d(comments, [condition]),
                                   in1d(aspect_kinds, [aspect_dict[kind]]))
            found_cond = np.where(goods)[0]
            if len(found_cond) != 1:
                fid.close()
                raise ValueError('condition "%s" (%s) not found, out of found '
                                 'datasets:\n  %s' % (condition, kind, t))
            condition = found_cond[0]

        if condition >= len(evoked_node) or condition < 0:
            fid.close()
            raise ValueError('Data set selector out of range')

        my_evoked = evoked_node[condition]

        # Identify the aspects
        aspects = dir_tree_find(my_evoked, FIFF.FIFFB_ASPECT)
        if len(aspects) > 1:
            logger.info('Multiple aspects found. Taking first one.')
        my_aspect = aspects[0]

        # Now find the data in the evoked block
        nchan = 0
        sfreq = -1
        chs = []
        comment = None
        for k in range(my_evoked['nent']):
            my_kind = my_evoked['directory'][k].kind
            pos = my_evoked['directory'][k].pos
            if my_kind == FIFF.FIFF_COMMENT:
                tag = read_tag(fid, pos)
                comment = tag.data
            elif my_kind == FIFF.FIFF_FIRST_SAMPLE:
                tag = read_tag(fid, pos)
                first = int(tag.data)
            elif my_kind == FIFF.FIFF_LAST_SAMPLE:
                tag = read_tag(fid, pos)
                last = int(tag.data)
            elif my_kind == FIFF.FIFF_NCHAN:
                tag = read_tag(fid, pos)
                nchan = int(tag.data)
            elif my_kind == FIFF.FIFF_SFREQ:
                tag = read_tag(fid, pos)
                sfreq = float(tag.data)
            elif my_kind == FIFF.FIFF_CH_INFO:
                tag = read_tag(fid, pos)
                chs.append(tag.data)

        if comment is None:
            comment = 'No comment'

        #   Local channel information?
        if nchan > 0:
            if chs is None:
                fid.close()
                raise ValueError('Local channel information was not found '
                                 'when it was expected.')

            if len(chs) != nchan:
                fid.close()
                raise ValueError('Number of channels and number of '
                                 'channel definitions are different')

            info['chs'] = chs
            info['nchan'] = nchan
            logger.info('    Found channel information in evoked data. '
                        'nchan = %d' % nchan)
            if sfreq > 0:
                info['sfreq'] = sfreq

        nsamp = last - first + 1
        logger.info('    Found the data of interest:')
        logger.info('        t = %10.2f ... %10.2f ms (%s)'
                    % (1000 * first / info['sfreq'],
                       1000 * last / info['sfreq'], comment))
        if info['comps'] is not None:
            logger.info('        %d CTF compensation matrices available'
                        % len(info['comps']))

        # Read the data in the aspect block
        nave = 1
        epoch = []
        for k in range(my_aspect['nent']):
            kind = my_aspect['directory'][k].kind
            pos = my_aspect['directory'][k].pos
            if kind == FIFF.FIFF_COMMENT:
                tag = read_tag(fid, pos)
                comment = tag.data
            elif kind == FIFF.FIFF_ASPECT_KIND:
                tag = read_tag(fid, pos)
                aspect_kind = int(tag.data)
            elif kind == FIFF.FIFF_NAVE:
                tag = read_tag(fid, pos)
                nave = int(tag.data)
            elif kind == FIFF.FIFF_EPOCH:
                tag = read_tag(fid, pos)
                epoch.append(tag)

        logger.info('        nave = %d - aspect type = %d'
                    % (nave, aspect_kind))

        nepoch = len(epoch)
        if nepoch != 1 and nepoch != info['nchan']:
            fid.close()
            raise ValueError('Number of epoch tags is unreasonable '
                             '(nepoch = %d nchan = %d)'
                             % (nepoch, info['nchan']))

        if nepoch == 1:
            # Only one epoch
            all_data = epoch[0].data.astype(np.float)
            # May need a transpose if the number of channels is one
            if all_data.shape[1] == 1 and info['nchan'] == 1:
                all_data = all_data.T.astype(np.float)
        else:
            # Put the old style epochs together
            all_data = np.concatenate([e.data[None, :] for e in epoch],
                                      axis=0).astype(np.float)

        if all_data.shape[1] != nsamp:
            fid.close()
            raise ValueError('Incorrect number of samples (%d instead of %d)'
                             % (all_data.shape[1], nsamp))

        # Calibrate
        cals = np.array([info['chs'][k]['cal']
                         * info['chs'][k].get('scale', 1.0)
                         for k in range(info['nchan'])])
        all_data *= cals[:, np.newaxis]

        times = np.arange(first, last + 1, dtype=np.float) / info['sfreq']
        self.info = info

        # Put the rest together all together
        self.nave = nave
        self._aspect_kind = aspect_kind
        self.kind = aspect_rev.get(str(self._aspect_kind), 'Unknown')
        self.first = first
        self.last = last
        self.comment = comment
        self.times = times
        self.data = all_data

        # bind info, proj, data to self so apply_proj can be used
        self.data = all_data
        self.proj = False
        if proj:
            self.apply_proj()
        # Run baseline correction
        self.data = rescale(self.data, times, baseline, 'mean', copy=False)

        fid.close()

    def save(self, fname):
        """Save dataset to file.

        Parameters
        ----------
        fname : string
            Name of the file where to save the data.
        """
        write_evokeds(fname, self)

    def __repr__(self):
        s = "comment : '%s'" % self.comment
        s += ", time : [%f, %f]" % (self.times[0], self.times[-1])
        s += ", n_epochs : %d" % self.nave
        s += ", n_channels x n_times : %s x %s" % self.data.shape
        return "<Evoked  |  %s>" % s

    @property
    def ch_names(self):
        return self.info['ch_names']

    def crop(self, tmin=None, tmax=None):
        """Crop data to a given time interval
        """
        times = self.times
        mask = np.ones(len(times), dtype=np.bool)
        if tmin is not None:
            mask = mask & (times >= tmin)
        if tmax is not None:
            mask = mask & (times <= tmax)
        self.times = times[mask]
        self.first = int(self.times[0] * self.info['sfreq'])
        self.last = len(self.times) + self.first - 1
        self.data = self.data[:, mask]

    def shift_time(self, tshift, relative=True):
        """Shift time scale in evoked data

        Parameters
        ----------
        tshift : float
            The amount of time shift to be applied if relative is True
            else the first time point. When relative is True, positive value
            of tshift moves the data forward while negative tshift moves it
            backward.
        relative : bool
            If true, move the time backwards or forwards by specified amount.
            Else, set the starting time point to the value of tshift.

        Notes
        -----
        Maximum accuracy of time shift is 1 / evoked.info['sfreq']
        """
        times = self.times
        sfreq = self.info['sfreq']

        offset = self.first if relative else 0

        self.first = int(tshift * sfreq) + offset
        self.last = self.first + len(times) - 1
        self.times = np.arange(self.first, self.last + 1,
                               dtype=np.float) / sfreq

    def plot(self, picks=None, exclude='bads', unit=True, show=True, ylim=None,
             proj=False, xlim='tight', hline=None, units=None, scalings=None,
             titles=None, axes=None):
        """Plot evoked data as butterfly plots

        Note: If bad channels are not excluded they are shown in red.

        Parameters
        ----------
        picks : array-like of int | None
            The indices of channels to plot. If None show all.
        exclude : list of str | 'bads'
            Channels names to exclude from being shown. If 'bads', the
            bad channels are excluded.
        unit : bool
            Scale plot with channel (SI) unit.
        show : bool
            Call pyplot.show() at the end or not.
        ylim : dict
            ylim for plots. e.g. ylim = dict(eeg=[-200e-6, 200e6])
            Valid keys are eeg, mag, grad
        xlim : 'tight' | tuple | None
            xlim for plots.
        proj : bool | 'interactive'
            If true SSP projections are applied before display. If
            'interactive', a check box for reversible selection of SSP
            projection vectors will be shown.
        hline : list of floats | None
            The values at which show an horizontal line.
        units : dict | None
            The units of the channel types used for axes lables. If None,
            defaults to `dict(eeg='uV', grad='fT/cm', mag='fT')`.
        scalings : dict | None
            The scalings of the channel types to be applied for plotting.
            If None, defaults to `dict(eeg=1e6, grad=1e13, mag=1e15)`.
        titles : dict | None
            The titles associated with the channels. If None, defaults to
            `dict(eeg='EEG', grad='Gradiometers', mag='Magnetometers')`.
        axes : instance of Axes | list | None
            The axes to plot to. If list, the list must be a list of Axes of
            the same length as the number of channel types. If instance of
            Axes, there must be only one channel type plotted.
        """
        return plot_evoked(self, picks=picks, exclude=exclude, unit=unit,
                           show=show, ylim=ylim, proj=proj, xlim=xlim,
                           hline=hline, units=units, scalings=scalings,
                           titles=titles, axes=axes)

    def plot_image(self, picks=None, exclude='bads', unit=True, show=True,
                   clim=None, proj=False, xlim='tight', units=None,
                   scalings=None, titles=None, axes=None, cmap='RdBu_r'):
        """Plot evoked data as images

        Parameters
        ----------
        picks : array-like of int | None
            The indices of channels to plot. If None show all.
        exclude : list of str | 'bads'
            Channels names to exclude from being shown. If 'bads', the
            bad channels are excluded.
        unit : bool
            Scale plot with channel (SI) unit.
        show : bool
            Call pyplot.show() at the end or not.
        clim : dict
            clim for images. e.g. clim = dict(eeg=[-200e-6, 200e6])
            Valid keys are eeg, mag, grad
        xlim : 'tight' | tuple | None
            xlim for plots.
        proj : bool | 'interactive'
            If true SSP projections are applied before display. If
            'interactive', a check box for reversible selection of SSP
            projection vectors will be shown.
        units : dict | None
            The units of the channel types used for axes lables. If None,
            defaults to `dict(eeg='uV', grad='fT/cm', mag='fT')`.
        scalings : dict | None
            The scalings of the channel types to be applied for plotting.
            If None, defaults to `dict(eeg=1e6, grad=1e13, mag=1e15)`.
        titles : dict | None
            The titles associated with the channels. If None, defaults to
            `dict(eeg='EEG', grad='Gradiometers', mag='Magnetometers')`.
        axes : instance of Axes | list | None
            The axes to plot to. If list, the list must be a list of Axes of
            the same length as the number of channel types. If instance of
            Axes, there must be only one channel type plotted.
        cmap : matplotlib colormap
            Colormap.
        """
        return plot_evoked_image(self, picks=picks, exclude=exclude, unit=unit,
                                 show=show, clim=clim, proj=proj, xlim=xlim,
                                 units=units, scalings=scalings,
                                 titles=titles, axes=axes, cmap=cmap)

    def plot_topomap(self, times=None, ch_type='mag', layout=None, vmin=None,
                     vmax=None, cmap='RdBu_r', sensors='k,', colorbar=True,
                     scale=None, scale_time=1e3, unit=None, res=64, size=1,
                     format="%3.1f", time_format='%01d ms', proj=False,
                     show=True, show_names=False, title=None, mask=None,
                     mask_params=None, outlines='head', contours=6,
                     image_interp='bilinear'):
        """Plot topographic maps of specific time points

        Parameters
        ----------
        times : float | array of floats | None.
            The time point(s) to plot. If None, 10 topographies will be shown
            will a regular time spacing between the first and last time
            instant.
        ch_type : 'mag' | 'grad' | 'planar1' | 'planar2' | 'eeg'
            The channel type to plot. For 'grad', the gradiometers are collec-
            ted in pairs and the RMS for each pair is plotted.
        layout : None | Layout
            Layout instance specifying sensor positions (does not need to
            be specified for Neuromag data). If possible, the correct
            layout file is inferred from the data; if no appropriate layout
            file was found, the layout is automatically generated from the
            sensor locations.
        vmin : float | callable
            The value specfying the lower bound of the color range.
            If None, and vmax is None, -vmax is used. Else np.min(data).
            If callable, the output equals vmin(data).
        vmax : float | callable
            The value specfying the upper bound of the color range.
            If None, the maximum absolute value is used. If vmin is None,
            but vmax is not, defaults to np.min(data).
            If callable, the output equals vmax(data).
        cmap : matplotlib colormap
            Colormap. Defaults to 'RdBu_r'
        sensors : bool | str
            Add markers for sensor locations to the plot. Accepts matplotlib
            plot format string (e.g., 'r+' for red plusses).
        colorbar : bool
            Plot a colorbar.
        scale : float | None
            Scale the data for plotting. If None, defaults to 1e6 for eeg, 1e13
            for grad and 1e15 for mag.
        scale_time : float | None
            Scale the time labels. Defaults to 1e3 (ms).
        units : str | None
            The units of the channel types used for colorbar lables. If
            scale == None the unit is automatically determined.
        res : int
            The resolution of the topomap image (n pixels along each side).
        size : scalar
            Side length of the topomaps in inches (only applies when plotting
            multiple topomaps at a time).
        format : str
            String format for colorbar values.
        time_format : str
            String format for topomap values. Defaults to "%01d ms"
        proj : bool | 'interactive'
            If true SSP projections are applied before display. If
            'interactive', a check box for reversible selection of SSP
            projection vectors will be shown.
        show : bool
            Call pyplot.show() at the end.
        show_names : bool | callable
            If True, show channel names on top of the map. If a callable is
            passed, channel names will be formatted using the callable; e.g.,
            to delete the prefix 'MEG ' from all channel names, pass the
            function
            lambda x: x.replace('MEG ', ''). If `mask` is not None, only
            significant sensors will be shown.
        title : str | None
            Title. If None (default), no title is displayed.
        mask : ndarray of bool, shape (n_channels, n_times) | None
            The channels to be marked as significant at a given time point.
            Indicies set to `True` will be considered. Defaults to None.
        mask_params : dict | None
            Additional plotting parameters for plotting significant sensors.
            Default (None) equals:
            dict(marker='o', markerfacecolor='w', markeredgecolor='k',
                 linewidth=0, markersize=4)
        outlines : 'head' | dict | None
            The outlines to be drawn. If 'head', a head scheme will be drawn.
            If dict, each key refers to a tuple of x and y positions. The
            values in 'mask_pos' will serve as image mask. If None,
            nothing will be drawn. Defaults to 'head'.
        image_interp : str
            The image interpolation to be used. All matplotlib options are
            accepted.

        """
        return plot_evoked_topomap(self, times=times, ch_type=ch_type,
                                   layout=layout, vmin=vmin,
                                   vmax=vmax, cmap=cmap, sensors=sensors,
                                   colorbar=colorbar, scale=scale,
                                   scale_time=scale_time,
                                   unit=unit, res=res, proj=proj, size=size,
                                   format=format, time_format=time_format,
                                   show=show, show_names=show_names,
                                   title=title, mask=mask,
                                   mask_params=mask_params,
                                   outlines=outlines, contours=contours,
                                   image_interp=image_interp)

    def plot_field(self, surf_maps, time=None, time_label='t = %0.0f ms',
                   n_jobs=1):
        """Plot MEG/EEG fields on head surface and helmet in 3D

        Parameters
        ----------
        surf_maps : list
            The surface mapping information obtained with make_field_map.
        time : float | None
            The time point at which the field map shall be displayed. If None,
            the average peak latency (across sensor types) is used.
        time_label : str
            How to print info about the time instant visualized.
        n_jobs : int
            Number of jobs to eggie in parallel.

        Returns
        -------
        fig : instance of mlab.Figure
            The mayavi figure.
        """
        return plot_evoked_field(self, surf_maps, time=time,
                                 time_label=time_label, n_jobs=n_jobs)

    def to_nitime(self, picks=None):
        """Export Evoked object to NiTime

        Parameters
        ----------
        picks : array-like of int | None
            Indices of channels to apply. If None, all channels will be
            exported.

        Returns
        -------
        evoked_ts : instance of nitime.TimeSeries
            The TimeSeries instance
        """
        try:
            from nitime import TimeSeries  # to avoid strong dependency
        except ImportError:
            raise Exception('the nitime package is missing')

        evoked_ts = TimeSeries(self.data if picks is None
                               else self.data[picks],
                               sampling_rate=self.info['sfreq'])
        return evoked_ts

    def as_data_frame(self, picks=None, scale_time=1e3, scalings=None,
                      use_time_index=True, copy=True):
        """Get the Evoked object as a Pandas DataFrame

        Export data in tabular structure: each row corresponds to a time point,
        and each column to a channel.

        Parameters
        ----------
        picks : array-like of int | None
            If None all channels are kept, otherwise the channels indices in
            picks are kept.
        scale_time : float
            Scaling to be applied to time units.
        scalings : dict | None
            Scaling to be applied to the channels picked. If None, defaults to
            ``scalings=dict(eeg=1e6, grad=1e13, mag=1e15, misc=1.0)`.
        use_time_index : bool
            If False, times will be included as in the data table, else it will
            be used as index object.
        copy : bool
            If true, evoked will be copied. Else data may be modified in place.

        Returns
        -------
        df : instance of pandas.core.DataFrame
            Evoked data exported into tabular data structure.
        """

        pd = _check_pandas_installed()

        if picks is None:
            picks = list(range(self.info['nchan']))
        else:
            if not in1d(picks, np.arange(len(self.ch_names))).all():
                raise ValueError('At least one picked channel is not present '
                                 'in this Evoked instance.')

        data, times = self.data, self.times

        if copy is True:
            data = data.copy()

        types = [channel_type(self.info, idx) for idx in picks]
        n_channel_types = 0
        ch_types_used = []

        scalings = _mutable_defaults(('scalings', scalings))[0]
        for t in scalings.keys():
            if t in types:
                n_channel_types += 1
                ch_types_used.append(t)

        for t in ch_types_used:
            scaling = scalings[t]
            idx = [picks[i] for i in range(len(picks)) if types[i] == t]
            if len(idx) > 0:
                data[idx] *= scaling

        assert times.shape[0] == data.shape[1]
        col_names = [self.ch_names[k] for k in picks]

        df = pd.DataFrame(data.T, columns=col_names)
        df.insert(0, 'time', times * scale_time)

        if use_time_index is True:
            if 'time' in df:
                df['time'] = df['time'].astype(np.int64)
            with warnings.catch_warnings(record=True):
                df.set_index('time', inplace=True)

        return df

    def resample(self, sfreq, npad=100, window='boxcar'):
        """Resample data

        This function operates in-place.

        Parameters
        ----------
        sfreq : float
            New sample rate to use
        npad : int
            Amount to pad the start and end of the data.
        window : string or tuple
            Window to use in resampling. See scipy.signal.resample.
        """
        o_sfreq = self.info['sfreq']
        self.data = resample(self.data, sfreq, o_sfreq, npad, -1, window)
        # adjust indirectly affected variables
        self.info['sfreq'] = sfreq
        self.times = (np.arange(self.data.shape[1], dtype=np.float) / sfreq
                      + self.times[0])
        self.first = int(self.times[0] * self.info['sfreq'])
        self.last = len(self.times) + self.first - 1

    def detrend(self, order=1, picks=None):
        """Detrend data

        This function operates in-place.

        Parameters
        ----------
        order : int
            Either 0 or 1, the order of the detrending. 0 is a constant
            (DC) detrend, 1 is a linear detrend.
        picks : array-like of int | None
            If None only MEG and EEG channels are detrended.
        """
        if picks is None:
            picks = pick_types(self.info, meg=True, eeg=True, ref_meg=False,
                               stim=False, eog=False, ecg=False, emg=False,
                               exclude='bads')
        self.data[picks] = detrend(self.data[picks], order, axis=-1)

    def copy(self):
        """Copy the instance of evoked

        Returns
        -------
        evoked : instance of Evoked
        """
        evoked = deepcopy(self)
        return evoked

    def __add__(self, evoked):
        """Add evoked taking into account number of epochs"""
        out = merge_evoked([self, evoked])
        out.comment = self.comment + " + " + evoked.comment
        return out

    def __sub__(self, evoked):
        """Add evoked taking into account number of epochs"""
        this_evoked = deepcopy(evoked)
        this_evoked.data *= -1.
        out = merge_evoked([self, this_evoked])
        out.comment = self.comment + " - " + this_evoked.comment
        return out

    def __hash__(self):
        return object_hash(dict(info=self.info, data=self.data))

    def get_peak(self, ch_type=None, tmin=None, tmax=None, mode='abs',
                 time_as_index=False):
        """Get location and latency of peak amplitude

        Parameters
        ----------
        ch_type : {'mag', 'grad', 'eeg', 'misc', None}
            The channel type to use. Defaults to None. If more than one sensor
            Type is present in the data the channel type has to be explicitly
            set.
        tmin : float | None
            The minimum point in time to be considered for peak getting.
        tmax : float | None
            The maximum point in time to be considered for peak getting.
        mode : {'pos', 'neg', 'abs'}
            How to deal with the sign of the data. If 'pos' only positive
            values will be considered. If 'neg' only negative values will
            be considered. If 'abs' absolute values will be considered.
            Defaults to 'abs'.
        time_as_index : bool
            Whether to return the time index instead of the latency in seconds.

        Returns
        -------
        ch_name : str
            The channel exhibiting the maximum response.
        latency : float | int
            The time point of the maximum response, either latency in seconds
            or index.
        """
        supported = ('mag', 'grad', 'eeg', 'misc', 'None')

        data_picks = pick_types(self.info, meg=True, eeg=True, ref_meg=False)
        types_used = set([channel_type(self.info, idx) for idx in data_picks])

        if str(ch_type) not in supported:
            raise ValueError('Channel type must be `{supported}`. You gave me '
                             '`{ch_type}` instead.'
                             .format(ch_type=ch_type,
                                     supported='` or `'.join(supported)))

        elif ch_type is not None and ch_type not in types_used:
            raise ValueError('Channel type `{ch_type}` not found in this '
                             'evoked object.'
                              .format(ch_type=ch_type))

        elif len(types_used) > 1 and ch_type is None:
            raise RuntimeError('More than one sensor type found. `ch_type` '
                               'must not be `None`, pass a sensor type '
                               'value instead')

        meg, eeg, misc, picks = False, False, False, None

        if ch_type == 'mag':
            meg = ch_type
        elif ch_type == 'grad':
            meg = ch_type
        elif ch_type == 'eeg':
            eeg = True
        elif ch_type == 'misc':
            misc = True

        if ch_type is not None:
            picks = pick_types(self.info, meg=meg, eeg=eeg, misc=misc,
                               ref_meg=False)

        data = self.data if picks is None else self.data[picks]
        ch_idx, time_idx = _get_peak(data, self.times, tmin,
                                     tmax, mode)

        return (self.ch_names[ch_idx],
                time_idx if time_as_index else self.times[time_idx])

class EvokedArray(Evoked):
    """Evoked object from numpy array

    Parameters
    ----------
    data : array of shape (n_channels, n_times)
        The channels' evoked response.
    info : instance of Info
        Info dictionary. Consider using ``create_info`` to populate
        this structure.
    tmin : float
        Start time before event.
    comment : string
        Comment on dataset. Can be the condition. Defaults to ''.
    nave : int
        Number of averaged epochs. Defaults to 1.
    kind : str
        Type of data, either average or standard_error. Defaults to 'average'.
    verbose : bool, str, int, or None
        If not None, override default verbose level (see mne.verbose).
        Defaults to raw.verbose.
    """

    @verbose
    def __init__(self, data, info, tmin, comment='', nave=1, kind='average',
                 verbose=None):

        dtype = np.complex128 if np.any(np.iscomplex(data)) else np.float64
        data = np.asanyarray(data, dtype=dtype)

        if data.ndim != 2:
            raise ValueError('Data must be a 2D array of shape (n_channels, '
                             'n_samples)')

        if len(info['ch_names']) != np.shape(data)[0]:
            raise ValueError('Info and data must have same number of '
                             'channels.')

        self.data = data

        # XXX: this should use round and be tested
        self.first = int(tmin * info['sfreq'])
        self.last = self.first + np.shape(data)[-1] - 1
        self.times = np.arange(self.first, self.last + 1, dtype=np.float)
        self.times /= info['sfreq']

        self.info = info
        self.nave = nave
        self.kind = kind
        self.comment = comment
        self.proj = None
        self.picks = None
        self.verbose = verbose
        self._projector = None
        if self.kind == 'average':
            self._aspect_kind = aspect_dict['average']
        else:
            self._aspect_kind = aspect_dict['standard_error']

def _get_entries(fid, evoked_node):
    """Helper to get all evoked entries"""
    comments = list()
    aspect_kinds = list()
    for ev in evoked_node:
        for k in range(ev['nent']):
            my_kind = ev['directory'][k].kind
            pos = ev['directory'][k].pos
            if my_kind == FIFF.FIFF_COMMENT:
                tag = read_tag(fid, pos)
                comments.append(tag.data)
        my_aspect = dir_tree_find(ev, FIFF.FIFFB_ASPECT)[0]
        for k in range(my_aspect['nent']):
            my_kind = my_aspect['directory'][k].kind
            pos = my_aspect['directory'][k].pos
            if my_kind == FIFF.FIFF_ASPECT_KIND:
                tag = read_tag(fid, pos)
                aspect_kinds.append(int(tag.data))
    comments = np.atleast_1d(comments)
    aspect_kinds = np.atleast_1d(aspect_kinds)
    if len(comments) != len(aspect_kinds) or len(comments) == 0:
        fid.close()
        raise ValueError('Dataset names in FIF file '
                         'could not be found.')
    t = [aspect_rev.get(str(a), 'Unknown') for a in aspect_kinds]
    t = ['"' + c + '" (' + tt + ')' for tt, c in zip(t, comments)]
    t = '  ' + '\n  '.join(t)
    return comments, aspect_kinds, t

def _get_evoked_node(fname):
    """Helper to get info in evoked file"""
    f, tree, _ = fiff_open(fname)
    with f as fid:
        _, meas = read_meas_info(fid, tree)
        evoked_node = dir_tree_find(meas, FIFF.FIFFB_EVOKED)
    return evoked_node

def merge_evoked(all_evoked):
    """Merge/concat evoked data

    Data should have the same channels and the same time instants.

    Parameters
    ----------
    all_evoked : list of Evoked
        The evoked datasets

    Returns
    -------
    evoked : Evoked
        The merged evoked data
    """
    evoked = deepcopy(all_evoked[0])

    ch_names = evoked.ch_names
    for e in all_evoked[1:]:
        assert e.ch_names == ch_names, ValueError("%s and %s do not contain "
                                                  "the same channels"
                                                  % (evoked, e))
        assert np.max(np.abs(e.times - evoked.times)) < 1e-7, \
            ValueError("%s and %s do not contain the same time "
                       "instants" % (evoked, e))

    # use union of bad channels
    bads = list(set(evoked.info['bads']).union(*(ev.info['bads']
                                                 for ev in all_evoked[1:])))
    evoked.info['bads'] = bads

    all_nave = sum(e.nave for e in all_evoked)
    evoked.data = sum(e.nave * e.data for e in all_evoked) / all_nave
    evoked.nave = all_nave
    return evoked

@deprecated("'read_evoked' will be removed in v0.9. Use 'read_evokeds.'")
def read_evoked(fname, setno=None, baseline=None, kind='average', proj=True):
    """Read an evoked dataset

    Parameters
    ----------
    fname : string
        The file name.
    setno : int or str | list of int or str | None
        The index or list of indices of the evoked dataset to read. FIF files
        can contain multiple datasets. If None and there is only one dataset in
        the file, this dataset is loaded.
    baseline : None (default) or tuple of length 2
        The time interval to apply baseline correction. If None do not apply it.
        If baseline is (a, b) the interval is between "a (s)" and "b (s)". If a
        is None the beginning of the data is used and if b is None then b is set
        to the end of the interval. If baseline is equal ot (None, None) all the
        time interval is used.
    kind : str
        Either 'average' or 'standard_error', the type of data to read.
    proj : bool
        If False, available projectors won't be applied to the data.

    Returns
    -------
    evoked : instance of Evoked or list of Evoked
        The evoked datasets.
    """
    evoked_node = _get_evoked_node(fname)
    if setno is None and len(evoked_node) > 1:
        fid, _, _ = fiff_open(fname)
        try:
            _, _, t = _get_entries(fid, evoked_node)
        except:
            t = 'None found, must use integer'
        else:
            fid.close()
        raise ValueError('%d datasets present, setno parameter must be set.'
                         'Candidate setno names:\n%s' % (len(evoked_node), t))
    elif isinstance(setno, list):
        return [Evoked(fname, s, baseline=baseline, kind=kind, proj=proj)
                for s in setno]
    else:
        if setno is None:
            setno = 0
        return Evoked(fname, setno, baseline=baseline, kind=kind, proj=proj)

@verbose
def read_evokeds(fname, condition=None, baseline=None, kind='average',
                 proj=True, verbose=None):
    """Read evoked dataset(s)

    Parameters
    ----------
    fname : string
        The file name, which should end with -ave.fif or -ave.fif.gz.
    condition : int or str | list of int or str | None
        The index or list of indices of the evoked dataset to read. FIF files
        can contain multiple datasets. If None, all datasets are returned as a
        list.
    baseline : None (default) or tuple of length 2
        The time interval to apply baseline correction. If None do not apply it.
        If baseline is (a, b) the interval is between "a (s)" and "b (s)". If a
        is None the beginning of the data is used and if b is None then b is set
        to the end of the interval. If baseline is equal ot (None, None) all the
        time interval is used.
    kind : str
        Either 'average' or 'standard_error', the type of data to read.
    proj : bool
        If False, available projectors won't be applied to the data.
    verbose : bool, str, int, or None
        If not None, override default verbose level (see mne.verbose).

    Returns
    -------
    evoked : Evoked (if condition is int or str) or list of Evoked (if
        condition is None or list)
        The evoked dataset(s).
    """
    check_fname(fname, 'evoked', ('-ave.fif', '-ave.fif.gz'))

    return_list = True
    if condition is None:
        evoked_node = _get_evoked_node(fname)
        condition = range(len(evoked_node))
    elif not isinstance(condition, list):
        condition = [condition]
        return_list = False

    out = [Evoked(fname, c, baseline=baseline, kind=kind, proj=proj,
           verbose=verbose) for c in condition]

    return out if return_list else out[0]

@deprecated("'write_evoked' will be removed in v0.9. Use 'write_evokeds.'")
def write_evoked(fname, evoked):
    """Write an evoked dataset to a file

    Parameters
    ----------
    fname : string
        The file name.
    evoked : instance of Evoked, or list of Evoked
        The evoked dataset to save, or a list of evoked datasets to save in one
        file. Note that the measurement info from the first evoked instance is
        used, so be sure that information matches.
    """

    if not isinstance(evoked, list):
        evoked = [evoked]

    # Create the file and save the essentials
    fid = start_file(fname)

    start_block(fid, FIFF.FIFFB_MEAS)
    write_id(fid, FIFF.FIFF_BLOCK_ID)
    if evoked[0].info['meas_id'] is not None:
        write_id(fid, FIFF.FIFF_PARENT_BLOCK_ID, evoked[0].info['meas_id'])

    # Write measurement info
    write_meas_info(fid, evoked[0].info)

    # One or more evoked data sets
    start_block(fid, FIFF.FIFFB_PROCESSED_DATA)
    for e in evoked:
        start_block(fid, FIFF.FIFFB_EVOKED)

        # Comment is optional
        if e.comment is not None and len(e.comment) > 0:
            write_string(fid, FIFF.FIFF_COMMENT, e.comment)

        # First and last sample
        write_int(fid, FIFF.FIFF_FIRST_SAMPLE, e.first)
        write_int(fid, FIFF.FIFF_LAST_SAMPLE, e.last)

        # The epoch itself
        start_block(fid, FIFF.FIFFB_ASPECT)

        write_int(fid, FIFF.FIFF_ASPECT_KIND, e._aspect_kind)
        write_int(fid, FIFF.FIFF_NAVE, e.nave)

        decal = np.zeros((e.info['nchan'], 1))
        for k in range(e.info['nchan']):
            decal[k] = 1.0 / (e.info['chs'][k]['cal']
                              * e.info['chs'][k].get('scale', 1.0))

        write_float_matrix(fid, FIFF.FIFF_EPOCH, decal * e.data)
        end_block(fid, FIFF.FIFFB_ASPECT)
        end_block(fid, FIFF.FIFFB_EVOKED)

    end_block(fid, FIFF.FIFFB_PROCESSED_DATA)
    end_block(fid, FIFF.FIFFB_MEAS)
    end_file(fid)

def write_evokeds(fname, evoked):
    """Write an evoked dataset to a file

    Parameters
    ----------
    fname : string
        The file name, which should end with -ave.fif or -ave.fif.gz.
    evoked : Evoked instance, or list of Evoked instances
        The evoked dataset, or list of evoked datasets, to save in one file.
        Note that the measurement info from the first evoked instance is used,
        so be sure that information matches.
    """
    check_fname(fname, 'evoked', ('-ave.fif', '-ave.fif.gz'))

    if not isinstance(evoked, list):
        evoked = [evoked]

    # Create the file and save the essentials
    with start_file(fname) as fid:

        start_block(fid, FIFF.FIFFB_MEAS)
        write_id(fid, FIFF.FIFF_BLOCK_ID)
        if evoked[0].info['meas_id'] is not None:
            write_id(fid, FIFF.FIFF_PARENT_BLOCK_ID, evoked[0].info['meas_id'])

        # Write measurement info
        write_meas_info(fid, evoked[0].info)

        # One or more evoked data sets
        start_block(fid, FIFF.FIFFB_PROCESSED_DATA)
        for e in evoked:
            start_block(fid, FIFF.FIFFB_EVOKED)

            # Comment is optional
            if e.comment is not None and len(e.comment) > 0:
                write_string(fid, FIFF.FIFF_COMMENT, e.comment)

            # First and last sample
            write_int(fid, FIFF.FIFF_FIRST_SAMPLE, e.first)
            write_int(fid, FIFF.FIFF_LAST_SAMPLE, e.last)

            # The epoch itself
            start_block(fid, FIFF.FIFFB_ASPECT)

            write_int(fid, FIFF.FIFF_ASPECT_KIND, e._aspect_kind)
            write_int(fid, FIFF.FIFF_NAVE, e.nave)

            decal = np.zeros((e.info['nchan'], 1))
            for k in range(e.info['nchan']):
                decal[k] = 1.0 / (e.info['chs'][k]['cal']
                                  * e.info['chs'][k].get('scale', 1.0))

            write_float_matrix(fid, FIFF.FIFF_EPOCH, decal * e.data)
            end_block(fid, FIFF.FIFFB_ASPECT)
            end_block(fid, FIFF.FIFFB_EVOKED)

        end_block(fid, FIFF.FIFFB_PROCESSED_DATA)
        end_block(fid, FIFF.FIFFB_MEAS)
        end_file(fid)

def _get_peak(data, times, tmin=None, tmax=None, mode='abs'):
    """Get feature-index and time of maximum signal from 2D array

    Note. This is a 'getter', not a 'finder'. For non-evoked type
    data and continuous signals, please use proper peak detection algorithms.

    Parameters
    ----------
    data : instance of numpy.ndarray (n_locations, n_times)
        The data, either evoked in sensor or source space.
    times : instance of numpy.ndarray (n_times)
        The times in seconds.
    tmin : float | None
        The minimum point in time to be considered for peak getting.
    tmax : float | None
        The maximum point in time to be considered for peak getting.
    mode : {'pos', 'neg', 'abs'}
        How to deal with the sign of the data. If 'pos' only positive
        values will be considered. If 'neg' only negative values will
        be considered. If 'abs' absolute values will be considered.
        Defaults to 'abs'.

    Returns
    -------
    max_loc : int
        The index of the feature with the maximum value.
    max_time : int
        The time point of the maximum response, index.
    """
    modes = ('abs', 'neg', 'pos')
    if mode not in modes:
        raise ValueError('The `mode` parameter must be `{modes}`. You gave '
                          'me `{mode}`'.format(modes='` or `'.join(modes),
                                               mode=mode))

    if tmin is None:
        tmin = times[0]
    if tmax is None:
        tmax = times[-1]

    if tmin < times.min():
        raise ValueError('The tmin value is out of bounds. It must be '
                         'within {0} and {1}'.format(times.min(), times.max()))
    if tmax > times.max():
        raise ValueError('The tmin value is out of bounds. It must be '
                         'within {0} and {1}'.format(times.min(), times.max()))
    if tmin >= tmax:
        raise ValueError('The tmin must be smaller than tmax')

    time_win = (times >= tmin) & (times <= tmax)
    mask = np.ones_like(data).astype(np.bool)
    mask[:, time_win] = False

    maxfun = np.argmax
    if mode == 'pos':
        if not np.any(data > 0):
            raise ValueError('No positive values encountered. Cannot '
                             'operate in pos mode.')
    elif mode == 'neg':
        if not np.any(data < 0):
            raise ValueError('No negative values encountered. Cannot '
                             'operate in neg mode.')
        maxfun = np.argmin

    masked_index = np.ma.array(np.abs(data) if mode == 'abs' else data,
                               mask=mask)

    max_loc, max_time = np.unravel_index(maxfun(masked_index), data.shape)

    return max_loc, max_time

max_stars_repo_name: jaeilepp/eggie

max_issues_repo_name: jaeilepp/eggie

max_forks_repo_name: jaeilepp/eggie


text:
# The class representing the cg iterative solver

from modules.model.labeling_module.Solvers.solver import Solver
from modules.model.labeling_module.ginkgo import Ginkgowrapper
import numpy as np

class CgSolver(Solver):

    # calculate the time it takes to solve the System Ax=b by calling the function in the c++ shared library
    #
    # @param self the instance of the class
    # @param ginkgo an instance of the class ginkgowrapper
    # @param the matrix which gets solved
    # @return the time it took to solve the system with the gmres solver
    def execute(self, ginkgo: Ginkgowrapper, matrix: np.ndarray) -> float:
        time = ginkgo.calculate_time_to_solve(matrix, 1)
        return time

ext: py

max_stars_repo_name: TheSlimvReal/PSE---LA-meets-ML

max_issues_repo_name: TheSlimvReal/PSE---LA-meets-ML

max_forks_repo_name: TheSlimvReal/PSE---LA-meets-ML

reason: import numpy


text:
import pytest
from py_wake.utils.check_input import check_input
import numpy as np

def test_check_input():
    input_space = [(0, 1), (100, 200)]

    with pytest.raises(ValueError, match="Input, index_0, with value, 2 outside range 0-1"):
        check_input(input_space, np.array([(2, 150)]).T)

    with pytest.raises(ValueError, match="Input, index_1, with value, 50 outside range 100-200"):
        check_input(input_space, np.array([(1, 50)]).T)

    with pytest.raises(ValueError, match="Input, wd, with value, 250 outside range 100-200"):
        check_input(input_space, np.array([(1, 250)]).T, ['ws', 'wd'])

    check_input(input_space, np.array([(1, 200)]).T, ['ws', 'wd'])

max_stars_repo_name: aemoser/PyWake

max_issues_repo_name: aemoser/PyWake

max_forks_repo_name: aemoser/PyWake


text:
import numpy as np
import os
import pickle
import matplotlib.pyplot as plt
import sys

np.set_printoptions(suppress=True)

datasets_uci = ['bostonHousing', 'concrete', 'energy', 'kin8nm', 'naval-propulsion-plant', 'power-plant', 'protein-tertiary-structure', 'wine-quality-red', 'yacht']
dataset_names = ['boston_housing', 'concrete', 'energy', 'kin8nm', 'naval', 'power', 'protein', 'wine', 'yacht']
input_dims = {'boston_housing': 13, 'concrete': 8, 'energy': 8, 'kin8nm': 8, 'power': 4, 'protein': 9, 'wine': 11, 'yacht': 6, 'naval': 16}

for i, dataset in enumerate(datasets_uci):
    data_directory = 'UCI_Datasets//' + dataset + '//data//'
    data = np.loadtxt(data_directory + 'data.txt')
    if dataset == 'naval-propulsion-plant':
        data = data[:,:-1] # regress the first of 2 targets
    split_directory = 'gap_datasets//' + dataset_names[i] + '_gap'
    os.mkdir(split_directory)

    for j in range(input_dims[dataset_names[i]]): # for each input dim, create a different split that has a gap in that dimension
        # get the split indices
        # look at the jth input
        jth_input = data[:,j]
        target = data[:,-1]
        sorting_indeces = np.argsort(jth_input)

        # remove 1/3 of data in the 'middle' as a test set
        num_data = target.size
        test_size = int(np.ceil(num_data/3))
        test_indices = sorting_indeces[int(num_data/2) - int(test_size/2):int(num_data/2) + int(test_size/2)]
        train_indices = np.concatenate((sorting_indeces[:int(num_data/2) - int(test_size/2)], sorting_indeces[int(num_data/2) + int(test_size/2):]))

        # extract train set
        train_set = data[train_indices,:]

        # extract test set
        test_set = data[test_indices,:]

        # normalise the train set
        train_mean = 1/(train_set.shape[0])*np.sum(train_set, axis=0)
        train_set_normalised = train_set - train_mean # broadcast

        train_sd = np.std(train_set_normalised, axis=0)
        for n in range(train_sd.size):
            if train_sd[n] < 1e-6:
                train_sd[n] = 1 # prevent division by zero if all the inputs are the same value, as in naval
        train_set_normalised = train_set_normalised/train_sd # broadcast

        # extract a validation set - 10% of the trainset
        val_size = int(train_indices.size/10)
        val_set_normalised = train_set_normalised[-val_size:,:]
        train_set_normalised = train_set_normalised[0:-val_size,:]

        # pickle
        filename = split_directory + '//' + dataset_names[i] + str(j) + '.pkl'
        with open(filename, 'wb') as f:
            pickle.dump([train_set, train_set_normalised, val_set_normalised, test_set, train_mean, train_sd], f)

        # write details to a text file
        np.savetxt(split_directory + '//train_indices_' + str(j) + '.txt', train_indices.astype(int), fmt='%i')
        np.savetxt(split_directory + '//test_indices' + str(j) + '.txt', test_indices.astype(int), fmt='%i')

        file = open(split_directory + '//details.txt','w')
        file.write('train set size : {} \n'.format(train_indices.shape[0]))
        file.write('test set size: {} \n'.format(test_set.shape[0]))
        file.close()


text:
import argparse
import datetime

import gym
import numpy as np

from deep_rlsp.model import StateVAE, ExperienceReplay

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("env_id", type=str)
    parser.add_argument("--n_rollouts", default=100, type=int)
    parser.add_argument("--state_size", default=30, type=int)
    parser.add_argument("--learning_rate", default=1e-5, type=float)
    parser.add_argument("--n_epochs", default=100, type=int)
    parser.add_argument("--batch_size", default=500, type=int)
    parser.add_argument("--n_layers", default=3, type=int)
    parser.add_argument("--layer_size", default=512, type=int)
    parser.add_argument("--prior_stdev", default=1, type=float)
    parser.add_argument("--divergence_factor", default=0.001, type=float)
    return parser.parse_args()

def main():
    args = parse_args()

    env = gym.make(args.env_id)
    experience_replay = ExperienceReplay(None)
    experience_replay.add_random_rollouts(
        env, env.spec.max_episode_steps, args.n_rollouts
    )

    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    label = "vae_{}_{}".format(args.env_id, timestamp)
    tensorboard_log = "tf_logs/tf_logs_" + label
    checkpoint_folder = "tf_ckpt/tf_ckpt_" + label

    vae = StateVAE(
        env.observation_space.shape[0],
        args.state_size,
        n_layers=args.n_layers,
        layer_size=args.layer_size,
        learning_rate=args.learning_rate,
        prior_stdev=args.prior_stdev,
        divergence_factor=args.divergence_factor,
        tensorboard_log=tensorboard_log,
        checkpoint_folder=checkpoint_folder,
    )
    # vae.checkpoint_folder = None

    vae.learn(experience_replay, args.n_epochs, args.batch_size, verbose=True)

    with np.printoptions(suppress=True):
        for _ in range(20):
            x = experience_replay.sample(1)[0][0]
            print("x", x)
            z = vae.encoder(x)
            print("z", z)
            x = vae.decoder(z)
            print("x2", x)

if __name__ == "__main__":
    main()

max_stars_repo_name: HumanCompatibleAI/deep-rlsp

max_issues_repo_name: HumanCompatibleAI/deep-rlsp

max_forks_repo_name: HumanCompatibleAI/deep-rlsp


text:
# Copyright 2015 The TensorFlow Authors. 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.
# ==============================================================================

import argparse
import numpy as np
import os
import tensorflow as tf
from fashion_mnist_classifier.models.dnn import DNN
from google.cloud import bigquery

_dataset_path = 'fashion_mnist_classifier/datasets/'
_weights_path = 'fashion_mnist_classifier/weights/'
_x_train_file = _dataset_path + 'x_train.npy'
_y_train_file = _dataset_path + 'y_train.npy'
_x_test_file = _dataset_path + 'x_test.npy'
_y_test_file = _dataset_path + 'y_test.npy'

def load_data_from_bigquery(project):
    def raw_to_feature(image_raw):
        return np.asarray(image_raw.split(','), 'float')

    dataset = 'fashion_mnist'
    train = 'train'
    test = 'test'

    # Get train and test tables.
    client = bigquery.Client(project=project)
    train_table = client.get_table(f'{project}.{dataset}.{train}')
    test_table = client.get_table(f'{project}.{dataset}.{test}')

    # # Load train data from BigQuery to x_train.npy and y_train.npy.
    if (not os.path.exists(_x_train_file) or
        not os.path.exists(_y_train_file)):
        # Load train data from BigQuery.
        rows_iter = client.list_rows(train_table)
        arrows = rows_iter.to_arrow()
        arrows_dict = arrows.to_pydict()

        # Revert x_train and y_train to ndarray.
        x_train, y_train = arrows_dict['image_raw'], arrows_dict['label']
        x_train = np.array([raw_to_feature(image_raw) for image_raw in x_train], 'float')
        y_train = np.array(y_train)

        # Save x_train as x_train.npy and y_train as y_train.npy.
        np.save(_x_train_file, x_train)
        np.save(_y_train_file, y_train)

        print(f'Downloaded {x_train.shape[0]} rows into {_x_train_file}')
        print(f'Downloaded {y_train.shape[0]} rows into {_y_train_file}')

    # Load test data from BigQuery to x_test.npy and y_test.npy.
    if (not os.path.exists(_x_test_file) or
        not os.path.exists(_y_test_file)):
        # Load test data from BigQuery.
        rows_iter = client.list_rows(test_table)
        arrows = rows_iter.to_arrow()
        arrows_dict = arrows.to_pydict()

        # Revert x_test and y_test to ndarray.
        x_test, y_test = arrows_dict['image_raw'], arrows_dict['label']
        x_test = np.array([raw_to_feature(image_raw) for image_raw in x_test], 'float')
        y_test = np.array(y_test)

        # Save x_test as x_test.npy and y_test as y_test.npy.
        np.save(_x_test_file, x_test)
        np.save(_y_test_file, y_test)

        print(f'Downloaded {x_test.shape[0]} rows into {_x_test_file}')
        print(f'Downloaded {y_test.shape[0]} rows into {_y_test_file}')

def generator(images, labels):
    def _generator():
        for image, label in zip(images, labels):
            yield image, label

    return _generator

def preprocess_data(image, label):
    image = image / 255.0
    image = tf.reshape(image, [28, 28, 1])

    # The image data are mapped to 'image' feature column.
    return {'image': image}, label

def train_input_fn():
    x_train = np.load(_x_train_file)
    y_train = np.load(_y_train_file)

    train_dataset = tf.data.Dataset.from_generator(
        generator(x_train, y_train),
        (tf.float32, tf.int32),
        (tf.TensorShape([28 * 28 * 1]), tf.TensorShape([]))
    )
    train_dataset = train_dataset.map(preprocess_data).batch(1024)

    return train_dataset

def predict_input_fn():
    x_test = np.load(_x_test_file)
    y_test = np.load(_y_test_file)

    test_dataset = tf.data.Dataset.from_generator(
        generator(x_test, y_test),
        (tf.float32, tf.int32),
        (tf.TensorShape([28 * 28 * 1]), tf.TensorShape([]))
    )
    test_dataset = test_dataset.map(preprocess_data).batch(1024)

    return test_dataset

def serving_input_receiver_fn():
    receiver_tensors = {
        'image': tf.compat.v1.placeholder(
            dtype=tf.float32, shape=[None, 28, 28, 1], name='image'
        )
    }
    features = {'image': receiver_tensors['image']}

    return tf.estimator.export.ServingInputReceiver(features, receiver_tensors)

if __name__ == "__main__":
    description = """
    This script generates fashion_mnist data and upload it as a
    BigQuery dataset under [PROJECT ID].fashion_mnist.{train,test}.

    To run this script successfully, you need to complete the following steps:
    1. Select or create a Cloud Platform project.
    2. Enable billing for your project.
    3. Enable the Google Cloud BigQuery API.
    4. Setup Authentication.

    Please see (https://googleapis.dev/python/bigquery/latest/index.html) for more information."""

    parser = argparse.ArgumentParser(description=description,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--project', type=str, default='mlops-basic', help='Your GCP project id')

    args = parser.parse_args()

    # Prepare data.
    load_data_from_bigquery(args.project)

    # Prepare model.
    dnn = DNN()
    estimator = dnn.get_estimator()

    # Train and evaluate model.
    estimator.train(input_fn=train_input_fn)
    evaluation = estimator.evaluate(input_fn=predict_input_fn)
    print(evaluation)

    # Save trained model.
    estimator.export_saved_model(
        export_dir_base=_weights_path,
        serving_input_receiver_fn=serving_input_receiver_fn
    )

max_stars_repo_name: jihoonsong/mlops-basic

max_issues_repo_name: jihoonsong/mlops-basic

max_forks_repo_name: jihoonsong/mlops-basic


text:
import os
import random
import xml.etree.ElementTree as ET
from functools import partial
from glob import glob

import numpy as np
import tensorflow as tf
from PIL import Image

from box_utils import compute_target
from image_utils import horizontal_flip, random_patching

class XRayDataset():
    """ Class for XRayDataset

    Attributes:
        root_dir: dataset root dir (ex: ./data/VOCdevkit)
        num_examples: number of examples to be used
                      (in case one wants to overfit small data)
    """

    def __init__(self, root_dir, default_boxes,
                 new_size, num_examples=-1, augmentation=None):
        super(XRayDataset, self).__init__()
        self.idx_to_name = [
            'sponge1', 'needle']
        self.name_to_idx = dict([(v, k)
                                 for k, v in enumerate(self.idx_to_name)])

        self.data_dir = root_dir
        self.ids = list(map(lambda x: os.path.basename(x).split('.')[0], glob(os.path.join(self.data_dir, '*implant*.xml'))))
        random.shuffle(self.ids)
        self.default_boxes = default_boxes
        self.new_size = new_size

        if num_examples != -1:
            self.ids = self.ids[:num_examples]

        self.train_ids = self.ids[:int(len(self.ids) * 0.75)]
        self.val_ids = self.ids[int(len(self.ids) * 0.75):]

        if augmentation == None:
            self.augmentation = ['original']
        else:
            self.augmentation = augmentation + ['original']

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

    def _get_image(self, index):
        """ Method to read image from file
            then resize to (300, 300)
            then subtract by ImageNet's mean
            then convert to Tensor

        Args:
            index: the index to get filename from self.ids

        Returns:
            img: tensor of shape (3, 300, 300)
        """
        filename = self.ids[index]
        img_path = os.path.join(self.data_dir, filename + '.implant.png')
        img = Image.open(img_path)

        return img

    def _get_annotation(self, index, orig_shape):
        """ Method to read annotation from file
            Boxes are normalized to image size
            Integer labels are increased by 1

        Args:
            index: the index to get filename from self.ids
            orig_shape: image's original shape

        Returns:
            boxes: numpy array of shape (num_gt, 4)
            labels: numpy array of shape (num_gt,)
        """
        h, w = orig_shape
        filename = self.ids[index]
        anno_path = os.path.join(self.data_dir, filename + '.implant.sponge1.posWin.xml')
        windows = ET.parse(anno_path).find('Windows') # Get the Windows tag
        objects = windows.findall('_') # Find all objects in image
        boxes = []
        labels = []

        for obj in objects:
            name = 'sponge1'
            # name = obj.find('name').text.lower().strip()
            xmin = (float(obj.find('x').text))
            ymin = (float(obj.find('y').text))

            xmin = (float(obj.find('x').text) - 1) / w
            ymin = (float(obj.find('y').text) - 1) / h
            xmax = (float(obj.find('width').text) - 1) / w + xmin
            ymax = (float(obj.find('height').text) - 1) / h + ymin
            boxes.append([xmin, ymin, xmax, ymax])

            labels.append(self.name_to_idx[name] + 1)

        return np.array(boxes, dtype=np.float32), np.array(labels, dtype=np.int64)

    def generate(self, subset=None):
        """ The __getitem__ method
            so that the object can be iterable

        Args:
            index: the index to get filename from self.ids

        Returns:
            img: tensor of shape (300, 300, 3)
            boxes: tensor of shape (num_gt, 4)
            labels: tensor of shape (num_gt,)
        """
        if subset == 'train':
            indices = self.train_ids
        elif subset == 'val':
            indices = self.val_ids
        else:
            indices = self.ids
        for index in range(len(indices)):
            # img, orig_shape = self._get_image(index)
            filename = indices[index]
            img = self._get_image(index)
            w, h = img.size
            boxes, labels = self._get_annotation(index, (h, w))
            boxes = tf.constant(boxes, dtype=tf.float32)
            labels = tf.constant(labels, dtype=tf.int64)

            augmentation_method = np.random.choice(self.augmentation)
            if augmentation_method == 'patch':
                img, boxes, labels = random_patching(img, boxes, labels)
            elif augmentation_method == 'flip':
                img, boxes, labels = horizontal_flip(img, boxes, labels)

            img = np.array(img.resize(
                (self.new_size, self.new_size)), dtype=np.float32)
            img = np.repeat(img[:,:,np.newaxis], 3, axis=2) # transform grayscale into "forced" rgb
            img = (img / 127.0) - 1.0
            img = tf.constant(img, dtype=tf.float32)

            # print(tf.shape(img))

            gt_confs, gt_locs = compute_target(
                self.default_boxes, boxes, labels)

            yield filename, img, gt_confs, gt_locs

def create_batch_generator(root_dir, default_boxes,
                           new_size, batch_size, num_batches,
                           mode,
                           augmentation=None):
    num_examples = batch_size * num_batches if num_batches > 0 else -1
    xray = XRayDataset(root_dir, default_boxes,
                       new_size, num_examples, augmentation)

    info = {
        'idx_to_name': xray.idx_to_name,
        'name_to_idx': xray.name_to_idx,
        'length': len(xray),
        'image_dir': xray.data_dir,
    }

    print(f'dataset contains {len(xray)} samples')

    if mode == 'train':
        train_gen = partial(xray.generate, subset='train')
        train_dataset = tf.data.Dataset.from_generator(
            train_gen, (tf.string, tf.float32, tf.int64, tf.float32))
        val_gen = partial(xray.generate, subset='val')
        val_dataset = tf.data.Dataset.from_generator(
            val_gen, (tf.string, tf.float32, tf.int64, tf.float32))

        train_dataset = train_dataset.shuffle(40).batch(batch_size)
        val_dataset = val_dataset.batch(batch_size)

        return train_dataset.take(num_batches), val_dataset.take(-1), info
    else:
        dataset = tf.data.Dataset.from_generator(
            xray.generate, (tf.string, tf.float32, tf.int64, tf.float32))
        dataset = dataset.batch(batch_size)
        return dataset.take(num_batches), info

from anchor import generate_default_boxes
import yaml
if __name__ == "__main__":

    with open('./config.yml') as f:
        cfg = yaml.safe_load(f)

    try:
        config = cfg['SSD300']
    except AttributeError:
        raise ValueError('Unknown architecture: {}'.format(config))

    default_boxes = generate_default_boxes(config)

    batch_generator, val_generator, info = create_batch_generator(
        '../xraydata/sponge1/',
        default_boxes,
        300,
        32,
        -1,
        'train'
    )

    for i, (_, imgs, gt_confs, gt_locs) in enumerate(batch_generator):
        print(f'Image {i}, {gt_confs}')


text:
# main.py - Semi-supervised node classfication

import os, inspect
from time import time
import numpy as np
from tqdm import tqdm

import torch
import torch.nn.functional as F
import torch.optim as optim

from config.config import config
from data.data_helper import BasicDataset
from src.models import BasicModel, MyLightGCN, MyGCN
from src.utils import accuracy, sample_negatives_

import warnings
warnings.filterwarnings("ignore")

def train(model: BasicModel, optimizer: optim.Adam, dataset: BasicDataset):
    Y = dataset.getLabels().to(config["device"])
    train_idx = dataset.train_idx

    model.train()

    t_total = time()
    for epoch in tqdm(range(config['epochs']), desc="Training"):
        t = time()
        optimizer.zero_grad()

        Z = model.forward()

        classification_loss = F.nll_loss(Z[train_idx], Y[train_idx])
        # acc = accuracy(output=Z[train_idx], labels=Y[train_idx])
        loss = config["lambda"] * classification_loss

        if config["lambda"] < 1:
            negatives = sample_negatives_(dataset, config['num_negatives'], config["device"])
            topology_loss = model.topology_loss(negatives)
            loss += (1-config["lambda"]) * topology_loss

        loss.backward()
        optimizer.step()

    train_time = time() - t_total
    print(' >> Optimization finished! Total elapsed time: {:.4f} s'.format(train_time))
    return train_time

def test(model: BasicModel, dataset):
    Y = dataset.getLabels().to(config["device"])

    model.eval()
    Z = model.forward()

    test_idx = dataset.test_idx
    test_loss = F.nll_loss(Z[test_idx], Y[test_idx])
    test_accuracy = accuracy(Z[test_idx], Y[test_idx])

    print("[Test set results]",
          "loss= {:.8f}".format(test_loss.item()),
          "accuracy= {:.8f}".format(test_accuracy.item()),
          "error= {:.8f}".format(100*(1-test_accuracy.item())),
          "\n")
    return test_accuracy.item()

np.random.seed(config['seed'])
torch.manual_seed(config['seed'])
if config['cuda']: torch.cuda.manual_seed(config['seed'])

# print("[data]", config['data'], "[split]", config['split'], "[n_hid]", config['latent_dim'])

# Load data
dataset = BasicDataset(config)

if config['use_light']:
    # LightGCN model
    print("Aggregator: LightGCN")
    model = MyLightGCN(config=config, dataset=dataset)
else:
    # GCN model
    print("Aggregator: GCN")
    model = MyGCN(config=config, dataset=dataset)

# Optimizer
optimizer = optim.Adam(
    filter(lambda p: p.requires_grad, list(model.parameters())),
    lr=config['lr'],
    weight_decay=config['weight_decay']
)

# Train and Test
train_time = train(model, optimizer, dataset)
test_accuracy = test(model, dataset)

# with open('light_gcn.txt', 'a') as f:
#     f.write(f"{config['data']} {config['split']} {test_accuracy}\t{100*(1-test_accuracy)}\t{train_time}\n")

if config["save_model"]:
    current = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
    d = os.path.join(current, 'result', config['data'], f"{config['data']}_lightgcn_s{config['split']}.pt")
    torch.save(model, d)

max_stars_repo_name: puzzlepcs/pygcn

max_issues_repo_name: puzzlepcs/pygcn

max_forks_repo_name: puzzlepcs/pygcn


text:
"""Perform PCA on Fashion-MNIST.

Author: Yuhuang Hu
Email : yuhuang.hu@ini.uzh.ch
"""
from __future__ import print_function, absolute_import

import numpy as np

from pnslib import utils
from pnslib import ml

# load data
(train_x, train_y, test_x, test_y) = utils.binary_fashion_mnist_load(
    flatten=True)

train_x = train_x.astype("float32")/255.
mean_train_x = np.mean(train_x, axis=0)
train_x -= mean_train_x
test_x = train_x.astype("float32")/255.
test_x -= mean_train_x

print (train_x.dtype)
print (test_x.dtype)

# perform PCA on training dataset
train_x, R, n_retained = ml.pca(train_x)

# perform PCA on testing dataset
test_x = ml.pca_fit(test_x, R, n_retained)

print (train_x.shape)
print (test_x.shape)


text:
#    _____           ______  _____
#  / ____/    /\    |  ____ |  __ \
# | |        /  \   | |__   | |__) | Caer - Modern Computer Vision
# | |       / /\ \  |  __|  |  _  /  Languages: Python, C, C++
# | |___   / ____ \ | |____ | | \ \  http://github.com/jasmcaus/caer
#  \_____\/_/    \_ \______ |_|  \_\

# Licensed under the MIT License <http://opensource.org/licenses/MIT>
# SPDX-License-Identifier: MIT
# Copyright (c) 2020-2021 The Caer Authors <http://github.com/jasmcaus>

#pylint:disable=unused-argument,unused-variable,eval-used

import numpy as np
import random
from PIL import Image, ImageEnhance
import cv2 as cv
import math

from ..adorad import Tensor, is_tensor

from .functional import (
    is_list,
    _hls,
    _exposure_process
)

from ..color import (
    hls_to_bgr,
    hls_to_rgb
)

__all__ = [
    'adjust_brightness',
    'adjust_contrast',
    'adjust_hue',
    'adjust_saturation',
    'adjust_gamma',
    'affine',
    'darken',
    'brighten',
    'random_brightness',
    'correct_exposure',
    'augment_random'
]

def adjust_brightness(img, coeff, rgb=True) -> Tensor:
    r"""
        Adjust the brightness of an image.

    Args:
        img (Tensor) : Any regular BGR/RGB image.
        coeff (int): Coefficient value.
            - ``coeff < 1``, the image is darkened.
            - ``coeff = 1``, the image is unchanged.
            - ``coeff > 1``, the image is lightened.
        rgb (bool): Operate on RGB images. Default: True.

    Returns:
        Array of shape ``(height, width, channels)``.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.adjust_brightness(img, coeff=1.4, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    img = _hls(img, rgb=rgb)

    img = np.array(img, dtype=np.float64)
    img[:,:,1] = img[:,:,1]*coeff ## scale pixel values up or down for channel 1 (for lightness)

    if coeff > 1:
        img[:,:,1][img[:,:,1]>255]  = 255 # Set all values > 255 to 255
    else:
        img[:,:,1][img[:,:,1]<0]=0

    img = np.array(img, dtype=np.uint8)

    if rgb:
        return hls_to_rgb(img)
    else:
        return hls_to_bgr(img)

def brighten(img, coeff=-1, rgb=True) -> Tensor:
    r"""
        Brighten an image.

    Args:
        img (Tensor) : Any regular BGR/RGB image.
        coeff (int): Coefficient value.
        rgb (bool): Operate on RGB images. Default: True.

    Returns:
        Array of shape ``(height, width, channels)``.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.brighten(img, coeff=-1, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    if coeff !=-1:
        if coeff < 0.0 or coeff > 1.0:
            raise ValueError('Brightness coefficient can only be between 0.0 and 1.0')

    if coeff == -1:
        coeff_t = 1 + random.uniform(0, 1) # coeff between 1.0 and 1.5
    else:
        coeff_t = 1 + coeff  # coeff between 1.0 and 2.0

    return adjust_brightness(img, coeff_t, rgb=rgb)

def darken(img, darkness_coeff = -1, rgb=True) -> Tensor:
    r"""
        Darken an image.

    Args:
        img (Tensor) : Any regular BGR/RGB image.
        darkness_coeff (int): Coefficient value.
        rgb (bool): Operate on RGB images. Default: True.

    Returns:
        Array of shape ``(height, width, channels)``.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.darken(img, coeff=-1, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    if darkness_coeff != -1:
        if darkness_coeff < 0.0 or darkness_coeff > 1.0:
            raise ValueError('Darkness coeff must only be between 0.0 and 1.0')

    if darkness_coeff == -1:
        darkness_coeff_t = 1 - random.uniform(0, 1)
    else:
        darkness_coeff_t = 1 - darkness_coeff

    return adjust_brightness(img, darkness_coeff_t, rgb=True)

def random_brightness(img, rgb=True) -> Tensor:
    r"""
        Add random brightness to an image.

    Args:
        img (Tensor) : Any regular BGR/RGB image.
        rgb (bool): Operate on RGB images. Default: True.

    Returns:
        Array of shape ``(height, width, channels)``.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.random_brightness(img, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    rand_br_coeff = 2 * np.random.uniform(0, 1) # Generates a value between 0.0 and 2.0
    return adjust_brightness(img, rand_br_coeff, rgb=rgb)

def adjust_contrast(img, contrast_factor) -> Tensor:
    """
        Adjust contrast of an image.

    Args:
        img (Tensor): Any valid BGR/RGB image.
        contrast_factor (float): How much to adjust the contrast. Can be any
            non negative number. 0 gives a solid gray image, 1 gives the
            original image while 2 increases the contrast by a factor of 2.
    Returns:
        numpy Tensor: Contrast adjusted image.
    """
    # It's much faster to use the LUT construction because you have to change dtypes multiple times
    if not is_tensor(img):
        raise TypeError('Expected Numpy Tensor. Got {}'.format(type(img)))

    table = np.array([(i - 74) * contrast_factor + 74
                      for i in range(0, 256)]).clip(0, 255).astype('uint8')

    # enhancer = ImageEnhance.Contrast(img)
    # img = enhancer.enhance(contrast_factor)
    return cv.LUT(img, table)

def adjust_saturation(img, saturation_factor) -> Tensor:
    """Adjust color saturation of an image.
    Args:
        img (numpy Tensor): Any valid BGR/RGB image.
        saturation_factor (float):  How much to adjust the saturation. 0 will
            give a black and white image, 1 will give the original image while
            2 will enhance the saturation by a factor of 2.

    Returns:
        Saturation-adjusted image.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.adjust_saturation(img, saturation_factor=1.5, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    # ~10ms slower than PIL!
    if not is_tensor(img):
        raise TypeError('Expected Numpy Tensor. Got {}'.format(type(img)))

    img = Image.fromarray(img)
    enhancer = ImageEnhance.Color(img)
    img = enhancer.enhance(saturation_factor)

    return np.array(img)

def adjust_hue(img, hue_factor) -> Tensor:
    r"""
        Adjust hue of an image.

        The image hue is adjusted by converting the image to HSV and cyclically shifting the intensities in the hue channel (H). The image is then converted back to original image mode.

        See `Hue`_ for more details.
        .. _Hue: https://en.wikipedia.org/wiki/Hue

    Args:
        img (Tensor): Any valid BGR/RGB image.
        hue_factor (float):  How much to shift the hue channel. Should be in the range [-0.5, 0.5].
            0.5 and -0.5 give complete reversal of hue channel in HSV space in positive and negative direction respectively.
            0 means no shift. Therefore, both -0.5 and 0.5 will give an image with complementary colors while 0 gives the original image.

    Returns:
        Hue adjusted image.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.adjust_hue(img, hue_factor=1, rgb=True)
        >> filtered
        (427, 640, 3)

    """

    if not (-0.5 <= hue_factor <= 0.5):
        raise ValueError('`hue_factor` is not in [-0.5, 0.5].')

    if not is_tensor(img):
        raise TypeError('Expected Numpy Tensor. Got {}'.format(type(img)))

    img = Image.fromarray(img)
    input_mode = img.mode
    if input_mode in {'L', '1', 'I', 'F'}:
        return np.array(img)

    h, s, v = img.convert('HSV').split()

    np_h = np.array(h, dtype=np.uint8)

    # uint8 addition take cares of rotation across boundaries
    with np.errstate(over='ignore'):
        np_h += np.uint8(hue_factor * 255)
    h = Image.fromarray(np_h, 'L')

    img = Image.merge('HSV', (h, s, v)).convert(input_mode)
    return np.array(img)

def adjust_gamma(img, gamma, gain=1) -> Tensor:
    r"""
        Perform gamma correction on an image.

        Also known as Power Law Transform. Intensities in RGB mode are adjusted
        based on the following equation:
        .. math::
            I_{\text{out}} = 255 \times \text{gain} \times \left(\frac{I_{\text{in}}}{255}\right)^{\gamma}
        See `Gamma Correction`_ for more details.
        .. _Gamma Correction: https://en.wikipedia.org/wiki/Gamma_correction

    Args:
        img (Tensor): Any valid BGR/RGB image.
        gamma (float): Non negative real number, same as :math:`\gamma` in the equation.
            gamma larger than 1 make the shadows darker,
            while gamma smaller than 1 make dark regions lighter.
        gain (float): The constant multiplier.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.adjust_gamma(img, gamma=1.5, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    if not is_tensor(img):
        raise TypeError('Expected Numpy Tensor. Got {}'.format(type(img)))

    if gamma < 0:
        raise ValueError('Gamma should be a non-negative real number')

    # from here
    # https://stackoverflow.com/questions/33322488/how-to-change-image-illumination-in-opencv-python/41061351
    table = np.array([((i / 255.0)**gamma) * 255 * gain
                      for i in np.arange(0, 256)]).astype('uint8')

    return cv.LUT(img, table)

def _get_affine_matrix(center, angle, translate, scale, shear) -> Tensor:
    # Helper method to compute matrix for affine transformation
    # We need compute affine transformation matrix: M = T * C * RSS * C^-1
    # where T is translation matrix: [1, 0, tx | 0, 1, ty | 0, 0, 1]
    #       C is translation matrix to keep center: [1, 0, cx | 0, 1, cy | 0, 0, 1]
    #       RSS is rotation with scale and shear matrix
    #       RSS(a, scale, shear) = [ cos(a)*scale    -sin(a + shear)*scale     0]
    #                              [ sin(a)*scale    cos(a + shear)*scale     0]
    #                              [     0                  0          1]

    angle = math.radians(angle)
    shear = math.radians(shear)
    # scale = 1.0 / scale

    T = np.array([[1, 0, translate[0]], [0, 1, translate[1]], [0, 0, 1]])
    C = np.array([[1, 0, center[0]], [0, 1, center[1]], [0, 0, 1]])
    RSS = np.array(
        [[math.cos(angle) * scale, -math.sin(angle + shear) * scale, 0],
         [math.sin(angle) * scale,
          math.cos(angle + shear) * scale, 0], [0, 0, 1]])
    matrix = T @ C @ RSS @ np.linalg.inv(C)

    return matrix[:2, :]

def affine(img,
           angle,
           translate,
           scale,
           shear,
           interpolation='bilinear',
           mode=0,
           fillcolor=0) -> Tensor:
    """
        Apply affine transformation on the image keeping image center invariant.

    Args:
        img (Tensor): Any valid BGR/RGB image.
        angle (float or int): Rotation angle in degrees between -180 and 180, clockwise direction.
        translate (list or tuple of integers): Horizontal and vertical translations (post-rotation translation)
        scale (float): Overall scale
        shear (float): Shear angle value in degrees between -180 to 180, clockwise direction.
        interpolation (int, str): Interpolation to use for resizing. Defaults to `'bilinear'`.
                Supports `'bilinear'`, `'bicubic'`, `'area'`, `'nearest'`.
        mode (int, str): Method for filling in border regions.
                Defaults to ``constant`` meaning areas outside the image are filled with a value (val, default 0).
                Supports ``'replicate'``, ``'reflect'``, ``'reflect-101'``.
        val (int): Optional fill color for the area outside the transform in the output image. Default: 0
    """

    if not is_tensor(img):
        raise TypeError('Expected Numpy Tensor. Got {}'.format(type(img)))

    assert isinstance(translate, (tuple, list)) and len(translate) == 2, \
        'Argument translate should be a list or tuple of length 2'

    assert scale > 0.0, 'Argument scale should be positive'

    interpolation_methods = {
        'nearest': 0,  '0': 0,  0: 0, # 0
        'bilinear': 1, '1': 1,  1: 1, # 1
        'bicubic': 2,  '2': 2,  2: 2, # 2
        'area': 3,     '3': 3,  3: 3 # 3
    }
    border_methods = {
        'constant': 0,    '0': 0, 0: 0, # 0
        'replicate': 1,   '1': 1, 1: 1, # 1
        'reflect': 2,     '2': 2, 2: 2, # 2
        'reflect-101': 4, '4': 4, 4: 4 # 4
    }

    if interpolation not in interpolation_methods:
        raise ValueError('Specify a valid interpolation type - area/nearest/bicubic/bilinear')

    if mode not in border_methods:
        raise ValueError('Specify a valid border type - constant/replicate/reflect/reflect-101')

    output_size = img.shape[0:2]
    center = (img.shape[1] * 0.5 + 0.5, img.shape[0] * 0.5 + 0.5)
    matrix = _get_affine_matrix(center, angle, translate, scale, shear)

    return cv.warpAffine(img,
                            matrix,
                            output_size[::-1],
                            interpolation,
                            borderMode=mode,
                            borderValue=fillcolor)

def correct_exposure(img, rgb=True) -> Tensor:
    r"""
        Correct the exposure of an image.

    Args:
        img (Tensor) : Any regular BGR/RGB image.
        rgb (bool): Operate on RGB images. Default: True.

    Returns:
        Array of shape ``(height, width, channels)``.

    Examples::

        >> img = caer.data.sunrise(rgb=True)
        >> filtered = caer.transforms.correct_exposure(img, rgb=True)
        >> filtered
        (427, 640, 3)

    """
    return _exposure_process(img, rgb=rgb)

def augment_random(img, aug_types='', volume='expand' ) -> Tensor:
    aug_types_all = ['random_brightness','add_shadow','add_snow','add_rain','add_fog','add_gravel','add_sun_flare','add_motion_blur','add_autumn','random_flip']

    if aug_types=='':
        aug_types=aug_types_all
    output=[]

    if not is_list(aug_types):
        raise ValueError('`aug_types` should be a list of function names (str)')

    if volume == 'expand':
        for aug_type in aug_types:
            if not(aug_type in aug_types_all):
                raise ValueError('Incorrect transformation function defined')

            command = aug_type + '(img)'
            result = eval(command)

            output.append(result)

    elif volume == 'same':
        for aug_type in aug_types:
            if not (aug_type in aug_types_all):
                raise ValueError('Incorrect transformation function defined')

        selected_aug = aug_types[random.randint(0, len(aug_types)-1)]
        command = selected_aug+'(img)'
        output = eval(command)

    else:
        raise ValueError('volume type can only be "same" or "expand"')

    return output

max_stars_repo_name: MalikMAlna/caer

max_issues_repo_name: MalikMAlna/caer

max_forks_repo_name: MalikMAlna/caer


text:
import numpy as np

def numpy_random9_test(low, high):
  return np.random.randint(low, high)

max_stars_repo_name: maropu/lljvm-translator

max_issues_repo_name: maropu/lljvm-as

max_forks_repo_name: maropu/lljvm-as


text:
import pytest
import os.path
import numpy as np
from astropy.utils import find_current_module
from astropy.nddata import NDDataArray
from ..io import NDMapIO
from ..data import FileName, DataFile, DataFileList, NDLater

# Determine path to data in common to some of the tests below:
# Using get_pkg_data_filename() from astropy here was causing problems with
# py.test expecting remote data.
#fn_mefnodq = get_pkg_data_filename('data/eqbprgS20120827S0069_flat.fits')
module = find_current_module(0, True)
module_path = os.path.dirname(module.__file__)
fn_mefnodq = os.path.join(module_path, 'data', 'eqbprgS20120827S0069_flat.fits')

def test_FileName_gemini_IRAF_1():

    fn = FileName('/home/james/rgS20120827S0066.fits')

    assert fn.dir == '/home/james' and \
           fn.prefix == 'rg' and \
           fn.base == 'S20120827S0066' and \
           fn.suffix == [] and \
           fn.ext == 'fits' and \
           fn.standard == True and \
           str(fn) == '/home/james/rgS20120827S0066.fits'

def test_FileName_nonconforming_1():

    fn = FileName('/home/fred/blah.fits')

    assert fn.dir == '/home/fred' and \
           fn.prefix == '' and \
           fn.base == 'blah' and \
           fn.suffix == [] and \
           fn.ext == 'fits' and \
           fn.standard == False and \
           str(fn) == '/home/fred/blah.fits'

def test_FileName_multiple_ext():

    fn = FileName('some.dir/rgS20120827S0066.fits.gz')

    assert fn.dir == 'some.dir' and \
           fn.prefix == 'rg' and \
           fn.base == 'S20120827S0066' and \
           fn.suffix == [] and \
           fn.ext == 'fits.gz' and \
           fn.standard == True and \
           str(fn) == 'some.dir/rgS20120827S0066.fits.gz'

def test_DataFile_gemini_IRAF_new_1():

    # Use a filename that can't exist, just to be sure it doesn't...
    df = DataFile('rgS20120827S9999.fits', mode='new')

    assert df.filename.dir == '' and \
           df.filename.prefix == 'rg' and \
           df.filename.base == 'S20120827S9999' and \
           df.filename.suffix == [] and \
           df.filename.ext == 'fits' and \
           df.filename.standard == True and \
           len(df) == 0

def test_DataFile_gemini_IRAF_existing_1():

    df = DataFile(filename=fn_mefnodq)

    # Ignore the directory name because the above call translates it to the
    # absolute location of this package on the system.
    assert df.filename.prefix == 'eqbprg' and \
           df.filename.base == 'S20120827S0069' and \
           df.filename.suffix == ['_flat'] and \
           df.filename.ext == 'fits' and \
           df.filename.standard == True and \
           df.meta['OBJECT'] == 'GCALflat' and \
           len(df) == 2 and \
           abs(np.mean(df[0]) - 1.0623) < 0.0001 and \
           abs(np.mean(df[1]) - 0.9461) < 0.0001

def test_DataFileList_gemini_IRAF_existing_1():

    dfl = DataFileList(fn_mefnodq)

    assert len(dfl) == 1 and \
           dfl[0].filename.prefix == 'eqbprg' and \
           dfl[0].filename.base == 'S20120827S0069' and \
           dfl[0].filename.suffix == ['_flat'] and \
           dfl[0].filename.ext == 'fits' and \
           dfl[0].filename.standard == True and \
           dfl[0].meta['OBJECT'] == 'GCALflat' and \
           len(dfl[0]) == 2 and \
           abs(np.mean(dfl[0][0]) - 1.0623) < 0.0001 and \
           abs(np.mean(dfl[0][1]) - 0.9461) < 0.0001

def test_DataFileList_replacing_data_1():

    # Replace the data & header from file with blank ones:
    dfl = DataFileList(filenames=fn_mefnodq, data=[], meta={})

    assert len(dfl) == 1 and len(dfl[0]) == 0 and len(dfl[0].meta) == 0

def test_DataFileList_len_mismatch_1():

    # Cannot have multiple data objects per filename:
    with pytest.raises(ValueError):
        dfl = DataFileList(filenames=fn_mefnodq, data=[DataFile(), DataFile()])

def test_DataFileList_broadcast_data_1():

    # But we can map the same dataset to multiple files:
    dfl = DataFileList([fn_mefnodq, 'test_name'],
                       data=DataFile(data=NDDataArray([1,2,3])),
                       mode='overwrite')

    # Produces 2 separate DataFiles referring to the same data array (as
    # long as the data aren't lazily-loaded separately).
    assert len(dfl) == 2 and dfl[0] is not dfl[1] \
        and len(dfl[0]) == len(dfl[1])\
        and [ndda.data is nddb.data for ndda, nddb in zip(dfl[0], dfl[1])]

def test_DataFileList_copy_self_1():

    dfl1 = DataFileList(fn_mefnodq)
    dfl2 = DataFileList(data=dfl1)

    assert dfl1 is not dfl2 and dfl1 == dfl2

def test_DataFileList_append_1():

    dfl = DataFileList(filenames=fn_mefnodq, mode='overwrite')
    dfl.append(DataFile('some_file', mode='new'))

    assert len(dfl) == 2 and str(dfl[0].filename) == fn_mefnodq \
        and str(dfl[1].filename) == 'some_file'

def test_DataFileList_nested_nddata_1():

    # Here the outer list maps to DataFiles and the inner one (if applicable)
    # to data extensions/groups within a file. The inner list can be omitted
    # where the file only contains a single NDData instance.
    dfl = DataFileList(filenames=['test_name_1', 'test_name_2'],
                       data=[[NDDataArray([1,2,3]), NDDataArray([4])], \
                             NDDataArray([5,6])], mode='new')

    assert len(dfl[0]) == 2 and len(dfl[1]) == 1

# Test our modified NDData init API & lazy loading behaviour:
def test_NDLater_1():

    ndd = NDLater(iomap=NDMapIO(fn_mefnodq, ident=2, data_idx=2))
    # For now, just check that the API works rather than trying to analyze
    # memory usage here (has been checked separately though).
    mean = np.mean(ndd.data)
    del ndd.data

    assert abs(mean - 0.9461) < 0.0001

max_stars_repo_name: jehturner/quadpype

max_issues_repo_name: jehturner/quadpype

max_forks_repo_name: jehturner/quadpype

reason: import numpy,from astropy