text:
"""Splits a multipage TIFF file of a FLIR IR camera into individual frames."""

import glob
import os
import csv
import json
import logging
from datetime import datetime
import cv2
import numpy as np
from tqdm import tqdm
import tifffile

from extractor.common import delete_output

logger = logging.getLogger(__name__)

def to_decimal_degrees(degrees_n, degrees_d, minutes_n, minutes_d,
    seconds_n, seconds_d):
    """Converts degrees, minutes and seconds into decimal degrees."""
    degrees = degrees_n / degrees_d
    minutes = minutes_n / minutes_d
    seconds = seconds_n / seconds_d
    deg_loc = degrees + (minutes/60) + (seconds/3600)
    return deg_loc

def exif_gps_to_degrees(gps_info):
    """Transforms EXIF GPS info into degrees north and east.

    Positive values correspond to north and east whereas negative
    values correspond to south and west.
    """
    latitude_deg = to_decimal_degrees(*gps_info['GPSLatitude'])
    longitude_deg = to_decimal_degrees(*gps_info['GPSLongitude'])
    if gps_info['GPSLatitudeRef'] == "S":
        latitude_deg *= -1.0
    if gps_info['GPSLongitudeRef'] == "W":
        longitude_deg *= -1.0
    return latitude_deg, longitude_deg

def get_gps_altitude(gps_info):
    """Obtain decimal representation of GPS altitude. Returns 0.0
    if EXIF GPS altitude tag is not available."""
    try:
        altitude = gps_info['GPSAltitude']
    except KeyError:
        altitude = 0.0
    else:
        altitude = altitude[0] / altitude[1]
    return altitude

def get_num_rgb_frames(rgb_files):
    """Return the number of frames in the provided videos.

    Args:
        rgb_files (`list` of `str`): List of video files names.

    Returns:
        Number of video frames (`int`).
    """
    n_rgb = 0
    for rgb_file in rgb_files:
        cap = cap = cv2.VideoCapture(rgb_file)
        n_rgb += cap.get(cv2.CAP_PROP_FRAME_COUNT)
    return int(n_rgb)

def get_num_ir_frames(tiff_files):
    """Return the number of frames in the provided videos (TIFF stacks)."""
    n_ir = 0
    for tiff_file in tiff_files:
        with tifffile.TiffFile(tiff_file) as tif:
            for page in tif.pages:
                n_ir += 1
    return int(n_ir)

def get_ir_frame_number(rgb_idx, n_ir, n_rgb):
    """Returns index of IR frame corresponding to the RGB frame idx."""
    ir_idx = round(n_ir*float(rgb_idx)/n_rgb)
    return ir_idx

def rotate(frame, rotation):
    if rotation == "90_deg_cw":
        frame = cv2.rotate(frame, cv2.ROTATE_90_CLOCKWISE)
    elif rotation == "180_deg_cw":
        frame = cv2.rotate(frame, cv2.ROTATE_180)
    elif rotation == "270_deg_cw":
        frame = cv2.rotate(frame, cv2.ROTATE_90_COUNTERCLOCKWISE)
    else:
        raise ValueError("Invalid value for rotation")
    return frame

def run(video_dir, output_dir, ir_file_extension=None, rgb_file_extension=None,
extract_timestamps=True,
    extract_gps=True, extract_gps_altitude=False, sync_rgb=True,
    rotate_frames=None):

    delete_output(output_dir)
    for dirname in ["radiometric", "gps"]:
        os.makedirs(os.path.join(output_dir, dirname), exist_ok=True)

    tiff_files = sorted(glob.glob(os.path.join(video_dir, "*.{}".format(ir_file_extension))))
    n_ir = get_num_ir_frames(tiff_files)
    logger.info("Found {} TIFF videos with {} frames for splitting".format(
        len(tiff_files), n_ir))

    if sync_rgb:
        os.makedirs(os.path.join(output_dir, "rgb"), exist_ok=True)
        rgb_files = sorted(glob.glob(os.path.join(video_dir, "*.{}".format(rgb_file_extension))))
        n_rgb = get_num_rgb_frames(rgb_files)
        assert get_ir_frame_number(n_rgb, n_ir, n_rgb) == n_ir
        logger.info("Found {} RGB videos with {} frames for splitting".format(
            len(rgb_files), n_rgb))

    frame_idx = 0
    gps = []
    timestamps = []
    for i, tiff_file in enumerate(tiff_files):
        logger.info("Splitting TIFF file {} of {}".format(i+1, len(tiff_files)))

        with tifffile.TiffFile(tiff_file) as tif:
            for page in tqdm(tif.pages, total=len(tif.pages)):
                image_radiometric = page.asarray().astype(np.uint16)  # BUG: for Optris camera conversion to int is detrimental to accuracy as raw values in TIFF are float
                radiometric_file = os.path.join(
                    output_dir, "radiometric", "frame_{:06d}.tiff".format(
                    frame_idx))
                if rotate_frames:
                    image_radiometric = rotate(image_radiometric, rotate_frames)
                cv2.imwrite(radiometric_file, image_radiometric)

                if extract_timestamps:
                    try:
                        timestamp = page.tags["ExifTag"].value['DateTimeOriginal']
                        timestamp_subsec = page.tags["ExifTag"].value['SubsecTimeOriginal']
                    except KeyError:
                        pass
                    else:
                        timestamp = "{}.{:06.0f}".format(timestamp, float(timestamp_subsec)*10000)
                        timestamp_iso = datetime.strptime(
                            timestamp, '%Y:%m:%d %H:%M:%S.%f').isoformat(timespec='microseconds')
                        timestamps.append(timestamp_iso)

                # extract GPS position
                if extract_gps:
                    try:
                        gps_info = page.tags["GPSTag"].value
                    except KeyError:
                        pass
                    else:
                        deg_lat, deg_long = exif_gps_to_degrees(gps_info)

                        if extract_gps_altitude:
                            alt = get_gps_altitude(gps_info)
                            gps.append((deg_long, deg_lat, alt))
                        else:
                            gps.append((deg_long, deg_lat))

                frame_idx += 1

    # synchronize RGB videos
    if sync_rgb:
        rgb_frame_idx = 0
        last_frame_idx = None
        for i, rgb_file in enumerate(rgb_files):
            logger.info("Splitting RGB file {} of {}".format(i+1, len(rgb_files)))
            cap = cv2.VideoCapture(rgb_file)
            for _ in tqdm(range(int(cap.get(cv2.CAP_PROP_FRAME_COUNT)))):
                res, frame = cap.read()
                if res:
                    frame_idx = get_ir_frame_number(rgb_frame_idx, n_ir, n_rgb)
                    if last_frame_idx is None or frame_idx != last_frame_idx:
                        out_path = os.path.join(output_dir,
                            "rgb", "frame_{:06d}.jpg".format(frame_idx))
                        if rotate_frames:
                            frame = rotate(frame, rotate_frames)
                        cv2.imwrite(
                            out_path, frame, [cv2.IMWRITE_JPEG_QUALITY, 100])
                    last_frame_idx = frame_idx

                    rgb_frame_idx += 1

    # store extracted timestamps to disk
    if extract_timestamps and len(timestamps) > 0:
        with open(os.path.join(
                output_dir, "timestamps.csv"), "w", newline="") as csvfile:
            writer = csv.writer(csvfile, delimiter=",")
            for timestamp in timestamps:
                writer.writerow([timestamp])

    # store extracted GPS positions to disk
    if extract_gps and len(gps) > 0:

        # save GPS trajectory to JSON
        json.dump(gps, open(os.path.join(
            output_dir, "gps", "gps.json"), "w"))

max_stars_repo_name: LukasBommes/PV-Drone-Inspect

max_issues_repo_name: LukasBommes/PV-Drone-Inspect

max_forks_repo_name: LukasBommes/PV-Hawk


text:
import numpy as np

from copy import deepcopy
from dnnbrain.io import fileio as fio
from dnnbrain.dnn.base import dnn_mask, dnn_fe, array_statistic
from dnnbrain.dnn.base import UnivariatePredictionModel, MultivariatePredictionModel
from dnnbrain.brain.algo import convolve_hrf

class Stimulus:
    """
    Store and handle stimulus-related information
    """
    def __init__(self, header=None, data=None):
        """
        Parameter:
        ---------
        header[dict]: meta-information of stimuli
        data[dict]: stimulus/behavior data
            Its values are arrays with shape as (n_stim,).
            It must have the key 'stimID'.
        """
        if header is None:
            self.header = dict()
        else:
            assert isinstance(header, dict), "header must be dict"
            self.header = header

        if data is None:
            self._data = dict()
        else:
            n_stim = len(data['stimID'])
            for v in data.values():
                assert isinstance(v, np.ndarray), "data's value must be an array."
                assert v.shape == (n_stim,), "data's value must be an array with shape as (n_stim,)"
            self._data = data

    def load(self, fname):
        """
        Load stimulus-related information

        Parameter:
        ---------
        fname[str]: file name with suffix as .stim.csv
        """
        stim_file = fio.StimulusFile(fname)
        stimuli = stim_file.read()
        self._data = stimuli.pop('data')
        self.header = stimuli

    def save(self, fname):
        """
        Save stimulus-related information

        Parameter:
        ---------
        fname[str]: file name with suffix as .stim.csv
        """
        stim_file = fio.StimulusFile(fname)
        header = self.header.copy()
        stim_file.write(header.pop('type'), header.pop('path'),
                        self._data, **header)

    def get(self, item):
        """
        Get a column of data according to the item

        Parameter:
        ---------
        item[str]: item name of each column

        Return:
        ------
        col[array]: a column of data
        """
        return self._data[item]

    def set(self, item, value):
        """
        Set a column of data according to the item

        Parameters:
        ----------
        item[str]: item name of the column
        value[array_like]: an array_like data with shape as (n_stim,)
        """
        self._data[item] = np.asarray(value)

    def delete(self, item):
        """
        Delete a column of data according to item

        Parameter:
        ---------
        item[str]: item name of each column
        """
        self._data.pop(item)

    @property
    def items(self):
        return list(self._data.keys())

    def __len__(self):
        """
        the length of the Stimulus object

        Return:
        ------
            [int]: the number of stimulus IDs
        """
        return len(self._data['stimID'])

    def __getitem__(self, indices):
        """
        Get part of the Stimulus object by imitating 2D array's subscript index

        Parameter:
        ---------
        indices[int|list|tuple|slice]: subscript indices

        Return:
        ------
        stim[Stimulus]: a part of the self.
        """
        # parse subscript indices
        if isinstance(indices, int):
            # regard it as row index
            # get all columns
            rows = [indices]
            cols = self.items
        elif isinstance(indices, (slice, list)):
            # regard it all as row indices
            # get all columns
            rows = indices
            cols = self.items
        elif isinstance(indices, tuple):
            if len(indices) == 0:
                # get all rows and columns
                rows = slice(None, None, None)
                cols = self.items
            elif len(indices) == 1:
                # regard the only element as row indices
                # get all columns
                if isinstance(indices[0], int):
                    # regard it as row index
                    rows = [indices[0]]
                elif isinstance(indices[0], (slice, list)):
                    # regard it all as row indices
                    rows = indices[0]
                else:
                    raise IndexError("only integer, slices (`:`), list are valid row indices")
                cols = self.items
            elif len(indices) == 2:
                # regard the first element as row indices
                # regard the second element as column indices
                rows, cols = indices
                if isinstance(rows, int):
                    # regard it as row index
                    rows = [rows]
                elif isinstance(rows, (slice, list)):
                    # regard it all as row indices
                    pass
                else:
                    raise IndexError("only integer, slices (`:`), list are valid row indices")

                if isinstance(cols, int):
                    # get a column according to an integer
                    cols = [self.items[cols]]
                elif isinstance(cols, str):
                    # get a column according to an string
                    cols = [cols]
                elif isinstance(cols, list):
                    if np.all([isinstance(i, int) for i in cols]):
                        # get columns according to a list of integers
                        cols = [self.items[i] for i in cols]
                    elif np.all([isinstance(i, str) for i in cols]):
                        # get columns according to a list of strings
                        pass
                    else:
                        raise IndexError("only integer [list], string [list] and slices (`:`) "
                                         "are valid column indices")
                elif isinstance(cols, slice):
                    # get columns according to a slice
                    cols = self.items[cols]
                else:
                    raise IndexError("only integer [list], string [list] and slices (`:`) "
                                     "are valid column indices")
            else:
                raise IndexError("This is a 2D data, "
                                 "and can't support more than 3 subscript indices!")
        else:
            raise IndexError("only integer, slices (`:`), list and tuple are valid indices")

        # get part of self
        stim = Stimulus()
        stim.header = deepcopy(self.header)
        for item in cols:
            stim.set(item, self.get(item)[rows])

        return stim

class Activation:
    """DNN activation"""

    def __init__(self, layer=None, value=None):
        """
        Parameters:
        ----------
        layer[str]: layer name
        value[array]: 4D DNN activation array with shape (n_stim, n_chn, n_row, n_col)
            It will be ignored if layer is None.
        """
        if layer is None:
            self._activation = dict()
        else:
            assert value is not None, "value can't be None if layer is not None."
            self.set(layer, value)

    def load(self, fname, dmask=None):
        """
        Load DNN activation

        Parameters:
        ----------
        fname[str]: DNN activation file
        dmask[Mask]: The mask includes layers/channels/rows/columns of interest.
        """
        if dmask is not None:
            dmask_dict = dict()
            for layer in dmask.layers:
                dmask_dict[layer] = dmask.get(layer)
        else:
            dmask_dict = None

        self._activation = fio.ActivationFile(fname).read(dmask_dict)

    def save(self, fname):
        """
        Save DNN activation

        Parameter:
        ---------
        fname[str]: output file of DNN activation
        """
        fio.ActivationFile(fname).write(self._activation)

    def get(self, layer):
        """
        Get DNN activation

        Parameter:
        ---------
        layer[str]: layer name

        Return:
        ------
            [array]: (n_stim, n_chn, n_row, n_col) array
        """
        return self._activation[layer]

    def set(self, layer, value):
        """
        Set DNN activation

        Parameters:
        ----------
        layer[str]: layer name
        value[array]: 4D DNN activation array with shape (n_stim, n_chn, n_row, n_col)
        """
        self._activation[layer] = value

    def delete(self, layer):
        """
        Delete DNN activation

        Parameter:
        ---------
        layer[str]: layer name
        """
        self._activation.pop(layer)

    def concatenate(self, activations):
        """
        Concatenate activations from different batches of stimuli

        Parameter:
        ---------
        activations[list]: a list of Activation objects

        Return:
        ------
        activation[Activation]: DNN activation
        """
        # check availability
        for i, v in enumerate(activations, 1):
            if not isinstance(v, Activation):
                raise TypeError('All elements in activations must be instances of Activation!')
            if sorted(self.layers) != sorted(v.layers):
                raise ValueError("The element{}'s layers mismatch with self!".format(i))

        # concatenate
        activation = Activation()
        for layer in self.layers:
            # concatenate activation
            data = [v.get(layer) for v in activations]
            data.insert(0, self.get(layer))
            data = np.concatenate(data)
            activation.set(layer, data)

        return activation

    @property
    def layers(self):
        return list(self._activation.keys())

    def mask(self, dmask):
        """
        Mask DNN activation

        Parameter:
        ---------
        dmask[Mask]: The mask includes layers/channels/rows/columns of interest.

        Return:
        ------
        activation[Activation]: DNN activation
        """
        activation = Activation()
        for layer in dmask.layers:
            mask = dmask.get(layer)
            data = dnn_mask(self.get(layer), mask.get('chn'),
                            mask.get('row'), mask.get('col'))
            activation.set(layer, data)

        return activation

    def pool(self, method):
        """
        Pooling DNN activation for each channel

        Parameter:
        ---------
        method[str]: pooling method, choices=(max, mean, median, L1, L2)

        Return:
        ------
        activation[Activation]: DNN activation
        """
        activation = Activation()
        for layer, data in self._activation.items():
            data = array_statistic(data, method, (2, 3), True)
            activation.set(layer, data)

        return activation

    def fe(self, method, n_feat, axis=None):
        """
        Extract features of DNN activation

        Parameters:
        ----------
        method[str]: feature extraction method, choices=(pca, hist, psd)
            pca: use n_feat principal components as features
            hist: use histogram of activation as features
                Note: n_feat equal-width bins in the given range will be used!
                psd: use power spectral density as features
        n_feat[int]: The number of features to extract
        axis{str}: axis for feature extraction, choices=(chn, row_col)

        Return:
        ------
        activation[Activation]: DNN activation
        """
        activation = Activation()
        for layer, data in self._activation.items():
            data = dnn_fe(data, method, n_feat, axis)
            activation.set(layer, data)

        return activation

    def convolve_hrf(self, onsets, durations, n_vol, tr, ops=100):
        """
        Convolve DNN activation with HRF and align with the timeline of BOLD signal

        Parameters:
        ----------
        onsets[array_like]: in sec. size = n_event
        durations[array_like]: in sec. size = n_event
        n_vol[int]: the number of volumes of BOLD signal
        tr[float]: repeat time in second
        ops[int]: oversampling number per second

        Return:
        ------
        activation[Activation]: DNN activation
        """
        activation = Activation()
        for layer, data in self._activation.items():
            n_stim, n_chn, n_row, n_col = data.shape
            data = convolve_hrf(data.reshape(n_stim, -1), onsets, durations,
                                n_vol, tr, ops)
            data = data.reshape(n_vol, n_chn, n_row, n_col)
            activation.set(layer, data)

        return activation

    def _check_arithmetic(self, other):
        """
        Check availability of the arithmetic operation for self

        Parameter:
        ---------
        other[Activation]: DNN activation
        """
        if not isinstance(other, Activation):
            raise TypeError("unsupported operand type(s): "
                            "'{0}' and '{1}'".format(type(self), type(other)))
        assert sorted(self.layers) == sorted(other.layers), \
            "The two object's layers mismatch!"
        for layer in self.layers:
            assert self.get(layer).shape == other.get(layer).shape, \
                "{}'s activation shape mismatch!".format(layer)

    def __add__(self, other):
        """
        Define addition operation

        Parameter:
        ---------
        other[Activation]: DNN activation

        Return:
        ------
        activation[Activation]: DNN activation
        """
        self._check_arithmetic(other)

        activation = Activation()
        for layer in self.layers:
            data = self.get(layer) + other.get(layer)
            activation.set(layer, data)

        return activation

    def __sub__(self, other):
        """
        Define subtraction operation

        Parameter:
        ---------
        other[Activation]: DNN activation

        Return:
        ------
        activation[Activation]: DNN activation
        """
        self._check_arithmetic(other)

        activation = Activation()
        for layer in self.layers:
            data = self.get(layer) - other.get(layer)
            activation.set(layer, data)

        return activation

    def __mul__(self, other):
        """
        Define multiplication operation

        Parameter:
        ---------
        other[Activation]: DNN activation

        Return:
        ------
        activation[Activation]: DNN activation
        """
        self._check_arithmetic(other)

        activation = Activation()
        for layer in self.layers:
            data = self.get(layer) * other.get(layer)
            activation.set(layer, data)

        return activation

    def __truediv__(self, other):
        """
        Define true division operation

        Parameter:
        ---------
        other[Activation]: DNN activation

        Return:
        ------
        activation[Activation]: DNN activation
        """
        self._check_arithmetic(other)

        activation = Activation()
        for layer in self.layers:
            data = self.get(layer) / other.get(layer)
            activation.set(layer, data)

        return activation

    def __getitem__(self, indices):
        """
        Get part of Activation along stimulus axis

        Parameter:
        ---------
        indices[int|list|slice]: indices of stimulus axis

        Return:
        ------
        activation[Activation]: DNN activation
        """
        if isinstance(indices, int):
            indices = [indices]
        elif isinstance(indices, (list, slice)):
            pass
        else:
            raise IndexError("only integer, slices (`:`), and list are valid indices")

        activation = Activation()
        for layer in self.layers:
            data = self.get(layer)[indices]
            activation.set(layer, data)

        return activation

class Mask:
    """DNN mask"""

    def __init__(self, layer=None, channels='all', rows='all', columns='all'):
        """
        Parameter:
        ---------
        layer[str]: layer name
            If layer is None, other parameters will be ignored.
        channels[str|list]: channels of interest.
            If is str, it must be 'all' which means all channels.
            If is list, its elements are serial numbers of channels.
        rows[str|list]: rows of interest.
            If is str, it must be 'all' which means all rows.
            If is list, its elements are serial numbers of rows.
        columns[str|list]: columns of interest.
            If is str, it must be 'all' which means all columns.
            If is list, its elements are serial numbers of columns.
        """
        self._dmask = dict()
        if layer is not None:
            self.set(layer, channels=channels, rows=rows, columns=columns)

    def load(self, fname):
        """
        Load DNN mask, the whole mask will be overrode.

        Parameter:
        ---------
        fname[str]: DNN mask file
        """
        self._dmask = fio.MaskFile(fname).read()

    def save(self, fname):
        """
        Save DNN mask

        Parameter:
        ---------
        fname[str]: output file name of DNN mask
        """
        fio.MaskFile(fname).write(self._dmask)

    def get(self, layer):
        """
        Get mask of a layer

        Parameter:
        ---------
        layer[str]: layer name

        Return:
        ------
            [dict]: layer mask
        """
        return self._dmask[layer]

    def set(self, layer, **kwargs):
        """
        Set DNN mask

        Parameters:
        ----------
        layer[str]: layer name
            If layer is new, its corresponding mask value will be initialized as 'all'.
        kwargs[dict]: keyword arguments
            Only three keywords ('channels', 'rows', 'columns') are valid.
            channels[str|list]: channels of interest.
                If is str, it must be 'all' which means all channels.
                If is list, its elements are serial numbers of channels.
            rows[str|list]: rows of interest.
                If is str, it must be 'all' which means all rows.
                If is list, its elements are serial numbers of rows.
            columns[str|list]: columns of interest.
                If is str, it must be 'all' which means all columns.
                If is list, its elements are serial numbers of columns.
        """
        # assertion
        for k, v in kwargs.items():
            assert k in ('channels', 'rows', 'columns'), \
                "keyword must be one of ('channels', 'rows', 'columns')"
            assert v == 'all' or isinstance(v, list), \
                f"{k} must be 'all' or list of non-negative integers"

        if layer not in self._dmask:
            self._dmask[layer] = {'chn': 'all', 'row': 'all', 'col': 'all'}
        if 'channels' in kwargs:
            self._dmask[layer]['chn'] = kwargs['channels']
        if 'rows' in kwargs:
            self._dmask[layer]['row'] = kwargs['rows']
        if 'columns' in kwargs:
            self._dmask[layer]['col'] = kwargs['columns']

    def copy(self):
        """
        Make a copy of the DNN mask

        Return:
        ------
        dmask[Mask]: The mask includes layers/channels/rows/columns of interest.
        """
        dmask = Mask()
        dmask._dmask = deepcopy(self._dmask)

        return dmask

    def delete(self, layer):
        """
        Delete a layer

        Parameter:
        ---------
        layer[str]: layer name
        """
        self._dmask.pop(layer)

    def clear(self):
        """
        Empty the DNN mask
        """
        self._dmask.clear()

    @property
    def layers(self):
        return list(self._dmask.keys())

class RDM:
    """representation distance matrix"""

    def __init__(self):
        self.rdm_type = None
        self._rdm_dict = dict()

    def load(self, fname):
        """
        Parameter:
        ---------
        fname[str]: file name with suffix as .rdm.h5
        """
        self.rdm_type, self._rdm_dict = fio.RdmFile(fname).read()

    def save(self, fname):
        """
        Parameter:
        ---------
        fname[str]: file name with suffix as .rdm.h5
        """
        fio.RdmFile(fname).write(self.rdm_type, self._rdm_dict)

    def get(self, key, triu=False):
        """
        Get RDM according its key.

        Parameters:
        ----------
        key[str]: the key of the RDM
        triu[bool]:
            If True, get RDM as the upper triangle vector.
            If False, get RDM as the square matrix.

        Return:
        ------
        rdm_arr[ndarray]: RDM
            If rdm_type is bRDM:
                Its shape is ((n_item^2-n_item)/2,) or (n_item, n_item).
            If rdm_type is dRDM:
                Its shape is (n_iter, (n_item^2-n_item)/2) or (n_iter, n_item, n_item).
        """
        rdm_arr = self._rdm_dict[key]
        if not triu:
            idx_arr = np.tri(self.n_item, k=-1, dtype=np.bool).T
            if self.rdm_type == 'bRDM':
                rdm_tmp = np.zeros((self.n_item, self.n_item))
                rdm_tmp[idx_arr] = rdm_arr
            elif self.rdm_type == 'dRDM':
                rdm_tmp = np.zeros((rdm_arr.shape[0], self.n_item, self.n_item))
                rdm_tmp[:, idx_arr] = rdm_arr
            else:
                raise TypeError("Set rdm_type to bRDM or dRDM at first!")
            rdm_arr = rdm_tmp

        return rdm_arr

    def set(self, key, rdm_arr, triu=False):
        """
        Set RDM according its key.

        Parameters:
        ----------
        key[str]: the key of the RDM
        rdm_arr[ndarray]: RDM
            If rdm_type is bRDM:
                Its shape is ((n_item^2-n_item)/2,) or (n_item, n_item).
            If rdm_type is dRDM:
                Its shape is (n_iter, (n_item^2-n_item)/2) or (n_iter, n_item, n_item).
        triu[bool]:
            If True, RDM will be regarded as the upper triangle vector.
            If False, RDM will be regarded as the square matrix.
        """
        if self.rdm_type == 'bRDM':
            if triu:
                assert rdm_arr.ndim == 1, \
                    "If triu is True, bRDM's shape must be ((n_item^2-n_item)/2,)."
                self._rdm_dict[key] = rdm_arr
            else:
                assert rdm_arr.ndim == 2 and rdm_arr.shape[0] == rdm_arr.shape[1], \
                    "If triu is False, bRDM's shape must be (n_item, n_item)."
                self._rdm_dict[key] = rdm_arr[np.tri(rdm_arr.shape[0], k=-1, dtype=np.bool).T]
        elif self.rdm_type == 'dRDM':
            if triu:
                assert rdm_arr.ndim == 2, \
                    "If triu is True, dRDM's shape must be (n_iter, (n_item^2-n_item)/2)."
                self._rdm_dict[key] = rdm_arr
            else:
                assert rdm_arr.ndim == 3 and rdm_arr.shape[1] == rdm_arr.shape[2], \
                    "If triu is False, dRDM's shape must be (n_iter, n_item, n_item)."
                self._rdm_dict[key] = rdm_arr[:, np.tri(rdm_arr.shape[1], k=-1, dtype=np.bool).T]
        else:
            raise TypeError("Set rdm_type to bRDM or dRDM at first!")

    @property
    def keys(self):
        """
        Get keys of RDM dictionary

        Return:
        ------
        keys[list]: the list of keys
        """
        if self._rdm_dict:
            keys = list(self._rdm_dict.keys())
        else:
            raise ValueError("The RDM dictionary is empty.")

        return keys

    @property
    def n_item(self):
        """
        Get the number of items of RDM

        Return:
        ------
        n_item[int]: the number of items
        """
        k = self.keys[0]
        if self.rdm_type == 'bRDM':
            n = self._rdm_dict[k].shape[0]
        elif self.rdm_type == 'dRDM':
            n = self._rdm_dict[k].shape[1]
        else:
            raise TypeError("Set rdm_type to bRDM or dRDM at first!")
        n_item = int((1 + np.sqrt(1+8*n)) / 2)

        return n_item

class DnnProbe:
    """
    Decode DNN activation to behavior data. As a result,
    probe the ability of DNN activation to predict the behavior.
    """
    def __init__(self, dnn_activ=None, model_type=None, model_name=None, cv=3):
        """
        Parameters:
        ----------
        dnn_activ[Activation]: DNN activation
        model_type[str]: choices=(uv, mv)
            'uv': univariate prediction model
            'mv': multivariate prediction model
        model_name[str]: name of a model used to do prediction
            If is 'corr', it just uses correlation rather than prediction.
                And the model_type must be 'uv'.
        cv[int]: cross validation fold number
        """
        self.set(dnn_activ, model_type, model_name, cv)

    def set(self, dnn_activ=None, model_type=None, model_name=None, cv=None):
        """
        Set some attributes

        Parameters:
        ----------
        dnn_activ[Activation]: DNN activation
        model_type[str]: choices=(uv, mv)
            'uv': univariate prediction model
            'mv': multivariate prediction model
        model_name[str]: name of a model used to do prediction
            If is 'corr', it just uses correlation rather than prediction.
                And the model_type must be 'uv'.
        cv[int]: cross validation fold number
        """
        if dnn_activ is not None:
            self.dnn_activ = dnn_activ

        if model_type is None:
            pass
        elif model_type == 'uv':
            self.model = UnivariatePredictionModel()
        elif model_type == 'mv':
            self.model = MultivariatePredictionModel()
        else:
            raise ValueError('model_type must be one of the (uv, mv).')

        if model_name is not None:
            if not hasattr(self, 'model'):
                raise RuntimeError('You have to set model_type first!')
            self.model.set(model_name)

        if cv is not None:
            if not hasattr(self, 'model'):
                raise RuntimeError('You have to set model_type first!')
            self.model.set(cv=cv)

    def probe(self, beh_data, iter_axis=None):
        """
        Probe the ability of DNN activation to predict the behavior.

        Parameters:
        ----------
        beh_data[ndarray]: behavior data with shape as (n_stim, n_beh)
        iter_axis[str]: iterate along the specified axis
            ---for uv---
            channel: Summarize the maximal prediction score for each channel.
            row_col: Summarize the maximal prediction score for each position (row_idx, col_idx).
            default: Summarize the maximal prediction score for the whole layer.
            ---for mv---
            channel: Do multivariate prediction using all units in each channel.
            row_col: Do multivariate prediction using all units in each position (row_idx, col_idx).
            default: Do multivariate prediction using all units in the whole layer.

        Return:
        ------
        probe_dict[dict]:
            ---for uv---
            layer:
                max_score[ndarray]: shape=(n_iter, n_beh)
                    max scores at each iteration
                max_loc[ndarray]: shape=(n_iter, n_beh, 3)
                    max locations of the max scores, the size 3 of the third dimension means
                    channel, row and column locations respectively.
                max_model[ndarray]: shape=(n_iter, n_beh)
                    fitted models of the max scores
                    Note: only exists when model is classifier or regressor
                score[ndarray]: shape=(n_iter, n_beh, cv)
                    The third dimension means scores of each cross validation folds of the max scores
                    Note: only exists when model is classifier or regressor
                conf_m[ndarray]: shape=(n_iter, n_beh, cv)
                    The third dimension means confusion matrices (n_label, n_label) of
                    each cross validation folds of the max scores
                    Note: only exists when model is classifier

            ---for mv---
            layer:
                score[ndarray]: shape=(n_iter, n_beh, cv)
                    The third dimension means scores of each cross validation folds
                    at each iteration and behavior
                model[ndarray]: shape=(n_iter, n_beh)
                    Each element is a model fitted at the corresponding iteration and behavior.
                conf_m[ndarray]: shape=(n_iter, n_beh, cv)
                    The third dimension means confusion matrices (n_label, n_label) of
                    each cross validation folds, at each iteration and behavior.
                    Note: only exists when model is classifier
        """
        _, n_beh = beh_data.shape

        probe_dict = dict()
        for layer in self.dnn_activ.layers:
            # get DNN activation and reshape it to 3D
            activ = self.dnn_activ.get(layer)
            n_stim, n_chn, n_row, n_col = activ.shape
            n_row_col = n_row * n_col
            activ = activ.reshape((n_stim, n_chn, n_row_col))

            # transpose axis to make activ's shape as (n_stimulus, n_iterator, n_element)
            if iter_axis is None:
                activ = activ.reshape((n_stim, 1, -1))
            elif iter_axis == 'row_col':
                activ = activ.transpose((0, 2, 1))
            elif iter_axis == 'channel':
                pass
            else:
                raise ValueError("Unsupported iter_axis:", iter_axis)
            n_stim, n_iter, n_elem = activ.shape

            # start probing
            if isinstance(self.model, UnivariatePredictionModel):
                # prepare layer dict
                probe_dict[layer] = {
                    'max_score': np.zeros((n_iter, n_beh)),
                    'max_loc': np.zeros((n_iter, n_beh, 3), dtype=np.int),
                    'max_model': np.zeros((n_iter, n_beh), dtype=np.object),
                    'score': np.zeros((n_iter, n_beh, self.model.cv)),
                    'conf_m': np.zeros((n_iter, n_beh, self.model.cv), dtype=np.object)
                }
                # start iteration
                for iter_idx in range(n_iter):
                    data = self.model.predict(activ[:, iter_idx, :], beh_data)
                    for k, v in data.items():
                        if k == 'max_loc':
                            if iter_axis is None:
                                chn_idx = v // n_row_col
                                row_idx = v % n_row_col // n_col
                                col_idx = v % n_row_col % n_col
                            elif iter_axis == 'channel':
                                chn_idx = iter_idx
                                row_idx = v // n_col
                                col_idx = v % n_col
                            else:
                                chn_idx = v
                                row_idx = iter_idx // n_col
                                col_idx = iter_idx % n_col
                            probe_dict[layer][k][iter_idx, :, 0] = chn_idx + 1
                            probe_dict[layer][k][iter_idx, :, 1] = row_idx + 1
                            probe_dict[layer][k][iter_idx, :, 2] = col_idx + 1
                        else:
                            probe_dict[layer][k][iter_idx] = v
                    print('Layer-{} iter-{}/{}'.format(layer, iter_idx+1, n_iter))
                # clear layer dict
                if self.model.model_type == 'corr':
                    probe_dict[layer].pop('max_model')
                    probe_dict[layer].pop('score')
                    probe_dict[layer].pop('conf_m')
                elif self.model.model_type == 'regressor':
                    probe_dict[layer].pop('conf_m')
            else:
                # prepare layer dict
                probe_dict[layer] = {
                    'score': np.zeros((n_iter, n_beh, self.model.cv)),
                    'model': np.zeros((n_iter, n_beh), dtype=np.object),
                    'conf_m': np.zeros((n_iter, n_beh, self.model.cv), dtype=np.object)
                }
                # start iteration
                for iter_idx in range(n_iter):
                    data = self.model.predict(activ[:, iter_idx, :], beh_data)
                    for k, v in data.items():
                        probe_dict[layer][k][iter_idx] = v

                    print('Layer-{} iter-{}/{}'.format(layer, iter_idx+1, n_iter))

        return probe_dict

ext: py

max_stars_repo_name: HoshinoKoji/dnnbrain

max_issues_repo_name: HoshinoKoji/dnnbrain

max_forks_repo_name: HoshinoKoji/dnnbrain

reason: import numpy


text:
# Preppin' Data 2021 Week 15
import pandas as pd
import numpy as np

# Load data
menu = pd.read_excel('unprepped_data\\PD 2021 Wk 15 Input - Menu and Orders.xlsx', engine='openpyxl', sheet_name = 'MENU')
order = pd.read_excel('unprepped_data\\PD 2021 Wk 15 Input - Menu and Orders.xlsx', engine='openpyxl', sheet_name = 'Order')

# Modify the structure of the Menu table so we can have one column for:
#  - the Type (pizza, pasta, house plate),
#  - the name of the plate,
#  - ID, and Price

# reduce to individual data frames
pizza = menu.loc[:, 'Pizza':'Pizza ID']
pasta = menu.loc[:, 'Pasta':'Pasta ID']
house_plates = menu.loc[:, 'House Plates':'House Plates ID']

# rename all columns
columns = ['Plate','Price','ID']

pizza.columns = columns
pasta.columns = columns
house_plates.columns = columns

# add food type
pizza['Type'] = 'Pizza'
pasta['Type'] = 'Pasta'
house_plates['Type'] = 'House Plates'

# concat all data frames and remove NaNs
menu_df = pd.concat([pizza,pasta,house_plates])
menu_df = menu_df[menu_df.ID.notnull()]

# Modify the structure of the Orders table to have each item ID in a different row

order['ID'] = order.index

# split string into seperate columns
a = pd.DataFrame(order['Order'].str.split('-',expand=True), index=order['ID'])

a.columns = ['1','2','3','4']
a['ID'] = a.index

# seperate into 2 column dataframes
a1 = a[['ID','1']]
a2 = a[['ID','2']]
a3 = a[['ID','3']]
a4 = a[['ID','4']]

 # rename all columns
a1.columns = ['ID','Order ID']
a2.columns = a1.columns
a3.columns = a1.columns
a4.columns = a1.columns

# stack dataframes and remove null entries
order_ids = pd.concat([a1,a2,a3,a4])
order_ids = order_ids[order_ids['Order ID'].notnull()]
del order_ids['ID']

# join back to original orders table and delete unused columns
order_df = pd.merge(order,order_ids, on='ID', how = 'left')
order_df['Order ID'] = np.where(order_df['Order ID'].notnull(), order_df['Order ID'], order_df['Order'])
del order_df['Order']
del order_df['ID']

# Join both tables

# convert join columns to int
order_df = order_df.astype({'Order ID': int})
menu_df = menu_df.astype({'ID': int})

# join menu with orders
takings_df = pd.merge(menu_df, order_df, left_on='ID', right_on='Order ID', how='left')

# On Monday's we offer a 50% discount on all items. Recalculate the prices to reflect this
# work out weekday and apply discount
takings_df['weekday'] = takings_df['Order Date'].dt.dayofweek
# Note Mondays are denoted as 0
takings_df['Price Paid'] = np.where(takings_df['weekday'] == 0 , takings_df['Price']/2, takings_df['Price'])

# convert weekday to day name
takings_df['weekday'] = np.select(
    [
        takings_df['weekday'] == 0,
        takings_df['weekday'] == 1,
        takings_df['weekday'] == 2,
        takings_df['weekday'] == 3,
        takings_df['weekday'] == 4,
        takings_df['weekday'] == 5,
        takings_df['weekday'] == 6
    ],
    [
        'Monday',
        'Tuesday',
        'Wednesday',
        'Thursday',
        'Friday',
        'Saturday',
        'Sunday'
    ],
    default='Unknown'
)
#print(takings_df)

# For Output 1, we want to calculate the total money for each day of the week
output_1 = takings_df[['weekday','Price Paid']]
output_1 = output_1.groupby(['weekday']).sum()

# For Output 2, we want to reward the customer who has made the most orders for their loyalty.
# Work out which customer has ordered the most single items.
output_2 = takings_df[['Customer Name','Order ID']]
output_2 = output_2.groupby(['Customer Name']).agg('count')

max_orders = output_2['Order ID'].max()
output_2 = output_2.loc[output_2['Order ID'] == max_orders]
output_2.columns = ['Count Items']

# Writing data to csv
output_1.to_csv('prepped_data\\PD 2021 Wk 15 Output - Weekday Sales.csv', index=True)
output_2.to_csv('prepped_data\\PD 2021 Wk 15 Output - Top Customer.csv', index=True)

print("data prepped!")

max_stars_repo_name: wjsutton/preppin-data

max_issues_repo_name: wjsutton/preppin-data

max_forks_repo_name: wjsutton/preppin-data


text:
"""Sample random variable."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import math
import random
import itertools

class SampleUniformDiscrete(object):
    def __init__(self, vals):
        """Sample uniform discrete values.

        Apply the Probability Integral Transform for uniform discrete r.v.,
          [x_0, ..., x_{n-1}] with probs 1/n:
        X = int(n*U), where U ~ Uniform(0, 1).

        Args:
          vals: A list. Values from which we want to sample.

        Time complexity: O(1).
        Space complexity: O(n), where n is the number of input values.
        """
        self.n = len(vals)
        self.vals = vals

    def sample(self):
        """Sample values in uniform probabilities.

        Time complexity: O(1).
        Space complexity: O(n).
        """
        # Sample a r.v. from Uniform(0, 1).
        u = random.uniform(0, 1)
        i = int(self.n * u)
        return self.vals[i]

class SampleNonUniformDiscrete(object):
    def __init__(self, vals, probs, n_bins=int(1e4)):
        """Sampling non-uniform discrete numbers.

        For r.v.'s with non-uniform probs, preprocess to "uniform" r.v.
        See https://www.keithschwarz.com/darts-dice-coins/
        - Specifically, preprocess [x_0, x_1, ...] with [p_0, p_1, ...] to
          repeated values [x_0, x_0, x_1, x_1, ...] with frequencies proportional to probs.
        - Apply the Probability Integral Transform for uniform discrete r.v.'s,
          [x_0, ..., x_m] with probs 1/m, where m = number of total repeated values:
          X = int(m*U), where U ~ Uniform(0, 1).

        Args:
          vals: A list. Values from which we want to sample.
          probs: A list. Sampling probabilities for values. Default: None.
          n_bins: A int. Bin number to duplicate values for uniform r.v. Default: 1000.
        """
        # Simulate non-uniform values by uniform values
        self.vals = vals
        self.probs = probs
        self.n_bins = n_bins
        self.is_constructed = False

    def construct(self):
        """Construct repeated values based probabilites for "uniform" r.v.

        Time complexity: O(nb),
          - n: number of values.
          - b: number of bins.
        Space complexity: O(nb).
        """
        # Construct only for the 1st call.
        if self.is_constructed:
            return None;

        # Construct repeated values with corresponding binned freqs.
        binned_freqs = [int(round(self.n_bins * p)) for p in self.probs]
        binned_vals_ls = [[v] * f for v, f in zip(self.vals, binned_freqs)]
        binned_vals = [v for ls in binned_vals_ls for v in ls]

        self.is_constructed = True
        self.n = len(binned_vals)
        self.binned_vals = binned_vals

    def sample(self):
        """Sample non-uniform values with "uniform" probabilities.

        Time complexity: O(1) for infinite calls.
        Space complexity: O(nb).
        """
        self.construct()

        # Sample a r.v. from Uniform(0, 1).
        u = random.uniform(0, 1)
        i = int(self.n * u)
        return self.binned_vals[i]

class SampleBiasedCoinWithFairCoin(object):
    def __init__(self, p):
        """Smple biased coin with p = 1/k by fair coin.

        First compute the number of fair coins we need to flip.
          n_cases = 1/p
          2^n_coins = n_cases => n_coins = ceiling(log2(n_cases))

        Note: This function only applies to rational p = 1/k.
        """
        if 1 / p != int(1 / p):
            raise ValueError("p must be an rational number 1 / k.")

        self.p = p
        self.is_constructed = False

    def construct(self):
        """Construct possible flips based on the number of coins needed.

        Time complexity: O(2^n),
          where n is the number of coins needed based on p.
        Space complexity: O(2^n).
        """
        if self.is_constructed:
            return None

        # Compute the number of coins needed and obtain possible flips.
        self.n_cases = int(1 / self.p)
        self.n_coins = int(math.ceil(math.log(self.n_cases, 2)))
        self.possible_flips = list(itertools.product([0, 1], repeat=self.n_coins))

        self.is_constructed = True

    def sample(self):
        """Sample biased coin.

        First flip number of coins, if the flipped coins is
        - the first case in possible flips, return success;
        - the (last) n_cases - 1 case, return failures;
        - the rest cases, retry.

        Time complexity: O(n).
        Space complexity: O(n).
        """
        self.construct()

        # Convert to tuple due to tuples in itertools's product ouput.
        flips = tuple([random.randint(0, 1) for _ in range(self.n_coins)])

        # Use only the 1st case and the last (n_cases - 1) as sampling basis.
        if flips == self.possible_flips[0]:
            return 1
        elif flips in set(self.possible_flips[-(self.n_cases-1):]):
            return 0
        else:
            return self.sample()

class SampleFairCoinWithBiasedCoin(object):
    def __init__(self, p):
        """Smple fair coin with biased coin having head probility p."""
        self.p = p

    def sample_biased(self):
        """Sample with biased coin."""
        u = random.uniform(0, 1)
        if u <= self.p:
            return 1
        else:
            return 0

    def sample(self):
        """Sample fair coin with biased coin.

        Note: For biased coin with head H probability p < 1/2,
        - The probability of (H, T): p*(1-p)
        - The probability of (T, H): (1-p)*p = that of (H, T).
        Thus we obtain "fair" probibilities for these two cases.
        For the rest cases, retry.

        Time complexity: O(1).
        Space complexity: O(1).
        """
        # Flip two biased coins.
        two_flips = [self.sample_biased() for _ in range(2)]

        # Use cases (1, 0) and (0, 1) for fair coin flipping.
        if two_flips == [1, 0]:
            return 1
        elif two_flips == [0, 1]:
            return 0
        else:
            return self.sample()

def main():
    import numpy as np

    n_exp = int(1e5)

    # Sample discrete random variable with equal probs.
    # Output: should be close to 0.5
    vals = [0, 1]
    sample_discrete = SampleUniformDiscrete(vals)

    samples = [None] * n_exp
    for i in range(n_exp):
        samples[i] = sample_discrete.sample()
    print(np.mean(samples))

    # Sample discrete random variable with unequal probs.
    # Output: should be close to 0.7
    vals = [0, 1, 2]
    probs = [0.5, 0.3, 0.2]
    sample_discrete = SampleNonUniformDiscrete(vals, probs)

    samples = [None] * n_exp
    for i in range(n_exp):
        samples[i] = sample_discrete.sample()
    print(np.mean(samples))

    # Sample biased coin with fair one.
    # Output: should be close to 1/4 = 0.25.
    p = 1 / 4
    sample_biased_coin = SampleBiasedCoinWithFairCoin(p)

    samples = [None] * n_exp
    for i in range(n_exp):
        samples[i] = sample_biased_coin.sample()
    print(np.mean(samples))

    # Sample fair coin with biased one.
    # Output: should be close to 0.5.
    p = 1 / 4
    sample_fair_coin = SampleFairCoinWithBiasedCoin(p)

    samples = [None] * n_exp
    for i in range(n_exp):
        samples[i] = sample_fair_coin.sample()
    print(np.mean(samples))

if __name__ == '__main__':
    main()


text:
import pandas as pd
import torch
from torch.utils import data
import numpy as np
import librosa
import os
from pathlib import Path

baum2ravdess = {0:0, 1:4, 2:1, 3:6, 4:5, 5:2, 6:3, 7:7}

class BAUM2_DATA(data.Dataset):
    def get_data(self):
        print("Loading dataset...")
        files = []
        table = pd.read_excel(self.excel_path)
        data = table[["Clip Name", "Emotion Code", "Audio useful?"]]
        data_len = len(data.values)
        for i, row in enumerate(data.values):
            print("{}/{}".format(i, data_len), end="\r")
            # load only the values with useful audio
            if row[2] == 1:
                rel_path = row[0].replace("_","/wav/")
                path = Path(self.data_path) / rel_path
                path = path.with_suffix(".wav")
                waveform, sr = librosa.load(path, sr=8000)

                # cut and calculate mel
                max_len = int(12.5 * 8000)   # in seconds * sample rate

                wav_cut, index = librosa.effects.trim(
                waveform, frame_length=2048, hop_length=512)
                if len(wav_cut) < max_len:
                    #padlen = (max_len - fcut.shape[0]) // 2
                    ff = np.pad(
                        wav_cut, [(0, max_len - wav_cut.shape[0]), ], mode='constant')
                    waveform = ff

                mel_spectogram = librosa.feature.melspectrogram(
                    y=waveform, sr=sr, n_mels=40)
                scaled_mel = librosa.power_to_db(mel_spectogram)
                files.append((scaled_mel, baum2ravdess[row[1]]))
        print("---DONE---")
        return files

    def __init__(self, data_path, excel_path):
        super(BAUM2_DATA, self).__init__()
        self.data_path = data_path
        self.excel_path = excel_path
        self.files = self.get_data()

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

    def __getitem__(self, index):
        return self.files[index]

if __name__ == '__main__':
    baum = BAUM2_DATA("BAUM2_dataset/data/", "BAUM2_dataset/Annotations.xlsx")
    print(baum.__getitem__(7))


text:
#!/usr/bin/python3

import gym
from gym import error, spaces, utils
from gym.utils import seeding
import numpy as np

class ArraySorting(gym.Env):
    """
    Description:
        Environment with a hidden state.
        The objective is to sort an array and to be sure that the array is indeed sorted.
        The agent only sees the last performed action, possibly with its boolean outcome.
    Observation: MultiDiscrete([4, size, size]),
        The first value encodes the last observation type,
        the other two are indices i,j of the array:
        Type   Meaning
          0  : initializaction of the array a[i] ... a[j]
          1  : swap of a[i], a[j]
          2  : a[i] < a[j]
          3  : a[j] > a[j]
    Actions:
        Type: MultiDiscrete(2, size, size)
        The first value encodes the action type second two indices i,j of the array:
        Type   Meaning
          0  : swap of a[i], a[j]
          1  : compare a[i], a[j]
    Reward and termination:
        Once the array is sorted and all pairs of consecutive elements were compared,
        the environment terminates with reward 1, otherwise, the reward is zero.
    Starting State:
        A random segment of the array is initialized to a random permutation.
        the actions are restricted to that segment.
    """
    metadata = {
        'render.modes': ['human', 'rgb_array'],
        'video.frames_per_second': 10
    }
    def __init__(self, max_size = 50):
        self.max_size = max_size
        self.action_space = spaces.MultiDiscrete([2, max_size, max_size])
        self.observation_space = spaces.MultiDiscrete([4, max_size, max_size])
        self.seed()
        self.viewer = None
        self._expert_data = None
        self.expert_proposal = None
        self.last_action = None
        self.a = None
        self.checked = None
        self.size = None
        self.min_index = None
        self.max_index = None

    def seed(self, seed=None):
        self.np_random, seed = seeding.np_random(seed)
        return [seed]

    def reset(self, goal_min = 0, goal_limit = None):
        self.a = np.zeros(self.max_size, dtype = int)
        self.size = self.np_random.randint(2, self.max_size+1)
        self.min_index = self.np_random.randint(0, self.max_size - self.size + 1)
        self.max_index = self.min_index + self.size-1
        self.a[self.min_index : self.max_index+1] = self.np_random.permutation(self.size)
        self.checked = np.zeros(self.size-1)

        self.last_action = 0, self.min_index, self.max_index
        self.expert_reset()

        return self.get_state()

    def step(self, action):
        action = np.array(action)
        err_msg = "%r (%s) invalid" % (action, type(action))
        assert self.action_space.contains(action), err_msg
        assert self.indices_valid(action[1:]), err_msg

        expert_aligned = (action == self.expert_proposal).all()
        t, i, j = action
        if t == 0:
            self.a[i], self.a[j] = self.a[j], self.a[i]
            self.last_action = 1, i, j
            if not expert_aligned: self.expert_reset()
        else:
            ai, aj = (self.a[x] for x in (i,j))
            if ai < aj: res = 2
            else: res = 3
            self.last_action = res, i, j
            if abs(ai - aj) == 1: self.checked[min(ai,aj)] = True

        if expert_aligned: self._expert_next()

        done = self.goal_accomplished()
        reward = float(done)
        return self.get_state(), reward, done, {}

    def close(self):
        if self.viewer:
            self.viewer.close()
            self.viewer = None

    def goal_accomplished(self):
        return self.checked.all() and\
            (np.diff(self.a[self.min_index : self.max_index+1]) == 1).all()
    def get_state(self):
        return np.array(self.last_action)
    def indices_valid(self, indices):
        return ((self.min_index <= indices) & (indices <= self.max_index)).all()

    def action_to_str(self, action):
        ai, aj = (self.a[x] for x in action[1:])
        i, j = (x - self.min_index for x in action[1:])
        args = i,ai,j,aj
        if action[0] == 0: return "swap a[{}] (= {}) <-> a[{}] (= {})".format(*args)
        else: return "compare a[{}] (= {}) with a[{}] (= {})".format(*args)

    def expert_action(self):
        if self.expert_proposal[0] < 0: return None
        return np.array(self.expert_proposal)
    def expert_reset(self):
        self._expert_data = [self.min_index,self.max_index], [(self.min_index, self.max_index)]
        self.expert_proposal = 1, self.min_index, self.min_index+1
    def _expert_next(self):
        t = self.last_action[0]
        cur_bounds, stack = self._expert_data
        i,j = cur_bounds
        next_segment = False
        if t == 1: # swap i j
            if i < j: self.expert_proposal = 1,i,i+1
            else: next_segment = True
        elif t == 2: # a[i] < a[j]
            if i+1 == j: next_segment = True
            else:
                self.expert_proposal = 0,i+1,j
                cur_bounds[1] -= 1
        elif t == 3: # a[i] > a[j]
            self.expert_proposal = 0,i,i+1
            cur_bounds[0] += 1

        if next_segment:
            i0, i1 = stack.pop()
            #print(np.array([i0,i,i1])-self.min_index)
            if i1 - i > 1: stack.append((i+1, i1))
            if i - i0 > 1: stack.append((i0, i-1))
            if not stack: self.expert_proposal = -1,-1,-1
            else:
                cur_bounds[:] = list(stack[-1])
                self.expert_proposal = 1, cur_bounds[0], cur_bounds[0]+1

    def render(self, mode='human'):
        screen_width = 600
        screen_height = 400
        h_offset = 50
        v_offset = 35
        rel_bar_dist = 0.3
        rel_conn_height = 0.3
        rel_min_height = 3
        observation_height = 10
        observation_dist = 10

        unchecked_color = 0,0,0
        checked_color = 0.5,0.5,0
        checked_placed_color = 0,1,0
        unactive_bar_color = 0.4,0.4,0.4
        active_bar_colors = [
            (1,0,0), (0,0,1),
        ]
        observation_colors = [
            (0.7, 0.7, 0.7), # init
            (0.8, 0.8, 0),   # swap
        ] + list(reversed(active_bar_colors))

        def rect_coor(l, r, t, b):
            return [(l, b), (l, t), (r, t), (r, b)]

        from gym.envs.classic_control import rendering

        if self.viewer is None:
            self.viewer = rendering.Viewer(screen_width, screen_height)
            self._view_size = 0
            self._observation_view = rendering.FilledPolygon(rect_coor(0,1,0,1))

        x_scale = (screen_width-2*h_offset) / (self.size - rel_bar_dist)
        x_shift = h_offset + x_scale * (1-rel_bar_dist)/2
        y_scale = (screen_height-2*v_offset) / (self.size + rel_min_height - 1)
        y_shift = v_offset

        if self._view_size != self.size: # update number of bars
            del self.viewer.geoms[:]

            self._bars_view = []
            self._bars_trans = []
            basic_trans = rendering.Transform(
                translation = (x_shift, y_shift),
                scale = (x_scale, y_scale),
            )
            self._check_view = []
            self._bars_trans = []
            self._bars_view = []
            for i in range(self.size):
                x = (1-rel_bar_dist)/2
                bar = rendering.FilledPolygon(rect_coor(-x,x,0,i+rel_min_height))
                self._bars_view.append(bar)
                objs = [bar]
                if i > 0:
                    conn1 = rendering.FilledPolygon(rect_coor(-0.5,0,0,rel_conn_height))
                    objs.append(conn1)
                    self._check_view.append((conn2, conn1))
                if i < self.size-1:
                    conn2 = rendering.FilledPolygon(rect_coor(0,0.5,0,rel_conn_height))
                    objs.append(conn2)

                trans = rendering.Transform()
                self._bars_trans.append(trans)
                for obj in objs:
                    self.viewer.add_geom(obj)
                    obj.add_attr(trans)
                    obj.add_attr(basic_trans)

            self.viewer.add_geom(self._observation_view)

        # update bars
        for i,b in enumerate(self.a[self.min_index : self.max_index+1]):
            bar = self._bars_view[b]
            bar.set_color(*unactive_bar_color)
            trans = self._bars_trans[b]
            trans.set_translation(i, 0)
        # update active bars
        for i,color in zip(self.last_action[1:], active_bar_colors):
            bar = self._bars_view[self.a[i]]
            bar.set_color(*color)

        # update checked connections
        b_to_i = np.zeros(self.size, dtype = int)
        b_to_i[self.a[self.min_index : self.max_index+1]] = np.arange(self.size)
        for b,(checked, objs) in enumerate(zip(self.checked, self._check_view)):
            if not checked: color = unchecked_color
            elif b_to_i[b] + 1 != b_to_i[b+1]: color = checked_color
            else: color = checked_placed_color
            for obj in objs: obj.set_color(*color)

        # update observed segment
        lr = np.array(self.last_action[1:], dtype = float) - self.min_index
        lr += np.array([-1,1]) * ((1-rel_bar_dist)/2)
        l,r = x_shift + lr*x_scale
        u = v_offset-observation_dist
        d = u - observation_height
        color = observation_colors[self.last_action[0]]
        self._observation_view.v = rect_coor(l,r,u,d)
        self._observation_view.set_color(*color)

        return self.viewer.render(return_rgb_array = mode=='rgb_array')

if __name__ == "__main__":

    from interactive import run_interactive
    run_interactive(ArraySorting(), {})

max_stars_repo_name: mirefek/neur-rec-seq

max_issues_repo_name: mirefek/neur-rec-seq

max_forks_repo_name: mirefek/neur-rec-seq


text:
import numpy as np
import torch
import torch.nn.functional as F
import torchvision.utils
import os,sys,time
import termcolor
import skvideo.io
import render

# convert to colored strings
def red(content): return termcolor.colored(str(content),"red",attrs=["bold"])
def green(content): return termcolor.colored(str(content),"green",attrs=["bold"])
def blue(content): return termcolor.colored(str(content),"blue",attrs=["bold"])
def cyan(content): return termcolor.colored(str(content),"cyan",attrs=["bold"])
def yellow(content): return termcolor.colored(str(content),"yellow",attrs=["bold"])
def magenta(content): return termcolor.colored(str(content),"magenta",attrs=["bold"])

def get_time(sec):
	h = int(sec//3600)
	m = int((sec//60)%60)
	s = sec%60
	return h,m,s

def restore_checkpoint_from_epoch(opt,model,keys):
	print(magenta("resuming from epoch {}...".format(opt.from_epoch)))
	with torch.cuda.device(opt.gpu):
		checkpoint = torch.load("checkpoint/{0}/{1}/ep{2}.npz".format(opt.group,opt.name,opt.from_epoch),map_location=opt.device)
		for k in keys:
			getattr(model,k).load_state_dict(checkpoint[k])

def restore_checkpoint(opt,model,load_name,keys):
	print(magenta("loading checkpoint {}...".format(load_name)))
	with torch.cuda.device(opt.gpu):
		checkpoint = torch.load(load_name,map_location=opt.device)
		for k in keys:
			getattr(model,k).load_state_dict(checkpoint[k])

def save_checkpoint(opt,model,keys,ep):
	os.makedirs("checkpoint/{0}/{1}".format(opt.group,opt.name),exist_ok=True)
	checkpoint = {}
	with torch.cuda.device(opt.gpu):
		for k in keys:
			checkpoint[k] = getattr(model,k).state_dict()
		torch.save(checkpoint,"checkpoint/{0}/{1}/ep{2}.npz".format(opt.group,opt.name,ep))
	print(green("checkpoint saved: ({0}) {1}, epoch {2}".format(opt.group,opt.name,ep)))

def make_tb_image(opt,image):
	return torchvision.utils.make_grid(image[:15],nrow=5,pad_value=0.5)

def save_mesh(opt,model,vertices):
	os.makedirs("optimized_mesh/{0}/{1}".format(opt.group,opt.name),exist_ok=True)
	with torch.cuda.device(opt.gpu):
		torch.save({
			"code": model.code,
			"sim3": model.sim3,
			"faces": model.faces,
			"vertices": vertices,
		},"optimized_mesh/{0}/{1}/{2}_{3}.npz".format(opt.group,opt.name,model.c,model.m))
	print(green("optimized mesh saved: ({0}) {1}".format(opt.group,opt.name)))

def get_edge_map(opt,mask,inv_depth):
	edge = torch.zeros_like(mask)
	depth = 1/inv_depth
	depth[~mask] = 0
	lapl_vert = depth[:-2,:]-2*depth[1:-1,:]+depth[2:,:]
	lapl_horiz = depth[:,:-2]-2*depth[:,1:-1]+depth[:,2:]
	thres = 0.1 # heuristically set for now
	edge[1:-1,:] |= lapl_vert>thres
	edge[:,1:-1] |= lapl_horiz>thres
	return edge

def get_normal_map(opt,index,vertices,faces):
	face_vertices = vertices[faces.long()]
	v1,v2,v3 = torch.unbind(face_vertices,dim=1)
	normal = F.normalize(torch.cross(v2-v1,v3-v2),dim=1)
	# face normals towards camera
	normal[normal[:,2]<0] *= -1
	normal_color = (normal+1)/2
	normal_color = torch.cat([torch.zeros(1,3,device=opt.device),normal_color],dim=0)
	normal_color[0] = 0
	normal_map = normal_color[index.long()+1].permute(2,0,1)
	return normal_map

def visualize_frames(opt,sequence,vertices,faces,vis_idx):
	vis_idx = torch.tensor(vis_idx,dtype=torch.int64,device=opt.device)
	RGB = sequence.RGB[vis_idx]
	cam_extr = sequence.cam_extr[vis_idx]
	cam_intr = sequence.cam_intr
	with torch.no_grad():
		index,_,mask,inv_depth,vertices_t = render.rasterize_3D_mesh(opt,vertices,faces,cam_extr,cam_intr)
	frames = []
	for RGB_i,index_i,mask_i,inv_depth_i,vertices_i in zip(RGB,index,mask,inv_depth,vertices_t):
		frame = visualize_frame_with_mesh(opt,RGB_i,index_i,mask_i,inv_depth_i,vertices_i,faces)
		frames.append(frame)
	return frames

def visualize_frame_with_mesh(opt,RGB,index,mask,inv_depth,vertices,faces):
	normal_color = get_normal_map(opt,index,vertices,faces)
	mask = mask.byte()
	edge = get_edge_map(opt,mask,inv_depth)
	frame = RGB*0.9+0.1
	frame[mask.repeat(3,1,1)] *= 0.4
	frame += normal_color*0.6
	frame[edge.repeat(3,1,1)] = 0
	return frame

def write_video(opt,sequence,vertices,faces,c,m):
	os.makedirs("video/{0}/{1}".format(opt.group,opt.name),exist_ok=True)
	fname = "video/{0}/{1}/{2}_{3}.mp4".format(opt.group,opt.name,c,m)
	video_writer = skvideo.io.FFmpegWriter(fname,outputdict={"-b":"30M","-r":"15"})
	frames = []
	batch_size = sequence.len() if opt.batch_size_pmo==-1 else opt.batch_size_pmo
	for b in range(0,sequence.len(),batch_size):
		idx = torch.arange(b,min(b+batch_size,sequence.len()),dtype=torch.int64,device=opt.device)
		frames_batch = visualize_frames(opt,sequence,vertices,faces,idx)
	frames += frames_batch
	for f in frames:
		f_uint8 = (f*255).byte().cpu().permute(1,2,0).numpy()
		video_writer.writeFrame(f_uint8)
	video_writer.close()

max_stars_repo_name: chenhsuanlin/photometric-mesh-optim

max_issues_repo_name: chenhsuanlin/photometric-mesh-optim

max_forks_repo_name: chenhsuanlin/photometric-mesh-optim


text:
import sys
from pathlib import Path

import numpy as np
import pandas as pd
import cv2
import random as rnd
from pathlib import Path
from pprint import pprint

# -----------------------------------------------------------------------------
# Local modules
_workpath = Path(__file__).resolve().parent
sys.path.append(str(_workpath))

from stereocam import CamResMode, StereoCam, CamUtils

cam = StereoCam(0, CamResMode.MEDIUM, False)
cam_utils = CamUtils()
cam.enable_calibration_eng(remap=False, crop_en=False, resize=False)

while True:
    lft_frame, rgt_frame = cam.get_frame()
    contours_info = cam_utils.get_contours(lft_frame, rgt_frame)
    print(f"lft_contours: {len(contours_info[0])}, rgt_contours: {len(contours_info[1])}")
    lft_frame, rgt_frame = cam_utils.draw_contours(lft_frame, rgt_frame, contours_info, draw_center=True, draw_contours=False, draw_bbox=True)

    cv2.imshow("lft_frame", lft_frame)
    cv2.imshow("rgt_frame", rgt_frame)

    if cv2.waitKey(1) == ord('q'):
        break


text:
from collections import OrderedDict
import matplotlib.pyplot as plt
import numpy as np

class Ledger():
    def __init__(self, users):
        super().__init__()
        self.users = users
        self.txs = []

    def __repr__(self):
        return "[{}]".format(",\n ".join([str(tx[0]) for tx in self.txs]))

    def list_with_spentness(self):
        return "[{}]".format(",\n ".join([str(tx[0]) + ", " + str(tx[1]) for tx in self.txs]))

    def add_transaction(self, tx):
        if tx.parent is None:
            self.txs.append([tx, "unspent"])
        else:
            for in_tx in self.txs:
                if in_tx[0].id == tx.parent:
                    if in_tx[1] == "unspent":
                        in_tx[1] = "spent"
                        self.txs.append([tx, "unspent"])
                    elif in_tx[1] in ["spent", "invalid double spend"]:
                        self.txs.append([tx, "invalid double spend"])
                    return

    def get_transactions(self):
        return [tx[0] for tx in self.txs]

    def get_parent_tx(self, tx):
        for in_tx in [tx[0] for tx in self.txs]:
            if in_tx.id == tx.parent:
                return in_tx

    def swap_order(self, a, b):
        txs = [tx[0] for tx in self.txs]
        txs[a], txs[b] = txs[b], txs[a]
        self.txs = []
        for tx in txs:
            self.add_transaction(tx)

    def get_balances(self, allow_doublespends=True):
        balances = OrderedDict()
        for user in self.users:
            balances[user] = 0

        if allow_doublespends:
            # calculate balances on the fly
            for tx in [tx[0] for tx in self.txs]:
                if tx.recipient_name in balances:
                    # Add 1 to the recipient's balance
                    balances[tx.recipient_name] += 1
                if tx.parent is not None:
                    # Remove 1 from the sender's balance
                    balances[self.get_parent_tx(tx).recipient_name] -= 1
        else:
            # calculate balances based on spentness
            for tx in self.txs:
                if tx[1] == "unspent" and tx[0].recipient_name in balances:
                    balances[tx[0].recipient_name] += 1

        return balances

    def draw_balances(self, allow_doublespends=True):
        fig, ax = plt.subplots()
        ind = np.arange(1, 4)

        balances = list(self.get_balances(allow_doublespends).values())

        n0, n1, n2 = plt.bar(ind, balances)

        n0.set_facecolor('r')
        n1.set_facecolor('r')
        n2.set_facecolor('r')
        ax.set_xticks(ind)
        ax.set_xticklabels(self.users)
        ax.set_ylim([0, max(balances) + 1])
        ax.set_yticks([0, 1, 2])
        ax.set_ylabel('Balance')
        ax.set_title('Balances')

        # ask the canvas to re-draw itself the next time it
        # has a chance.
        # For most of the GUI backends this adds an event to the queue
        # of the GUI frameworks event loop.
        fig.canvas.draw()


text:
import numpy as np
import pandas as pd
import matplotlib.pyplot as mp

works = pd.read_csv("works.csv")

# 1
#print("Общее количество записей:", works.shape[0])

# 2
#print("Количество мужчин:", works[works["gender"] == "Мужской"].shape[0])
#print("Количество женщин:", (works["gender"] == "Женский").sum())

# 3
#print("Количество не NaN значений:", works["skills"].count())

# 4
#print("Все заполненные скиллы\n", works['skills'].dropna())

# 5
#skills_bool = works['skills'].str.lower().str.contains('python | питон') & works['skills'].notnull()
#print("Зарплата тех, у кого в скиллах есть Python (Питон)\n", works[skills_bool]['salary'])

# 6
#salary_p = np.linspace(0.1, 1, 10)
#w = works[works.gender == "Женский"]['salary'].quantile(salary_p)
#m = works[works.gender == "Мужской"]['salary'].quantile(salary_p)

#mp.plot(m, salary_p, color='blue')
#mp.plot(w, salary_p, color='r')
#mp.xlabel('salary')
#mp.ylabel('quantile')
#mp.show()

#7

men_salary = works.query("gender == 'Мужской'").groupby("educationType").agg("mean").reset_index()
women_salary = works.query("gender == 'Женский'").groupby("educationType").agg("mean").reset_index()

educationTypes = men_salary["educationType"].values
men_salaries = men_salary["salary"].values
women_salary = women_salary["salary"].values

index = np.arange(len(educationTypes))

bw = 0.4
mp.bar(index-bw/2, men_salaries, bw, color="b", label="Средняя зарплата мужчин")
mp.bar(index+bw/2, women_salary, bw, color="r", label="Средняя зарплата женщин")
mp.xticks(index, educationTypes, rotation=45)
mp.legend()
mp.show()


text:
"""
LIF (Leaky integrate-and-fire) Neuron model
Copyright(c) HiroshiARAKI
"""
import numpy as np
import matplotlib.pyplot as plt

from .neuron import Neuron
from ..tools import kernel

class LIF(Neuron):
    """
    LIF: leaky integrate-and-fire model
    """

    def __init__(self,
                 time: int,
                 dt: float = 1.0,
                 rest=-65,
                 th=-40,
                 ref=3,
                 tc_decay=100,
                 k='single',
                 tau: tuple = (20, ),
                 **kwargs):
        """
        Initialize Neuron parameters
        :param time:      experimental time
        :param dt:        time step
        :param rest:      resting potential
        :param th:        threshold
        :param ref:       refractory period
        :param tc_decay:  time constance
        :param k:         kernel {'single', 'double'}
        :param tau:       exponential decays as tuple(tau_1 ,tau_2) or float
        """
        super().__init__(time, dt)

        if k not in ['single', 'double']:
            print('The kernel is selected "single".')
            k = 'single'

        self.rest = kwargs.get('rest', rest)
        self.th = kwargs.get('th', th)
        self.ref = kwargs.get('ref', ref)
        self.tc_decay = kwargs.get('tc_decay', tc_decay)
        self.monitor = {}

        self.kernel = kernel[kwargs.get('k', k)]  # default: single exp filter
        self.tau = tau if type(tau) is tuple else (tau, )

    def calc_v(self, data):
        """
        Calculate Membrane Voltage
        :param data: tuple(spikes[], weight[])
        :return:
        """
        spikes = np.array(data[0])
        weights = np.array(data[1])
        data = [
            spikes[i] * weights[i]
            for i in range(weights.size)
        ]
        time = int(self.time / self.dt)

        data = np.sum(data, 0)
        data = np.convolve(data,
                           self.kernel(np.arange(0, self.time, self.dt),
                                       self.tau)
                           )[0:time]

        # initialize
        f_last = 0  # last firing time
        vpeak = 20  # the peak of membrane voltage
        spikes = np.zeros(time)
        v = self.rest  # set to resting voltage

        v_monitor = []  # monitor voltage

        # Core of LIF
        for t in range(time):
            dv = ((self.dt * t) > (f_last + self.ref)) * (-v + self.rest) / self.tc_decay + data[t]
            v = v + self.dt * dv  # calc voltage

            f_last = f_last + (self.dt * t - f_last) * (v >= self.th)  # if fires, memory the firing time
            v = v + (vpeak - v) * (v >= self.th)  # set to peak

            v_monitor.append(v)

            spikes[t] = (v >= self.th) * 1  # set to spike

            v = v + (self.rest - v) * (v >= self.th)  # return to resting voltage

        self.monitor['s'] = spikes
        self.monitor['v'] = v_monitor
        self.monitor['f'] = np.arange(0, self.time, self.dt)[v >= self.th],  # real firing times

        return v_monitor, spikes, self.monitor['f']

    def plot_v(self, save=False, filename='lif.png', **kwargs):
        """
        plot membrane potential
        :param save:
        :param filename:
        :param kwargs:
        :return:
        """
        x = np.arange(0, self.time, self.dt)
        plt.title('LIF Neuron model Simulation')
        plt.plot(x, self.monitor['v'])
        plt.ylabel('V [mV]')
        plt.xlabel('time [ms]')
        if not save:
            plt.show()
        else:
            plt.savefig(filename, dpi=kwargs.get('dpi', 150))
        plt.close()

class IF(LIF):
    """
    IF: integrate-and-fire model
    """
    def __init__(self,
                 time: int,
                 dt: float = 1.0,
                 rest=-65,
                 th=-40,
                 ref=3,
                 k='single',
                 tau: tuple = (20, ),
                 **kwargs):
        """
        Initialize Neuron parameters
        :param time:
        :param dt:
        :param rest:
        :param th:
        :param ref:
        :param k:
        :param tau:
        :param kwargs:
        """
        super().__init__(time=time,
                         dt=dt,
                         rest=rest,
                         th=th,
                         ref=ref,
                         k=k,
                         tau=tau,
                         **kwargs)

    def calc_v(self, data):
        """
        Calculate Membrane Voltage
        :param data: tuple(spikes[], weight[])
        :return membrane voltage, output spikes, firing times:
        """
        spikes = np.array(data[0])
        weights = np.array(data[1])

        peak = 20  # the peak of membrane voltage
        f_last = -100  # last firing time
        t_ref = int(self.ref / self.dt)  # refractory period [x dt ms]

        v = np.zeros(int(self.time / self.dt)) + self.rest  # all membrane voltage set to resting vol.

        input_spikes = np.array([
            spikes[i] * weights[i]
            for i in range(weights.size)
        ])
        input_spikes = np.sum(input_spikes, 0)
        for t, s in enumerate(input_spikes):
            if s:  # if fires,
                # and be not in refractory period, calculate voltage
                v[t:] += (t > (f_last + t_ref)) * s * self.kernel(np.arange(0, v[t:].size, 1) * self.dt, self.tau)

            if v[t] >= self.th:
                v[t] = peak
                v[t+1:] = self.rest  # return to resting voltage
                f_last = t  # memory the firing time

        self.monitor = {
            'v': v,
            's': [v >= self.th],  # boolean spike trains
            'f': np.arange(0, self.time, self.dt)[v >= self.th],  # real firing times
        }

        return v, self.monitor['s'], self.monitor['f']

max_stars_repo_name: HiroshiARAKI/spine

max_issues_repo_name: HiroshiARAKI/spine

max_forks_repo_name: HiroshiARAKI/spine


text:
# Copyright 2020 - 2021 MONAI Consortium
# 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 os
import tempfile
import unittest

import itk
import nibabel as nib
import numpy as np
from parameterized import parameterized
from PIL import Image

from monai.data import ITKReader
from monai.transforms import EnsureChannelFirst, LoadImage
from tests.utils import TEST_NDARRAYS

TEST_CASE_1 = [{"image_only": False}, ["test_image.nii.gz"], None]

TEST_CASE_2 = [{"image_only": False}, ["test_image.nii.gz"], -1]

TEST_CASE_3 = [{"image_only": False}, ["test_image.nii.gz", "test_image2.nii.gz", "test_image3.nii.gz"], None]

TEST_CASE_4 = [{"reader": ITKReader(), "image_only": False}, ["test_image.nii.gz"], None]

TEST_CASE_5 = [{"reader": ITKReader(), "image_only": False}, ["test_image.nii.gz"], -1]

TEST_CASE_6 = [
    {"reader": ITKReader(), "image_only": False},
    ["test_image.nii.gz", "test_image2.nii.gz", "test_image3.nii.gz"],
    None,
]

TEST_CASE_7 = [{"image_only": False, "reader": ITKReader(pixel_type=itk.UC)}, "tests/testing_data/CT_DICOM", None]

class TestEnsureChannelFirst(unittest.TestCase):
    @parameterized.expand([TEST_CASE_1, TEST_CASE_2, TEST_CASE_3, TEST_CASE_4, TEST_CASE_5, TEST_CASE_6])
    def test_load_nifti(self, input_param, filenames, original_channel_dim):
        if original_channel_dim is None:
            test_image = np.random.rand(128, 128, 128)
        elif original_channel_dim == -1:
            test_image = np.random.rand(128, 128, 128, 1)

        with tempfile.TemporaryDirectory() as tempdir:
            for i, name in enumerate(filenames):
                filenames[i] = os.path.join(tempdir, name)
                nib.save(nib.Nifti1Image(test_image, np.eye(4)), filenames[i])
            for p in TEST_NDARRAYS:
                result, header = LoadImage(**input_param)(filenames)
                result = EnsureChannelFirst()(p(result), header)
                self.assertEqual(result.shape[0], len(filenames))

    @parameterized.expand([TEST_CASE_7])
    def test_itk_dicom_series_reader(self, input_param, filenames, original_channel_dim):
        result, header = LoadImage(**input_param)(filenames)
        result = EnsureChannelFirst()(result, header)
        self.assertEqual(result.shape[0], 1)

    def test_load_png(self):
        spatial_size = (256, 256, 3)
        test_image = np.random.randint(0, 256, size=spatial_size)
        with tempfile.TemporaryDirectory() as tempdir:
            filename = os.path.join(tempdir, "test_image.png")
            Image.fromarray(test_image.astype("uint8")).save(filename)
            result, header = LoadImage(image_only=False)(filename)
            result = EnsureChannelFirst()(result, header)
            self.assertEqual(result.shape[0], 3)

    def test_check(self):
        with self.assertRaises(ValueError):  # no meta
            EnsureChannelFirst()(np.zeros((1, 2, 3)), None)
        with self.assertRaises(ValueError):  # no meta channel
            EnsureChannelFirst()(np.zeros((1, 2, 3)), {"original_channel_dim": None})
        EnsureChannelFirst(strict_check=False)(np.zeros((1, 2, 3)), None)
        EnsureChannelFirst(strict_check=False)(np.zeros((1, 2, 3)), {"original_channel_dim": None})

if __name__ == "__main__":
    unittest.main()


text:
from collections import defaultdict

import networkx as nx

class Node:
    """Class representing a node in the KB.
    """
    def __init__(self, kb, name, data, watches=[]):
        super().__setattr__('_kb', kb)
        super().__setattr__('_name', name)
        nx.set_node_attributes(self._kb.G, {self._name: data})
        self._watches = defaultdict(list)
        for watch in watches:
            self._watches[watch[0]].append(watch[1])

    def __repr__(self):
        return self._name

    def __eq__(self, comparator):
        return self._name == str(comparator)

    def __ne__(self, comparator):
        return self._name != str(comparator)

    def __getattr__(self, key):
        try:
            return self._kb.G.nodes(data=True)[self._name][key]
        except KeyError as e:
            return None

    def __setattr__(self, key, value):
        attrs = self._kb.G.nodes(data=True)[self._name]
        prev_val = attrs.get(key, None)
        attrs.update({key: value})
        nx.set_node_attributes(self._kb.G, {self._name: attrs})
        for watch_fn in self._watches.get(key, []):
            watch_fn(self, prev_val)

    def __getitem__(self, key):
        return self.__getattr__(key)

    def __setitem__(self, key, value):
        return self.__setattr__(key, value)

    @property
    def attrs(self):
        """Returns attributes of the node stored in the KB
        """
        attributes = self._kb.G.nodes(data=True)[self._name]
        try:
            del attributes['_watches']
            del attributes['_new_neighbor_fn']
        except:
            pass
        return attributes

    @property
    def neighbors(self):
        """Returns the node's neighbors, in the format of tuples:
        [(neighbor_name, [{'pred': predicate aka edge_relation}])]
        """
        return self._kb.neighbors(self._name)

    def watch(self, attribute, fn):
        """Execute user-defined function when the value of attribute changes.
        Function takes two args: `node` which has access to all
        its own attributes, including neighbors and edges, and the second
        arg is the previous value of the attribute that changed.

        :returns int: id of the watch

        :Example:

        >>> kb.store('node(node1)')
        >>> node = kb.node('node1')
        >>> node.grains = 3
        >>> print(node.grains)
        3
        >>> node.watch('grains', lambda x: print('grains changed to ' + x.grains))
        ('grains', 0)
        >>> node.grains += 1
        grains changed to 4

        """
        self._watches[attribute].append(fn)
        return (attribute, len(self._watches) - 1)

    def remove_watch(self, attribute_or_watch_id):
        """Stop watching `attribute_or_watch_id`.
        If it is a string, delete all watches for that attribute.
        If it is a tuple of (attribute, watch_id): delete that specific watch.
        """
        if isinstance(attribute_or_watch_id, tuple):
            self._watches[attribute_or_watch_id[0]].pop(attribute_or_watch_id[1])
        else:
            self._watches[attribute_or_watch_id] = []

    def watch_for_new_neighbor(self, fn):
        """Execute `fn` when node receives a new neighbor."""
        self.__setattr__('_new_neighbor_fn', fn)

reason: import networkx


text:
import numpy as np
import pandas as pd
import os, sys
from collections import OrderedDict
import torch
from torch.utils.data import TensorDataset, DataLoader
import argparse

import math

from scipy import signal

ROBOTICS_CODESIGN_DIR = os.environ['ROBOTICS_CODESIGN_DIR']
sys.path.append(ROBOTICS_CODESIGN_DIR)
sys.path.append(ROBOTICS_CODESIGN_DIR + '/utils/')

from textfile_utils import *
from plotting_utils import *
from sklearn.preprocessing import MinMaxScaler

DEFAULT_PARAMS = {'batch_size':  1024, 'shuffle': True, 'num_workers': 1, 'pin_memory': True}

device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

np.random.seed(0)

def create_dataloader_from_tensors(inputs_tensor, outputs_tensor, params = DEFAULT_PARAMS):

    tensor_dataset = TensorDataset(inputs_tensor, outputs_tensor)
    tensor_dataloader = DataLoader(tensor_dataset, **params)

    return tensor_dataset, tensor_dataloader

if __name__ == '__main__':

    parser = argparse.ArgumentParser()
    parser.add_argument('--model_name', type=str)
    args = parser.parse_args()

    model_name = args.model_name

    SCRATCH_DIR = ROBOTICS_CODESIGN_DIR + '/scratch/' + model_name

    pjm_folder = ROBOTICS_CODESIGN_DIR + 'data/PJM/'

    # NI has only 2004-2010 data
    zones = [ 'AEP', 'COMED', 'DAYTON', 'DEOK', 'DOM', 'DUQ', 'EKPC', 'FE']

    num_dimensions = len(zones)

    x_dim = num_dimensions
    u_dim = num_dimensions
    s_dim = num_dimensions

    A = np.identity(num_dimensions)
    B = np.identity(num_dimensions)
    C = -np.identity(num_dimensions)
    Q = np.identity(num_dimensions)
    R = np.identity(num_dimensions)

    W = 24
    H = 24

    T = 168
    num_samples = 30

    zone_dfs = []

    for zone in zones:
        csv_path = pjm_folder + '{}_hourly.csv'.format(zone)
        basic_df = pd.read_csv(csv_path, parse_dates=['Datetime'])
        basic_df = basic_df[basic_df.Datetime.dt.year == 2017]
        basic_df = basic_df.set_index('Datetime')
        zone_dfs.append(basic_df)
        print(basic_df)
        # print(list(basic_df))

    data_df = pd.concat(zone_dfs, axis=1)

    print(data_df)

    data_list = np.zeros((num_samples*T, s_dim))

    for i in range(s_dim):
        column_name = "{}_MW".format(zones[i])
        data_list[ :, i] = data_df[column_name].to_list()[:num_samples*T]

    print(data_list)

    # min_max_scaler = MinMaxScaler()
    min_max_scaler = MinMaxScaler(feature_range=(0, 1))
    data_list = min_max_scaler.fit_transform(data_list)

    print("Min: {}".format(min_max_scaler.data_min_))
    print("Max: {}".format(min_max_scaler.data_max_))

    data_list = data_list.reshape(num_samples, T, s_dim)

    print(data_list)

    train_inputs = []
    train_outputs = []

    for i in range(int(num_samples / 2)):
        for t in range(W-1, T-H+1):
            train_inputs.append(data_list[i, t-W+1:t+1, :].reshape(W*s_dim, 1))
            train_outputs.append(data_list[i, t:t+H, :].reshape(H*s_dim, 1))

    train_inputs_tensor = torch.tensor(train_inputs, dtype=torch.float32)
    train_outputs_tensor = torch.tensor(train_outputs, dtype=torch.float32)

    _, train_dataloader = create_dataloader_from_tensors(train_inputs_tensor, train_outputs_tensor)

    train_time_series = data_list[:int(num_samples/2), : ,:].reshape(int(num_samples/2), T*s_dim, 1)
    train_dataset = torch.tensor(train_time_series, dtype=torch.float32)

    val_inputs = []
    val_outputs = []

    for i in range(int(num_samples / 2), num_samples):
        for t in range(W-1, T-H+1):
            val_inputs.append(data_list[i, t-W+1:t+1, :].reshape(W*s_dim, 1))
            val_outputs.append(data_list[i, t:t+H, :].reshape(H*s_dim, 1))

    val_inputs_tensor = torch.tensor(val_inputs, dtype=torch.float32)
    val_outputs_tensor = torch.tensor(val_outputs, dtype=torch.float32)

    _, val_dataloader = create_dataloader_from_tensors(val_inputs_tensor, val_outputs_tensor)

    val_time_series = data_list[int(num_samples/2):num_samples, : ,:].reshape(int(num_samples/2), T*s_dim, 1)
    val_dataset = torch.tensor(val_time_series, dtype=torch.float32)

    RESULTS_DIR = SCRATCH_DIR + '/dataset/'
    remove_and_create_dir(RESULTS_DIR)

    data_dict = OrderedDict()

    data_dict['train_inputs'] = train_inputs_tensor
    data_dict['train_outputs'] = train_outputs_tensor
    data_dict['train_dataloader'] = train_dataloader
    data_dict['train_dataset'] = train_dataset

    data_dict['val_inputs'] = val_inputs_tensor
    data_dict['val_outputs'] = val_outputs_tensor
    data_dict['val_dataloader'] = val_dataloader
    data_dict['val_dataset'] = val_dataset

    data_dict['x_dim'] = x_dim
    data_dict['u_dim'] = u_dim
    data_dict['s_dim'] = s_dim

    data_dict['A'] = A
    data_dict['B'] = B
    data_dict['C'] = C
    data_dict['Q'] = Q
    data_dict['R'] = R

    data_dict['W'] = W
    data_dict['H'] = H

    # length of the time series for evaluation
    data_dict['T'] = T

    write_pkl(fname = RESULTS_DIR + '/results.pkl', input_dict = data_dict)

reason: import numpy,from scipy


text:
# Copyright 2021 Huawei Technologies Co., Ltd
#
# 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 os.path as ops
import os
import math
import time
import sys
import tensorflow as tf
import numpy as np
import glog as log

cur_path = os.path.abspath(os.path.dirname(__file__))
working_dir = os.path.join(cur_path, '../')
sys.path.append(working_dir)

from config import global_config
from data_provider import tf_io_pipline_fast_tools
from local_utils import evaluation_tools

CFG = global_config.cfg

def init_args():
    """
    :return: parsed arguments and (updated) config.cfg object
    """
    parser = argparse.ArgumentParser()
    parser.add_argument('-d', '--input_dir', type=str,default='./output_fp32',
                        help='Directory of npu predict output')
    parser.add_argument('-c', '--char_dict_path', type=str,default='data/char_dict/char_dict.json',
                        help='Directory where character dictionaries for the dataset were stored')
    parser.add_argument('-o', '--ord_map_dict_path', type=str,default='data/char_dict/ord_map.json',
                        help='Directory where ord map dictionaries for the dataset were stored')
    parser.add_argument('-l', '--label_dir', type=str, default='./labels',
                        help='Path of ground truth labels')
    parser.add_argument('-b', '--batchsize', type=int, default=64,
                        help='batchsize of input dataset')

    return parser.parse_args()

def evaluate_shadownet(input_dir, label_dir, char_dict_path,
                       ord_map_dict_path,batchsize=64):
    # setup decoder
    decoder = tf_io_pipline_fast_tools.CrnnFeatureReader(
        char_dict_path=char_dict_path,
        ord_map_dict_path=ord_map_dict_path
    )
    test_inference_ret = tf.placeholder(tf.float32, shape=[25,batchsize,37], name="test_inference_ret")
    test_decoded, test_log_prob = tf.nn.ctc_greedy_decoder(
        test_inference_ret,
        CFG.ARCH.SEQ_LENGTH * np.ones(batchsize),
        merge_repeated=True
    )

    config = tf.ConfigProto()
    sess = tf.Session(config=config)
    inputPath = input_dir
    labelPath = label_dir
    with sess.as_default():

        log.info('Start predicting...')

        per_char_accuracy = 0
        full_sequence_accuracy = 0.0
        filelist = os.listdir(inputPath)
        filelist.sort()
        i = 0

        for file in filelist:
        if file.endswith(".bin"):
            with open(os.path.join(labelPath, 'batch_label_'+str(i).rjust(3,'0')+'.txt'),'r') as f:
                data = f.read()
            batch_label = data.replace('\r','').replace('\n','').split(',')
            input_data = np.fromfile(os.path.join(inputPath,file),dtype="float32").reshape(25,batchsize,37)
            test_predictions_value = sess.run(test_decoded,feed_dict={test_inference_ret: input_data})
            test_predictions_value = decoder.sparse_tensor_to_str(test_predictions_value[0])

            per_char_accuracy += evaluation_tools.compute_accuracy(
                        batch_label, test_predictions_value, display=False, mode='per_char'
                    )

            full_sequence_accuracy += evaluation_tools.compute_accuracy(
                        batch_label, test_predictions_value, display=False, mode='full_sequence'
                    )
            i += 1

        avg_per_char_accuracy = per_char_accuracy / i
        avg_full_sequence_accuracy = full_sequence_accuracy / i
        log.info('Mean test per char accuracy is {:5f}'.format(avg_per_char_accuracy))
        log.info('Mean test full sequence accuracy is {:5f}'.format(avg_full_sequence_accuracy))

if __name__ == '__main__':
    args = init_args()

    evaluate_shadownet(
        input_dir=args.input_dir,
        label_dir=args.label_dir,
        char_dict_path=args.char_dict_path,
        ord_map_dict_path=args.ord_map_dict_path,
        batchsize=args.batchsize
    )

max_stars_repo_name: Ascend/modelzoo

max_issues_repo_name: Ascend/modelzoo

max_forks_repo_name: Ascend/modelzoo