text:
#!/usr/bin/env python

"""Tests for `prime_sieve` package."""

import numpy as np
import pytest
from prime_sieve.array import PrimeArraySieve
from prime_sieve.base import SegmentedPrimeSieve
from prime_sieve.list import PrimeListSieve
from prime_sieve.math_utils import smallest_multiple_of_n_geq_m

first_100_primes = [2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83,
                    89, 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151, 157, 163, 167, 173, 179,
                    181, 191, 193, 197, 199, 211, 223, 227, 229, 233, 239, 241, 251, 257, 263, 269, 271,
                    277, 281, 283, 293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367, 373, 379,
                    383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461, 463, 467, 479,
                    487, 491, 499, 503, 509, 521, 523, 541]

@pytest.fixture(params=[PrimeArraySieve, PrimeListSieve])
def sieve(request) -> SegmentedPrimeSieve:
    return request.param()

@pytest.mark.parametrize("test_input,expected",
                         [((2, 3), 4),
                          ((3, 2), 3),
                          ((2, 4), 4),
                          ((3, 3), 3),
                          ((3, 6), 6),
                          ((3, 7), 9)])
def test_smallest_multiple_of_n_geq_m(test_input, expected):
    assert smallest_multiple_of_n_geq_m(*test_input) == expected

@pytest.mark.parametrize("test_input,expected", enumerate(first_100_primes))
def test_nth_prime(test_input, expected, sieve):
    """
    See for expected answers:
    https://en.wikipedia.org/wiki/List_of_prime_numbers
    """
    assert sieve.nth_prime(test_input) == expected

@pytest.mark.parametrize("test_input,expected", enumerate(first_100_primes))
def test_getitem_nth_prime(test_input, expected, sieve):
    assert sieve[test_input] == expected

@pytest.mark.parametrize("test_input", [slice(0, 10), slice(4, 19), slice(4, 4), slice(6, 5), slice(5, 20)])
def test_getitem_slice(test_input, sieve):
    assert list(sieve[test_input]) == first_100_primes[test_input]

@pytest.mark.parametrize("test_input,expected",
                         [(0, False),
                          (1, False),
                          (2, True),
                          (3, True),
                          (4, False),
                          (5, True),
                          (6, False),
                          (97, True),
                          (100, False),
                          (100 * 200, False),
                          (86 * 97, False),
                          (2 ** 11 - 1, False),
                          (2 ** 13 - 1, True)])
def test_is_prime(test_input, expected, sieve):
    assert sieve.is_prime(test_input) == expected

@pytest.mark.parametrize("test_input,expected",
                         [(0, False),
                          (1, False),
                          (2, True),
                          (3, True),
                          (4, False),
                          (5, True),
                          (6, False),
                          (97, True),
                          (100, False),
                          (100 * 200, False),
                          (86 * 97, False),
                          (2 ** 11 - 1, False),
                          (2 ** 13 - 1, True)])
def test_is_prime__contains__alias(test_input, expected, sieve):
    assert (test_input in sieve) == expected

@pytest.mark.parametrize("test_input,expected",
                         [((0, 5), [2, 3]),
                          ((2, 5), [2, 3]),
                          ((2, 6), [2, 3, 5]),
                          ((10, 50), [11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47]),
                          ((2, 100), first_100_primes[:25])
                          ])
def test_primes_in_range(test_input, expected, sieve):
    assert list(sieve.primes_in_range(*test_input)) == expected

@pytest.mark.parametrize("test_input,expected",
                         [(p, i + 1) for i, p in enumerate(first_100_primes)] +
                         [(12, 5),
                          (13, 6),
                          (14, 6),
                          (15, 6),
                          (16, 6),
                          (17, 7)])
def test_index_of_next_prime_greater_than(test_input, expected, sieve):
    assert sieve.index_of_next_prime_greater_than(test_input) == expected

@pytest.mark.parametrize("test_input,expected",
                         [(0, 2),
                          (1, 2),
                          (2, 3),
                          (3, 5),
                          (4, 5),
                          (100, 101),
                          (104, 107),
                          (107, 109),
                          (7907, 7919)])
def test_next_prime_greater_than(test_input, expected, sieve):
    assert sieve.next_prime_greater_than(test_input) == expected

@pytest.mark.parametrize("test_input,expected",
                         [(12, 4),
                          (13, 4),
                          (14, 5),
                          (15, 5),
                          (16, 5),
                          (17, 5),
                          (18, 6)])
def test_index_of_prev_prime_less_than(test_input, expected, sieve):
    assert sieve.index_of_prev_prime_less_than(test_input) == expected

def test_index_of_prev_prime_less_than_error(sieve):
    with pytest.raises(ValueError):
        sieve.index_of_prev_prime_less_than(2)

@pytest.mark.parametrize("test_input,expected",
                         [(3, 2),
                          (4, 3),
                          (5, 3),
                          (6, 5),
                          (7, 5),
                          (8, 7),
                          (101, 97),
                          (104, 103),
                          (109, 107),
                          (7907, 7901)])
def test_prev_prime_less_than(test_input, expected, sieve):
    assert sieve.prev_prime_less_than(test_input) == expected

@pytest.mark.parametrize("test_input,expected",
                         [(1, 0),
                          (2, 1),
                          (3, 2),
                          (4, 2),
                          (5, 3),
                          (6, 3),
                          (7, 4),
                          (8, 4),
                          (9, 4),
                          (10, 4),
                          (10 ** 2, 25),
                          (10 ** 3, 168),
                          (10 ** 4, 1229),
                          (10 ** 5, 9592),
                          (10 ** 6, 78498),
                          (10 ** 7, 664579)])
def test_count_primes_less_or_equal(test_input, expected, sieve):
    """
    See for expected answers:
    https://en.wikipedia.org/wiki/Prime-counting_function
    """
    assert sieve.count_primes_less_or_equal(test_input) == expected

@pytest.mark.parametrize("test_input,expected",
                         [((4, 3), 0),
                          ((3, 3), 0),
                          ((2, 3), 1),
                          ((2, 4), 2),
                          ((0, 10), 4),
                          ((2, 10), 4),
                          ((3, 10), 3),
                          ((3, 9), 3),
                          ((3, 8), 3),
                          ((3, 7), 2),
                          ((1, 100), 25), ])
def test_count_primes_in_range(test_input, expected, sieve):
    assert sieve.count_primes_in_range(*test_input) == expected

@pytest.mark.parametrize("test_input", [1, 10, 100, 100])
def test_ensure_len_greater_or_equal(test_input, sieve):
    sieve.ensure_len_greater_or_equal(test_input)
    assert len(sieve) >= test_input

@pytest.mark.parametrize("test_input", [1, 10, 100, 1000])
def test_ensure_max_greater_or_equal(test_input, sieve):
    sieve.ensure_max_greater_or_equal(test_input)
    assert sieve[-1] >= test_input

@pytest.mark.parametrize("dtype", [np.int8, np.uint8, np.int16, np.uint16])
def test_numpy_overflow_error(dtype):
    sieve = PrimeArraySieve(dtype)
    max = np.iinfo(dtype).max
    with pytest.raises(RuntimeError):
        sieve.ensure_max_greater_or_equal(max)

def test_iter_all_primes(sieve):
    primes_found = []
    for p in sieve.iter_all_primes():
        primes_found.append(p)
        if p % 60 == 1:
            break
    assert primes_found == first_100_primes[:18]

def test_find_primes_until(sieve):
    def stop_cb(s):
        return len(s) >= 100

    call_count = 0

    def progress_cb(s):
        nonlocal call_count
        call_count += 1

    sieve.find_primes_until(stop_cb, progress_cb)
    assert len(sieve) >= 100
    assert call_count > 0

@pytest.mark.parametrize("expected,test_input", enumerate(first_100_primes))
def test_index_of_prime(test_input, expected, sieve):
    assert sieve.index_of(test_input) == expected

@pytest.mark.parametrize("test_input", [0, 4, 6, 9, 25, 2 ** 10 + 2])
def test_index_of_not_prime_raises(test_input, sieve):
    with pytest.raises(ValueError):
        sieve.index_of(test_input)


text:
import json
import numpy as np
from rl.analytics import ideal_fitness_score
from rl.hyperoptimizer.base_hyperoptimizer import HyperOptimizer
from rl.util import PROBLEMS, to_json, logger

class RandomSearch(HyperOptimizer):

    '''
    Random Search by sampling on hysphere around a search path
    algo:
    1. init x a random position in space
    2. until termination (max_eval or fitness, e.g. solved all), do:
        2.1 sample new pos some radius away: next_x = x + r
        2.2 if f(next_x) > f(x) then set x = next_x

    Extra search memory units:
    - search_path
    - best_point

    save for experiment resume, search_history:
    - search_path
    - best_point
    - param_search_list
    '''

    # # calculate the constant radius needed to traverse unit cube
    # def cube_traversal_radius(self):
    #     traversal_diameter = 1/np.power(self.max_evals,
    #                                     1/self.search_dim)
    #     traversal_radius = traversal_diameter/2
    #     return traversal_radius

    def decay_search_radius(self):
        '''
        start of half cube for diameter (0.25 radius) then decay
        at 100 searches, will shrink to 1/10 of initial radius 0.025
        clip to prevent going too small (0.01)
        '''
        min_radius = 0.01
        linear_decay_rate = self.next_trial_num/10./self.PARALLEL_PROCESS_NUM
        self.search_radius = np.clip(
            self.init_search_radius / linear_decay_rate,
            min_radius, self.init_search_radius)

    @classmethod
    def sample_hypersphere(cls, dim, r=1):
        '''Marsaglia algo for sampling uniformly on a hypersphere'''
        v = np.random.randn(dim)
        v = v * r / np.linalg.norm(v)
        return v

    def sample_cube(self):
        return np.random.rand(self.search_dim)

    def sample_r(self):
        return self.sample_hypersphere(
            self.search_dim, self.search_radius)

    # biject [0, 1] to [x_min, x_max]
    def biject_continuous(self, norm_val, x_min, x_max):
        return np.around(norm_val*(x_max - x_min) + x_min, self.precision)

    # biject [0, 1] to x_list = [a, b, c, ...] by binning
    def biject_discrete(self, norm_val, x_list):
        list_len = len(x_list)
        inds = np.arange(list_len)
        cont_val = self.biject_continuous(norm_val, 0, list_len)
        ind = np.digitize(cont_val, inds) - 1
        return x_list[ind]

    # biject one dimension: [0, 1] to a param_range val
    def biject_dim(self, norm_val, dim_spec):
        if isinstance(dim_spec, list):  # discrete
            return self.biject_discrete(norm_val, dim_spec)
        else:  # cont
            return self.biject_continuous(
                norm_val, dim_spec['min'], dim_spec['max'])
        return

    # biject a vector on unit cube into a param in param_space
    def biject_param(self, v):
        param = {}
        for i, param_key in enumerate(self.param_range_keys):
            dim_spec = self.param_range[param_key]
            param[param_key] = self.biject_dim(v[i], dim_spec)
        return param

    def init_search(self):
        '''
        Initialize the random search internal variables
        '''
        self.max_evals = self.experiment_spec['param']['max_evals']
        self.num_of_trials = self.max_evals
        self.search_dim = len(self.param_range_keys)
        self.precision = 4  # decimal roundoff biject_continuous
        self.search_radius = self.init_search_radius = 0.5
        self.search_count = 0  # number of times search() has ran
        self.search_exhausted = False
        self.search_path = []
        self.best_point = {
            'trial_num': None,
            'param': None,
            'x': self.sample_cube(),
            'fitness_score': float('-inf'),
        }
        problem = PROBLEMS.get(self.experiment_spec['problem'])
        self.ideal_fitness_score = ideal_fitness_score(problem)
        logger.info(
            'ideal_fitness_scrore: {}'.format(self.ideal_fitness_score))

        self.filename = './data/{}/random_search_history.json'.format(
            self.experiment_id)
        if self.experiment_id_override is not None:
            self.load()  # resume

    def search(self):
        '''
        algo step 2.1 sample new pos some radius away: next_x = x + r
        update search_path and param_search_list
        '''
        self.search_count += 1
        if self.next_trial_num < len(self.search_path):  # resuming
            next_x = self.search_path[self.next_trial_num]
            next_param = self.param_search_list[self.next_trial_num]
        else:
            next_x = np.clip(self.best_point['x'] + self.sample_r(), 0., 1.)
            # check if too close to previously searched x
            distances = [np.linalg.norm(next_x - old_x)
                         for old_x in self.search_path]
            distances = np.around(distances, self.precision)

            if self.search_count > (10 * self.max_evals):
                self.search_exhausted = True  # exhausted search space
                next_param = self.biject_param(next_x)
                self.search_path.append(next_x)
                self.param_search_list.append(next_param)
            elif len(distances) > 0 and np.amin(distances) == 0:
                self.search()
            else:
                next_param = self.biject_param(next_x)
                self.search_path.append(next_x)
                self.param_search_list.append(next_param)

    def update_search(self):
        '''
        algo step 2.2 if f(next_x) > f(x) then set x = next_x
        invoked right after the latest run_trial()
        update self.best_point
        '''
        if (self.next_trial_num < self.PARALLEL_PROCESS_NUM or
                self.next_trial_num < len(self.search_path)):
            # yet to have history or still resuming from history
            return
        assert len(self.experiment_data) > 0, \
            'self.experiment_data must not be empty for update_search'

        self.decay_search_radius()

        x = self.search_path[-1]
        trial_data = self.experiment_data[-1]
        trial_num, param, fitness_score = self.get_fitness(trial_data)
        if fitness_score > self.best_point['fitness_score']:
            self.best_point = {
                'trial_num': trial_num,
                'param': param,
                'x': x,
                'fitness_score': fitness_score,
            }
        self.save()

    def save(self):
        search_history = {
            'search_path': self.search_path,
            'search_count': self.search_count,
            'best_point': self.best_point,
            'param_search_list': self.param_search_list,
        }
        with open(self.filename, 'w') as f:
            f.write(to_json(search_history))
        logger.info(
            'Save search history to {}'.format(self.filename))
        return

    def load(self):
        try:
            search_history = json.loads(open(self.filename).read())
            self.search_path = search_history['search_path']
            self.best_point = search_history['best_point']
            self.param_search_list = search_history['param_search_list']
            logger.info('Load search history from {}'.format(self.filename))
        except (FileNotFoundError, json.JSONDecodeError):
            logger.info(
                'Fail to load search history from {}'.format(self.filename))
            return None

    def satisfy_fitness(self):
        '''
        break on the first strong solution
        '''
        best_fitness_score = self.best_point['fitness_score']
        if self.next_trial_num < self.PARALLEL_PROCESS_NUM:
            return False
        elif best_fitness_score > self.ideal_fitness_score:
            logger.info(
                'fitness_score {} > ideal_fitness_score {}, '
                'could terminate early'.format(
                    best_fitness_score, self.ideal_fitness_score))
            # return True
            # TODO fix ideal_fitness_score
            return False
        else:
            return False

    def to_terminate(self):
        return (self.search_exhausted or
                self.next_trial_num >= self.max_evals or
                self.satisfy_fitness())

ext: py

reason: import numpy


text:
#!/usr/bin/python

import grizzly.grizzly as gr
import numpy as np
import unittest

class PandasWeldTestMethods(unittest.TestCase):
    # TODO: Add more tests here

    def test_unique(self):
        input = gr.SeriesWeld(
            np.array(["aaa", "bbb", "aaa", "ccc"], dtype=str),
            gr.WeldVec(gr.WeldChar()))
        self.assertItemsEqual(["aaa", "bbb", "ccc"],
                              input.unique().evaluate(False))

    def test_sum(self):
        inp = gr.SeriesWeld(
            np.array([1, 2, 3, 4, 5], dtype=np.int32), gr.WeldInt())
        self.assertEqual(15, inp.sum().evaluate(False))

    def test_count(self):
        input = gr.SeriesWeld(
            np.array([1, 2, 3, 4, 5], dtype=np.int32), gr.WeldInt())
        self.assertEqual(5, input.count().evaluate(False))

    def test_eq(self):
        input = gr.SeriesWeld(
            np.array(["aaaa", "bbb", "aa", "cccc"], dtype=str),
            gr.WeldVec(gr.WeldChar()))
        expected_output = [True, False, False, False]
        self.assertSequenceEqual(
            expected_output, list(
                (input == "aaaa").evaluate(False)))

    def test_simpler_eq(self):
        input = gr.SeriesWeld(
            np.array([1, 2, 100, 1, 2, 65], dtype=np.int32), gr.WeldInt())
        expected_output = [False, False, True, False, False, False]
        self.assertSequenceEqual(
            expected_output, list(
                (input == 100).evaluate(False)))

    def test_mask(self):
        input = gr.SeriesWeld(
            np.array(["aaaa", "bbb", "aa", "cccc"], dtype=str),
            gr.WeldVec(gr.WeldChar()))
        predicates = np.array([True, False, False, True], dtype=bool)
        expected_output = ["bbbb", "bbb", "aa", "bbbb"]
        self.assertSequenceEqual(
            expected_output,
            list(
                input.mask(
                    predicates,
                    "bbbb").evaluate(False)))

    def test_filter(self):
        input = gr.SeriesWeld(
            np.array(["aaaa", "bbb", "aa", "cccc"], dtype=str),
            gr.WeldVec(gr.WeldChar()))
        predicates = np.array([True, False, False, True], dtype=bool)
        expected_output = ["aaaa", "cccc"]
        self.assertSequenceEqual(expected_output, list(
            input.filter(predicates).evaluate(False)))

    def test_eq_and_mask(self):
        input = gr.SeriesWeld(
            np.array(["aaaa", "bbb", "aa", "cccc"], dtype=str),
            gr.WeldVec(gr.WeldChar()))
        predicates = input == "bbb"
        expected_output = ["aaaa", "eeee", "aa", "cccc"]
        self.assertSequenceEqual(
            expected_output,
            list(
                input.mask(
                    predicates,
                    "eeee").evaluate(False)))

    def test_slice(self):
        input = gr.SeriesWeld(
            np.array(["aaaaaa", "bbbbbb", "aaaaaaa", "cccccccc"], dtype=str),
            gr.WeldVec(gr.WeldChar()))
        expected_output = ["aaa", "bbb", "aaa", "ccc"]
        self.assertSequenceEqual(
            expected_output, list(
                input.str.slice(
                    1, 3).evaluate(False)))

    def test_add(self):
        input = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        other = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        expected_output = [2, 4, 6, 8]
        self.assertSequenceEqual(
            expected_output, list(
                input.add(other).evaluate(False)))

    def test_sub(self):
        input = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        other = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        expected_output = [0, 0, 0, 0]
        self.assertSequenceEqual(
            expected_output, list(
                input.sub(other).evaluate(False)))

    def test_mul(self):
        input = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        other = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        expected_output = [1, 4, 9, 16]
        self.assertSequenceEqual(
            expected_output, list(
                input.mul(other).evaluate(False)))

    def test_div(self):
        input = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        other = gr.SeriesWeld(
            np.array([1, 2, 3, 4], dtype=np.int32), gr.WeldInt())
        expected_output = [1, 1, 1, 1]
        self.assertSequenceEqual(
            expected_output, list(
                input.div(other).evaluate(False)))

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

max_stars_repo_name: hvanhovell/weld

max_issues_repo_name: mateiz/weld

max_forks_repo_name: mateiz/weld


text:
https://pythonhosted.org/scikit-fuzzy/
https://www.guru99.com/numpy-tutorial.html

import numpy as np

st=np.dtype([('name','S20'),('age','i1')])
a= np.array([('abc',20)], dtype=st)
print(a)

#zeros and ones
np.zeros((2,2))

np.zeros((2,2), dtype=np.int16)

np.ones((2,2))

#reshaping and flatten
e  = np.array([(1,2,3), (4,5,6)])
print(e)
e.reshape(3,2)
e.flatten()

## Horitzontal Stack

f = np.array([1,2,3])
g = np.array([4,5,6])

print('Horizontal Append:', np.hstack((f, g)))

f = np.array([1,2,3])
g = np.array([4,5,6])

print('Vertical Append:', np.vstack((f, g)))

## Generate random nmber from normal distribution
#Loc: the mean. The center of distribution
#scale: standard deviation.
#Size: number of returns

normal_array = np.random.normal(5, 0.5, 10)
print(normal_array)

#Using asarray
A = np.matrix(np.ones((4,4)))
print(A)
np.array(A)[2]=2 #Matrices are immutable
print(A)
np.asarray(A)[2]=2
print(A)

#numpy.arange

Start: Start of interval
Stop: End of interval
Step: Spacing between values. Default step is 1
np.arange(1, 11)
np.arange(1, 14, 4)


text:
import numpy as np
import torch

from torchvision import transforms

class Augmentation(object):
    """Augment gerste patches from uv_dataloader.
    """

    def __call__(self, sample):
        images, label = sample
        random_fliplr = transforms.RandomApply([np.fliplr], p=0.5)
        random_flipud = transforms.RandomApply([np.flipud], p=0.5)
        random_rot180 = transforms.RandomApply([np.rot90, np.rot90], p=0.5)
        transformations = transforms.Compose([
            random_fliplr,
            random_flipud,
            random_rot180
        ])

        augmented_batches = []
        for image in images:
            img = transformations(image)
            augmented_batches.append(torch.from_numpy(img.copy().transpose((2, 0, 1))))

        return torch.stack(augmented_batches), label

class AugmentationSingle(object):
    """Augment gerste image from uv_dataloader.
    """

    def __call__(self, sample):
        image, label = sample
        random_fliplr = transforms.RandomApply([np.fliplr], p=0.5)
        random_flipud = transforms.RandomApply([np.flipud], p=0.5)
        random_rot180 = transforms.RandomApply([np.rot90, np.rot90], p=0.5)
        transformations = transforms.Compose([
            random_fliplr,
            random_flipud,
            random_rot180
        ])

        img = transformations(image)

        return torch.from_numpy(img.copy().transpose((2, 0, 1))), label


text:
"""package for GW Instek power supply GPDx303S"""

import time
import serial
import numpy as np

__author__ = "Yuta Kawai <pygo3xmdy11u@gmail.com>"
__status__ = "production"
__version__ = "0.0.0"
__date__ = "2021/09/02"

class GWInstekGPDx303S:
    def __init__(self, port):
        """establishe communication with the GWInstek GPDx303S

        Parameters
        ----------
        port : str
            port of GWInstek GPDx303S
        """
        self.port = port
        self.instrument = None
        self.current_step = 0.05
        self.voltage_step = 0.05
        self.time_interval = 0.05
        self.baudrate = 9600
        self.time_out = 1.0
        self.connect(port=self.port)

    def __del__(self):
        """close communication with the GWInstek GPDx303S if the program shutdown
        suddenly by error"""
        self.close()

    def set_current_step(self, current_step):
        """set current step

        Parameters
        ----------
        current_step : float
            current step when changing the current value of the power supply
            gradually
        """
        self.current_step = current_step

    def get_current_step(self):
        """get current step

        Returns
        -------
        float
            current step when changing the current value of the power supply
            gradually
        """
        return self.current_step

    def set_voltage_step(self, voltage_step):
        """set voltage step

        Parameters
        ----------
        voltage_step : float
            voltage step when changing the voltage value of the power supply
            gradually
        """
        self.voltage_step = voltage_step

    def get_voltage_step(self):
        """get voltage step

        Returns
        -------
        float
            voltage step when changing the voltage value of the power supply
            gradually
        """
        return self.voltage_step

    def set_time_interval(self, time_interval):
        """set time interval

        Parameters
        ----------
        time_interval : float
            time interval of command
        """
        self.time_interval = time_interval

    def get_time_interval(self):
        """get time interval

        Returns
        -------
        float
            time interval of command
        """
        return self.time_interval

    def set_time_out(self, time_out):
        """set time out

        Parameters
        ----------
        time_out : float
            The time taken to forcibly terminate a process or data transfer in the
            middle when it is taking too long.
        """
        self.time_out = time_out

    def get_time_out(self):
        """get time out

        Returns
        -------
        float
            The time taken to forcibly terminate a process or data transfer in the
            middle when it is taking too long.
        """
        return self.time_out

    def connect(self, port):
        """establishe communication with the GWInstek GPDx303S

        Parameters
        ----------
        port : str
            port of GWInstek GPDx303S
        """
        if port != self.port:
            if self.instrument != None:
                self.close()
            self.port = port
        if self.instrument == None:
            self.instrument = serial.Serial(
                port=port, baudrate=self.baudrate, timeout=self.time_out
            )
            self.REMOTE()

    def close(self):
        """close communication with the GWInstek GPDx303S"""
        if self.instrument != None:
            self.instrument.close()
            self.instrument = None

    def write_command(self, command):
        """write command to the GWInstek GPDx303S

        Parameters
        ----------
        command : str
            command sent to GWInstek GPDx303S
        """
        command = "{}\r\n".format(command)
        self.instrument.write(command.encode("utf-8"))

    def read_command(self):
        """read command to the GWInstek GPDx303S

        Returns
        -------
        str
            string returned from GWInstek GPDx303S
        """
        return str(self.instrument.read_until())[2:-5]

    def query_command(self, command):
        """query command to the GWInstek GPDx303S

        Parameters
        ----------
        command : str
            command sent to GWInstek GPDx303S

        Returns
        -------
        str
            string returned from GWInstek GPDx303S
        """
        self.write_command(command)
        time.sleep(self.time_interval)
        return self.read_command()

    def get_output_status(self):
        """get output status of the GWInstek GPDx303S

        Returns
        -------
        int
            output status of the GWInstek GPDx303S
            (0: OFF, 1: ON)
        """
        return int(self.STATUS()[5])

    def on(self):
        """set output ON"""
        self.OUT(1)

    def off(self):
        """set output OFF"""
        self.OUT(0)

    def set_current(self, channel, current):
        """set current

        Parameters
        ----------
        channel : int
            channel to change the value of the current
        current : float
            value of current
        """
        self.ISET(channel, current)

    def set_voltage(self, channel, voltage):
        """set voltage

        Parameters
        ----------
        channel : int
            channel to change the value of the voltage
        current : float
            value of voltage
        """
        self.VSET(channel, voltage)

    def set_current_quickly(self, channel, current):
        """set current quickly

        Parameters
        ----------
        channel : int
            channel to change the value of the current
        current : float
            value of current
        """
        self.ISET(channel, current)

    def set_voltage_quickly(self, channel, voltage):
        """set voltage quickly

        Parameters
        ----------
        channel : int
            channel to change the value of the voltage
        current : float
            value of voltage
        """
        self.VSET(channel, voltage)

    def set_current_slowly(self, channel, current):
        """set current slowly

        Parameters
        ----------
        channel : int
            channel to change the value of the current
        current : float
            value of current
        """
        if self.get_output_status() == 1:
            present_current = self.IGET(channel)
            if present_current > current:
                current_list = np.arange(present_current, current, -self.current_step)
            else:
                current_list = np.arange(present_current, current, self.current_step)
            for current_value in current_list:
                self.ISET(channel, round(current_value, 3))
                time.sleep(self.time_interval)
        self.ISET(channel, current)

    def set_voltage_slowly(self, channel, voltage):
        """set voltage slowly

        Parameters
        ----------
        channel : int
            channel to change the value of the voltage
        current : float
            value of voltage
        """
        if self.get_output_status() == 1:
            present_voltage = self.VGET(channel)
            if present_voltage > voltage:
                voltage_list = np.arange(present_voltage, voltage, -self.voltage_step)
            else:
                voltage_list = np.arange(present_voltage, voltage, self.voltage_step)
            for voltage_value in voltage_list:
                self.VSET(channel, round(voltage_value, 3))
                time.sleep(self.time_interval)
        self.VSET(channel, voltage)

    def turn_off(self):
        """turn off the power supply"""
        self.idn = self.IDN()
        if self.idn[14:18] == "3303":
            channel_list = (1, 2)
        elif self.idn[14:18] == "4303":
            channel_list = (1, 2, 3, 4)
        for ch in channel_list:
            self.set_voltage_slowly(ch, 0.0)
        self.Off()

    def ISET(self, channel, current):
        """set the value of current

        Parameters
        ----------
        channel : int
            channel to change the value of the current
        current : float
            value of current
        """
        command = "ISET{0}:{1}".format(channel, round(current, 3))
        self.write_command(command)

    def IGET(self, channel):
        """get the value of current

        Returns
        -------
        float
            value of current
        """
        command = "ISET{}?".format(channel)
        return float(self.query_command(command)[:-1])

    def VSET(self, channel, voltage):
        """set the value of voltage

        Parameters
        ----------
        channel : int
            channel to change the value of the voltage
        voltage : float
            value of voltage
        """
        command = "VSET{0}:{1}".format(channel, round(voltage, 3))
        self.write_command(command)

    def VGET(self, channel):
        """get the value of voltage

        Returns
        -------
        float
            value of voltage
        """
        command = "VSET{}?".format(channel)
        return float(self.query_command(command)[:-1])

    def IOUT(self, channel):
        """get actual output current

        Parameters
        ----------
        channel : int
            channel to change the value of the current
        current : float
            value of actual output current
        """
        command = "IOUT{}?".format(channel)
        return float(self.query_command(command)[:-1])

    def VOUT(self, channel):
        """get actual output voltage

        Parameters
        ----------
        channel : int
            channel to change the value of the voltage
        voltage : float
            value of actual output voltage
        """
        command = "VOUT{}?".format(channel)
        return float(self.query_command(command))

    def TRACK(self, mode):
        """set the output of the power supply working on independent or tracking mode

        Parameters
        ----------
        mode : int
            tracking mode of GWInstek GPDx303S
            (0: independence, 1: series tracking, 2: parallel tracking)
        """
        command = "TRACK{}".format(mode)
        self.write_command(command)

    def BEEP(self, io):
        """set the BEEP state on or off

        Parameters
        ----------
        io : int
            BEEP state of the GWInstek GPDx303S
            (0: OFF, 1: ON)
        """
        command = "BEEP{}".format(io)
        self.write_command(command)

    def OUT(self, io):
        """Set the output state on or off

        Parameters
        ----------
        io : int
            output state of the GWInstek GPDx303S
            (0: OFF, 1: ON)
        """
        command = "OUT{}".format(io)
        self.write_command(command)

    def STATUS(self):
        """get the power supply state

        Returns
        -------
        str
            state of GWInstek GPDx303S
        """
        command = "STATUS?"
        return self.query_command(command)

    def IDN(self):
        """get instrument identification

        Returns
        -------
        str
            instrument identification
        """
        command = "*IDN?"
        return self.query_command(command)

    def RCL(self, memory_number):
        """recall the setting data from the memory which previous saved

        Parameters
        ----------
        memory_number : int
            number of memory which previous saved
        """
        command = "RCL{}".format(memory_number)
        self.write_command(command)

    def SAV(self, memory_number):
        """save the setting data to memory

        Paremeters
        ----------
        memory_number : int
            number of memory which you want to save the setting data to
        """
        command = "SAV{}".format(memory_number)
        self.write_command(command)

    def BAUD(self, baudrate):
        """set the value of baud rate

        Parameters
        ----------
        baudrate : int
            baud rate(0: 115200, 1: 57600, 2: 9600)
        """
        command = "BAUD{}".format(baudrate)
        self.write_command(command)

    def LOCAL(self):
        """return to local mode"""
        command = "LOCAL"
        self.write_command(command)

    def REMOTE(self):
        """return to remote mode"""
        command = "REMOTE"
        self.write_command(command)

    def ERR(self):
        """get instrument error messages

        Returns
        -------
        str
            error messages
        """
        command = "ERR?"
        return self.query_command(command)

    def HELP(self):
        """get command list

        Returns
        -------
        str
            command list
        """
        command = "HELP?"
        return self.query_command(command)


text:
import numpy as np
from rand_param_envs.gym import utils
from rand_param_envs.gym.envs.mujoco import mujoco_env

class HalfCheetahEnv(mujoco_env.MujocoEnv, utils.EzPickle):
    def __init__(self):
        mujoco_env.MujocoEnv.__init__(self, 'half_cheetah.xml', 5)
        utils.EzPickle.__init__(self)

    def _step(self, action):
        xposbefore = self.model.data.qpos[0, 0]
        self.do_simulation(action, self.frame_skip)
        xposafter = self.model.data.qpos[0, 0]
        ob = self._get_obs()
        reward_ctrl = - 0.1 * np.square(action).sum()
        reward_run = (xposafter - xposbefore)/self.dt
        reward = reward_ctrl + reward_run
        done = False
        return ob, reward, done, dict(reward_run=reward_run, reward_ctrl=reward_ctrl)

    def _get_obs(self):
        return np.concatenate([
            self.model.data.qpos.flat[1:],
            self.model.data.qvel.flat,
        ])

    def reset_model(self):
        qpos = self.init_qpos + self.np_random.uniform(low=-.1, high=.1, size=self.model.nq)
        qvel = self.init_qvel + self.np_random.randn(self.model.nv) * .1
        self.set_state(qpos, qvel)
        return self._get_obs()

    def viewer_setup(self):
        self.viewer.cam.distance = self.model.stat.extent * 0.5

max_stars_repo_name: erinaldi/MetaRL

max_issues_repo_name: erinaldi/MetaRL

max_forks_repo_name: erinaldi/MetaRL


text:
#
# A module to parse e-mail text for a plankton word
# input is the e-mail text sans spaces,tabs and newlines
# input is the number of words in the e-mail
# input is the actual word to find this is currently a strict match (case counts)
# The function returns the word index of the plankton word or an empty list
# The code will NOT catch misspellings.
import re
import numpy as np
import pdb
def find_plankton_word(mytext,myword,plankton):
    """ find_plankton_word(mytext,myword,plankton) module to find a particular plankton word
    in the e-mail payload as both the word and the whole text.
    mytext is the whole payload of the e-mail without breaking it up so we can count characters.
    myword is the words in the payload that we want to match against.  The output is the index
    where the word occurrs.
    """
# def find_plankton_word(emailtext,plankton):
# mytext=emailtext.replace(" ","")
# myword=emailtext.split()
    #mytext=mytext.replace("-","")
# to make case insensitive
# plankton=plankton.lower()
    # Code could be made to do the remove spaces and parse words....
    # mytext is the e-mail message with spaces and some special characters removed
    # myword is the e-mail split into words based on space/tab/newline
    # plankton is the plankton word to search for (i.e. 'Akashiwo')
    # this code so far only works for single word searches....
    pword=plankton.split()
    lp=len(pword)
    wordstart=[]
    wordstop=[]
    wordind=[]
    iflag=0
    nospacep=plankton.replace(" ","")
    # need to find multiple word plankton word
    for k in re.finditer(nospacep,mytext):
#    for k in re.finditer(plankton,mytext):
        wordstart.append(k.start())
        wordstop.append(k.end())
        iflag=1
    if iflag==0:
        # no match
        return wordind
    lko=np.arange(0,len(myword)) # total number of "words" in the e-mail
    lw=np.arange(0,len(wordstart)) # number of instances of the plankton word we need to find
    for myi in lw:
        msum=0
        for j in lko:
            msum=msum+len(myword[j]) # sum up the lenght of words to find which index
            #print str(msum)+' '+str(j)
            #print myword[j]
            # j corresponds to the text index location
            if int(msum) >= wordstart[myi]:
                if msum==wordstart[myi]:
                    wordind.append(j+1)
                else:
                    wordind.append(j)
                break # this should only work on the inner for loop and not the outer
    # okay we should have the index of all words that match the plankton word sent in
    # for some reason we are off by one for words that start with "-" why is this?
    #pdb.set_trace()
    return wordind

max_stars_repo_name: CeNCOOS/habs-email

max_issues_repo_name: CeNCOOS/habs-email

max_forks_repo_name: CeNCOOS/habs-email


text:
import networkx as nx
import matplotlib.pyplot as plt

def plot(G, components):
    # plotting
    node_group = [-1 for _ in G.nodes]
    for group_ID, component in enumerate(components):
        for u in component:
            node_group[u-1] = group_ID
    # print(node_group)
    plt.plot()
    pos = nx.spring_layout(G)
    # refer to networkx docs for more drawing parameters
    nx.draw_networkx(G, pos, with_labels=True, connectionstyle="arc3,rad=0.1")
    nx.draw_networkx_nodes(G, pos, node_color=node_group,
                           cmap=plt.cm.Set1, node_size=300)
    # nx.draw_networkx_edges(G, pos, arrows=True, arrowsize=20, width=3, alpha=1)
    plt.show()

max_stars_repo_name: ChiragKr/python-algorithms

max_issues_repo_name: ChiragKr/python-algorithms

max_forks_repo_name: ChiragKr/python-algorithms

reason: import networkx


text:
# Copyright 2019 the ProGraML authors.
#
# Contact Chris Cummins <chrisc.101@gmail.com>.
#
# 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.
"""Utilities for working with LLVM statements."""
import difflib
import re
import typing

import networkx as nx

from deeplearning.ml4pl.graphs import nx_utils
from deeplearning.ml4pl.graphs import programl_pb2
from deeplearning.ncc import rgx_utils as rgx
from deeplearning.ncc.inst2vec import inst2vec_preprocess
from labm8.py import app

FLAGS = app.FLAGS

def GetAllocationStatementForIdentifier(g: nx.Graph, identifier: str) -> str:
  for node, data in g.nodes(data=True):
    if data["type"] != programl_pb2.Node.STATEMENT:
      continue
    if " = alloca " in data["text"]:
      allocated_identifier = data["text"].split(" =")[0]
      if allocated_identifier == identifier:
        return node
  raise ValueError(
    f"Unable to find `alloca` statement for identifier `{identifier}`"
  )

def StripIdentifiersAndImmediates(stmt: str) -> str:
  """This is a copy of inst2vec_preprocess.PreprocessStatement(), but instead
  of substituting placeholders values, immediates and labels are removed
  entirely from the string.
  """
  # Remove local identifiers
  stmt = re.sub(rgx.local_id, "", stmt)
  # Global identifiers
  stmt = re.sub(rgx.global_id, "", stmt)
  # Remove labels
  if re.match(r"; <label>:\d+:?(\s+; preds = )?", stmt):
    stmt = re.sub(r":\d+", ":", stmt)
  elif re.match(rgx.local_id_no_perc + r":(\s+; preds = )?", stmt):
    stmt = re.sub(rgx.local_id_no_perc + ":", ":", stmt)

  # Remove floating point values
  stmt = re.sub(rgx.immediate_value_float_hexa, "", stmt)
  stmt = re.sub(rgx.immediate_value_float_sci, "", stmt)

  # Remove integer values
  if (
    re.match("<%ID> = extractelement", stmt) is None
    and re.match("<%ID> = extractvalue", stmt) is None
    and re.match("<%ID> = insertelement", stmt) is None
    and re.match("<%ID> = insertvalue", stmt) is None
  ):
    stmt = re.sub(r"(?<!align)(?<!\[) " + rgx.immediate_value_int, " ", stmt)

  # Remove string values
  stmt = re.sub(rgx.immediate_value_string, " ", stmt)

  # Remove index types
  if (
    re.match(" = extractelement", stmt) is not None
    or re.match(" = insertelement", stmt) is not None
  ):
    stmt = re.sub(r"i\d+ ", " ", stmt)

  return stmt

def GetLlvmStatementDefAndUses(
  statement: str,
) -> typing.Tuple[str, typing.List[str]]:
  """Get the destination identifier for an LLVM statement (if any), and a list
  of operand identifiers (if any).
  """
  # Left hand side.
  destination = ""
  if "=" in statement:
    first_equals = statement.index("=")
    destination = statement[:first_equals]
    statement = statement[first_equals:]

  # Strip the identifiers and immediates from the statement, then use the
  # diff to construct the set of identifiers and immediates that were stripped.
  stripped = StripIdentifiersAndImmediates(statement)
  tokens = []

  last_token = []
  last_index = -1
  for i, diff in enumerate(difflib.ndiff(statement, stripped)):
    if diff[0] == "-":
      if i != last_index + 1 and last_token:
        tokens.append("".join(last_token))
        last_token = []

      last_token.append(diff[-1])
      last_index = i

  if last_token:
    tokens.append("".join(last_token))

  return destination.strip(), tokens

def GetCalledFunctionName(statement) -> typing.Optional[str]:
  """Get the name of a function called in the statement."""
  if "call " not in statement:
    return None
  # Try and resolve the call destination.
  _, m_glob, _, _ = inst2vec_preprocess.get_identifiers_from_line(statement)
  if not m_glob:
    return None
  return m_glob[0][1:]  # strip the leading '@' character

def FindCallSites(graph, source_function, destination_function):
  """Find the statements in function that call another function."""
  call_sites = []
  for node, data in nx_utils.StatementNodeIterator(graph):
    if data["function"] != source_function:
      continue
    called_function = GetCalledFunctionName(data["text"])
    if not called_function:
      continue
    if called_function == destination_function:
      call_sites.append(node)
  return call_sites


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

"""Models for Text and Image Composition."""

import numpy as np
import torch
import torchvision
import torch.nn.functional as F
import text_model
import torch_functions

class ConCatModule(torch.nn.Module):

  def __init__(self):
    super(ConCatModule, self).__init__()

  def forward(self, x):
    x = torch.cat(x, dim=1)
    return x

class ImgTextCompositionBase(torch.nn.Module):
  """Base class for image + text composition."""

  def __init__(self):
    super(ImgTextCompositionBase, self).__init__()
    self.normalization_layer = torch_functions.NormalizationLayer(
        normalize_scale=4.0, learn_scale=True)
    self.soft_triplet_loss = torch_functions.TripletLoss()

  def extract_img_feature(self, imgs):
    raise NotImplementedError

  def extract_text_feature(self, texts, vocab):
    raise NotImplementedError

  def compose_img_text(self, imgs, texts):
    raise NotImplementedError

  def compute_loss(self,
                   imgs_query,
                   modification_texts,
                   imgs_target,
                   soft_triplet_loss=True):
    mod_img1 = self.compose_img_text(imgs_query, modification_texts)
    mod_img1 = self.normalization_layer(mod_img1)
    img2 = self.extract_img_feature(imgs_target)
    img2 = self.normalization_layer(img2)
    assert (mod_img1.shape[0] == img2.shape[0] and
            mod_img1.shape[1] == img2.shape[1])
    if soft_triplet_loss:
      # return self.compute_soft_triplet_loss_(mod_img1, img2)
      return self.compute_batch_based_classification_loss_(mod_img1, img2)
    else:
      return self.compute_batch_based_classification_loss_(mod_img1, img2)

  def compute_soft_triplet_loss_(self, mod_img1, img2):
    triplets = []
    labels = list(range(mod_img1.shape[0])) + list(range(img2.shape[0]))
    for i in range(len(labels)):
      triplets_i = []
      for j in range(len(labels)):
        if labels[i] == labels[j] and i != j:
          for k in range(len(labels)):
            if labels[i] != labels[k]:
              triplets_i.append([i, j, k])
      np.random.shuffle(triplets_i)
      triplets += triplets_i[:3]
    assert (triplets and len(triplets) < 2000)
    return self.soft_triplet_loss(torch.cat([mod_img1, img2]), triplets)

  def compute_batch_based_classification_loss_(self, mod_img1, img2):
    x = torch.mm(mod_img1, img2.transpose(0, 1))
    labels = torch.tensor(range(x.shape[0])).long()
    labels = labels.to(x.device)
    return F.cross_entropy(x, labels)

class ImgEncoderTextEncoderBase(ImgTextCompositionBase):
  """Base class for image and text encoder."""

  def __init__(self, vocab_size, embed_dim=512):
    super(ImgEncoderTextEncoderBase, self).__init__()
    # img model
    img_model = torchvision.models.resnet18(pretrained=True)

    class GlobalAvgPool2d(torch.nn.Module):

      def forward(self, x):
        return F.adaptive_avg_pool2d(x, (1, 1))

    img_model.avgpool = GlobalAvgPool2d()
    img_model.fc = torch.nn.Sequential(torch.nn.Linear(512, embed_dim))
    self.img_model = img_model

    # text model
    self.text_model = text_model.TextLSTMModel(
        vocab_size,
        word_embed_dim=embed_dim,
        lstm_hidden_dim=embed_dim)

  def extract_img_feature(self, imgs):
    return self.img_model(imgs)

  def extract_text_feature(self, texts, lengths):
    return self.text_model(texts, lengths)

class SimpleModelImageOnly(ImgEncoderTextEncoderBase):

  def compose_img_text(self, imgs, texts):
    return self.extract_img_feature(imgs)

class SimpleModelTextOnly(ImgEncoderTextEncoderBase):

  def compose_img_text(self, imgs, texts):
    return self.extract_text_feature(texts)

class Concat(ImgEncoderTextEncoderBase):
  """Concatenation model."""

  def __init__(self, texts_to_build_vocab=None, built_vocab=None, embed_dim=512):
    super(Concat, self).__init__(texts_to_build_vocab=None, built_vocab=None, embed_dim=512)

    # composer
    class Composer(torch.nn.Module):
      """Inner composer class."""

      def __init__(self):
        super(Composer, self).__init__()
        self.m = torch.nn.Sequential(
            torch.nn.BatchNorm1d(2 * embed_dim), torch.nn.ReLU(),
            torch.nn.Linear(2 * embed_dim, 2 * embed_dim),
            torch.nn.BatchNorm1d(2 * embed_dim), torch.nn.ReLU(),
            torch.nn.Dropout(0.1), torch.nn.Linear(2 * embed_dim, embed_dim))

      def forward(self, x):
        f = torch.cat(x, dim=1)
        f = self.m(f)
        return f

    self.composer = Composer()

  def compose_img_text(self, imgs, texts):
    img_features = self.extract_img_feature(imgs)
    text_features = self.extract_text_feature(texts)
    return self.compose_img_text_features(img_features, text_features)

  def compose_img_text_features(self, img_features, text_features):
    return self.composer((img_features, text_features))

class TIRG(ImgEncoderTextEncoderBase):
  """The TIGR model.
  The method is described in
  Nam Vo, Lu Jiang, Chen Sun, Kevin Murphy, Li-Jia Li, Li Fei-Fei, James Hays.
  "Composing Text and Image for Image Retrieval - An Empirical Odyssey"
  CVPR 2019. arXiv:1812.07119
  """

  def __init__(self, vocab_size, embed_dim=512):
    super(TIRG, self).__init__(vocab_size, embed_dim=512)

    self.a = torch.nn.Parameter(torch.tensor([1.0, 10.0, 1.0, 1.0]))
    self.gated_feature_composer = torch.nn.Sequential(
        ConCatModule(), torch.nn.BatchNorm1d(2 * embed_dim), torch.nn.ReLU(),
        torch.nn.Linear(2 * embed_dim, embed_dim))
    self.res_info_composer = torch.nn.Sequential(
        ConCatModule(), torch.nn.BatchNorm1d(2 * embed_dim), torch.nn.ReLU(),
        torch.nn.Linear(2 * embed_dim, 2 * embed_dim), torch.nn.ReLU(),
        torch.nn.Linear(2 * embed_dim, embed_dim))

  def compose_img_text(self, imgs, itexts, lengths):
    img_features = self.extract_img_feature(imgs)
    text_features = self.extract_text_feature(itexts, lengths)
    return self.compose_img_text_features(img_features, text_features)

  def compose_img_text_features(self, img_features, text_features):
    f1 = self.gated_feature_composer((img_features, text_features))
    f2 = self.res_info_composer((img_features, text_features))
    f = F.sigmoid(f1) * img_features * self.a[0] + f2 * self.a[1]
    return f

class TIRGLastConv(ImgEncoderTextEncoderBase):
  """The TIGR model with spatial modification over the last conv layer.
  The method is described in
  Nam Vo, Lu Jiang, Chen Sun, Kevin Murphy, Li-Jia Li, Li Fei-Fei, James Hays.
  "Composing Text and Image for Image Retrieval - An Empirical Odyssey"
  CVPR 2019. arXiv:1812.07119
  """

  def __init__(self, texts_to_build_vocab=None, built_vocab=None, embed_dim=512):
    super(TIRGLastConv, self).__init__(texts_to_build_vocab=None, built_vocab=None, embed_dim=512)

    self.a = torch.nn.Parameter(torch.tensor([1.0, 10.0, 1.0, 1.0]))
    self.mod2d = torch.nn.Sequential(
        torch.nn.BatchNorm2d(512 + embed_dim),
        torch.nn.Conv2d(512 + embed_dim, 512 + embed_dim, [3, 3], padding=1),
        torch.nn.ReLU(),
        torch.nn.Conv2d(512 + embed_dim, 512, [3, 3], padding=1),
    )

    self.mod2d_gate = torch.nn.Sequential(
        torch.nn.BatchNorm2d(512 + embed_dim),
        torch.nn.Conv2d(512 + embed_dim, 512 + embed_dim, [3, 3], padding=1),
        torch.nn.ReLU(),
        torch.nn.Conv2d(512 + embed_dim, 512, [3, 3], padding=1),
    )

  def compose_img_text(self, imgs, texts):
    text_features = self.extract_text_feature(texts)

    x = imgs
    x = self.img_model.conv1(x)
    x = self.img_model.bn1(x)
    x = self.img_model.relu(x)
    x = self.img_model.maxpool(x)

    x = self.img_model.layer1(x)
    x = self.img_model.layer2(x)
    x = self.img_model.layer3(x)
    x = self.img_model.layer4(x)

    # mod
    y = text_features
    y = y.reshape((y.shape[0], y.shape[1], 1, 1)).repeat(
        1, 1, x.shape[2], x.shape[3])
    z = torch.cat((x, y), dim=1)
    t = self.mod2d(z)
    tgate = self.mod2d_gate(z)
    x = self.a[0] * F.sigmoid(tgate) * x + self.a[1] * t

    x = self.img_model.avgpool(x)
    x = x.view(x.size(0), -1)
    x = self.img_model.fc(x)
    return x


text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Sun Jan 23 22:19:32 2022

@author: baecker
"""

import sys
sys.path.append('./napari_j/_tests/surrogate')
if __name__ == '__main__':
    from surrogate.surrogate import surrogate
else:
    from surrogate import surrogate
from unittest.mock import patch
from unittest.mock import Mock, MagicMock
import napari
import numpy as np

def getImage():
    imageMock = MagicMock()
    imageMock.getDimensions.return_value = [3, 2, 1, 1, 1]
    imageMock.getShortTitle.return_value = 'blobs'
    stackMock = MagicMock()
    stackMock.getVoxels.return_value = [255.0, 0.0 ,128.0, 0.0, 64.0, 32.0]
    imageMock.getStack.return_value = stackMock
    calibrationMock = MagicMock()
    calibrationMock.getZ.return_value = 2.5
    calibrationMock.getX.return_value = 1
    imageMock.getCalibration.return_value = calibrationMock
    return imageMock

IJMock = Mock()
IJMock.getImage = getImage
HyperStackConverterMock = Mock()

@patch('napari.Viewer')
@surrogate('ij.measure.ResultsTable')
@surrogate('ij.IJ')
@patch('ij.IJ', IJMock)
@surrogate('ij.ImagePlus')
@surrogate('ij.WindowManager')
@surrogate('ij.plugin.HyperStackConverter')
@patch('ij.plugin.HyperStackConverter', HyperStackConverterMock)
def test_constructor(Viewer):
    if __name__ == '__main__':
        from bridge import Bridge
    else:
        from ..bridge import Bridge
    viewer = napari.Viewer()
    bridge = Bridge(viewer)
    assert(bridge.viewer==viewer)

@patch('napari.Viewer')
@surrogate('ij.measure.ResultsTable')
@surrogate('ij.IJ')
@patch('ij.IJ', IJMock)
@surrogate('ij.ImagePlus')
@surrogate('ij.WindowManager')
@surrogate('ij.plugin.HyperStackConverter')
@patch('ij.plugin.HyperStackConverter', HyperStackConverterMock)
def test_getActiveImageFromIJ(Viewer):
    if __name__ == '__main__':
        from bridge import Bridge
    else:
        from ..bridge import Bridge
    viewer = napari.Viewer()
    bridge = Bridge(viewer)
    bridge.getActiveImageFromIJ()

    # Display should be set to 3D
    assert(viewer.dims.ndisplay==3)

    # The voxel data should be rearranged in the form expected by napari
    actual = viewer.add_image.call_args[0]
    expected = np.array([[[[255.0, 0.0, 128.0], [0.0, 64.0, 32.0]]]])
    comparison = actual == expected
    assert(comparison.all())

    # The name should be C<channel-nr.>-<short title of the ij-image>
    assert(viewer.add_image.call_args[1]['name']=='C1-blobs')

    # The lut should be the first (or n-th) color defined in Bridge
    assert(viewer.add_image.call_args[1]['colormap']==bridge.colors[0])

    # The default blending should be additive
    assert(viewer.add_image.call_args[1]['blending']=='additive')

    # The z-scale should have been calculated as the ratio of the z-step and
    # the x-pixel size.
    assert(viewer.add_image.call_args[1]['scale'][0]==2.5)
    assert(viewer.add_image.call_args[1]['scale'][1]==1)
    assert(viewer.add_image.call_args[1]['scale'][2]==1)

    # All layers are removed from napari before an image is fetched from ij.
    viewer.layers.__len__.return_value = 1
    bridge.getActiveImageFromIJ()
    bridge.viewer.layers.pop.assert_called_once()

@patch('napari.Viewer')
@surrogate('ij.measure.ResultsTable')
@surrogate('ij.IJ')
@patch('ij.IJ', IJMock)
@surrogate('ij.ImagePlus')
@surrogate('ij.WindowManager')
@surrogate('ij.plugin.HyperStackConverter')
@patch('ij.plugin.HyperStackConverter', HyperStackConverterMock)
def test_getLabelsFromIJ(Viewer):
    if __name__ == '__main__':
        from bridge import Bridge
    else:
        from ..bridge import Bridge
    viewer = napari.Viewer()
    bridge = Bridge(viewer)
    bridge.getLabelsFromIJ()

    # The image data should be of type int, independent from the image type in ij.
    actual = viewer.add_labels.call_args[0]
    expected = np.array([[[[255, 0, 128], [0, 64, 32]]]])
    comparison = actual == expected
    assert(comparison.all())
    assert(isinstance(actual[0][0][0][0][0], np.int_))

    # The name should be short title of the ij-image
    assert(viewer.add_labels.call_args[1]['name']=='blobs')

    # The z-scale should have been calculated as the ratio of the z-step and
    # the x-pixel size.
    assert(viewer.add_labels.call_args[1]['scale'][0]==2.5)
    assert(viewer.add_labels.call_args[1]['scale'][1]==1)
    assert(viewer.add_labels.call_args[1]['scale'][2]==1)

@patch('napari.Viewer')
@surrogate('ij.measure.ResultsTable')
@surrogate('ij.IJ')
@patch('ij.IJ', IJMock)
@surrogate('ij.ImagePlus')
@surrogate('ij.WindowManager')
@surrogate('ij.plugin.HyperStackConverter')
@patch('ij.plugin.HyperStackConverter', HyperStackConverterMock)
def test_getPixelsFromImageJ(Viewer):
    if __name__ == '__main__':
        from bridge import Bridge
    else:
        from ..bridge import Bridge
    viewer = napari.Viewer()
    bridge = Bridge(viewer)
    title, dims, zFactor, pixels = bridge.getPixelsFromImageJ()

    # The title should be the short-title of the image in ImageJ
    assert(title=='blobs')

    # The test image has a width of 3 pixels, a height of 2 pixels, one channel,
    # one z-slice and one time-frame.
    expected = [3, 2, 1, 1, 1]
    comparison = np.array(dims) == np.array(expected)
    assert(comparison.all())

    # The z-scale should have been calculated as the ratio of the z-step and
    # the x-pixel size.
    assert(zFactor==2.5)

    # The pixel data is returned as a linear list with the order of dimensions
    # given by dim, i.e. xyczt.
    expected = np.array([255.0, 0.0, 128.0, 0.0, 64.0, 32.0])
    comparison = pixels == expected
    assert(comparison.all())

@patch('napari.Viewer')
@surrogate('ij.measure.ResultsTable')
@surrogate('ij.IJ')
@patch('ij.IJ', IJMock)
@surrogate('ij.ImagePlus')
@surrogate('ij.WindowManager')
@surrogate('ij.plugin.HyperStackConverter')
@patch('ij.plugin.HyperStackConverter', HyperStackConverterMock)
def test_getMetadataFromImage(Viewer):
    if __name__ == '__main__':
        from bridge import Bridge
    else:
        from ..bridge import Bridge
    viewer = napari.Viewer()
    bridge = Bridge(viewer)
    image = getImage()
    title, dims, zFactor, size = bridge.getMetadataFromImage(image)

    # The title should be the short-title of the image in ImageJ
    assert(title=='blobs')

    # The test image has a width of 3 pixels, a height of 2 pixels, one channel,
    # one z-slice and one time-frame.
    expected = [3, 2, 1, 1, 1]
    comparison = np.array(dims) == np.array(expected)
    assert(comparison.all())

    # The z-scale should have been calculated as the ratio of the z-step and
    # the x-pixel size.
    assert(zFactor==2.5)

    # The size of one channel of the image is the product of the sizes of the
    # remaining dimensions (without c)
    assert(size==dims[0]*dims[1]*dims[3]*dims[4])
    assert(size==6)

@patch('napari.Viewer')
@surrogate('ij.measure.ResultsTable')
@surrogate('ij.IJ')
@patch('ij.IJ', IJMock)
@surrogate('ij.ImagePlus')
@surrogate('ij.WindowManager')
@surrogate('ij.plugin.HyperStackConverter')
@patch('ij.plugin.HyperStackConverter', HyperStackConverterMock)
def test_toHyperstack(Viewer):
    if __name__ == '__main__':
        from bridge import Bridge
    else:
        from ..bridge import Bridge
    viewer = napari.Viewer()
    bridge = Bridge(viewer)
    image = getImage()
    title, dims, zFactor, size = bridge.getMetadataFromImage(image)
    bridge.toHyperstack(image, dims)

    # If the converted image has a different id from the input image, the
    # input image is closed and the new image is shown.
    image.close.assert_called_once()

if __name__ == '__main__':
    test_constructor()
    test_getActiveImageFromIJ()
    test_getLabelsFromIJ()
    test_getPixelsFromImageJ()
    test_getMetadataFromImage()
    test_toHyperstack()

max_stars_repo_name: jaimergp/napari-J

max_issues_repo_name: jaimergp/napari-J

max_forks_repo_name: jaimergp/napari-J


text:
# Copyright (C) 2021 Intel Corporation
#
# 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 importlib
from typing import Iterable, Union
import yaml
import time

import numpy as np

from ote_sdk.entities.train_parameters import UpdateProgressCallback
from ote_sdk.usecases.reporting.time_monitor_callback import TimeMonitorCallback
from ote_sdk.utils.argument_checks import (
    YamlFilePathCheck,
    check_input_parameters_type,
)

@check_input_parameters_type({"path": YamlFilePathCheck})
def load_template(path):
    with open(path) as f:
        template = yaml.safe_load(f)
    return template

@check_input_parameters_type()
def get_task_class(path: str):
    module_name, class_name = path.rsplit('.', 1)
    module = importlib.import_module(module_name)
    return getattr(module, class_name)

@check_input_parameters_type()
def get_activation_map(features: Union[np.ndarray, Iterable, int, float]):
    min_soft_score = np.min(features)
    max_soft_score = np.max(features)
    factor = 255.0 / (max_soft_score - min_soft_score + 1e-12)

    float_act_map = factor * (features - min_soft_score)
    int_act_map = np.uint8(np.floor(float_act_map))

    return int_act_map

class TrainingProgressCallback(TimeMonitorCallback):
    def __init__(self, update_progress_callback: UpdateProgressCallback):
        super().__init__(0, 0, 0, 0, update_progress_callback=update_progress_callback)

    def on_train_batch_end(self, batch, logs=None):
        super().on_train_batch_end(batch, logs)
        self.update_progress_callback(self.get_progress())

    def on_epoch_end(self, epoch, logs=None):
        self.past_epoch_duration.append(time.time() - self.start_epoch_time)
        self._calculate_average_epoch()
        score = None
        if hasattr(self.update_progress_callback, 'metric') and isinstance(logs, dict):
            score = logs.get(self.update_progress_callback.metric, None)
            score = float(score) if score is not None else None
        self.update_progress_callback(self.get_progress(), score=score)

class InferenceProgressCallback(TimeMonitorCallback):
    def __init__(self, num_test_steps, update_progress_callback: UpdateProgressCallback):
        super().__init__(
            num_epoch=0,
            num_train_steps=0,
            num_val_steps=0,
            num_test_steps=num_test_steps,
            update_progress_callback=update_progress_callback)

    def on_test_batch_end(self, batch=None, logs=None):
        super().on_test_batch_end(batch, logs)
        self.update_progress_callback(int(self.get_progress()))

class OptimizationProgressCallback(TrainingProgressCallback):
    """ Progress callback used for optimization using NNCF
        There are three stages to the progress bar:
           - 5 % model is loaded
           - 10 % compressed model is initialized
           - 10-100 % compressed model is being fine-tuned
    """
    def __init__(self, update_progress_callback: UpdateProgressCallback, loading_stage_progress_percentage: int = 5,
                 initialization_stage_progress_percentage: int = 5):
        super().__init__(update_progress_callback=update_progress_callback)
        if loading_stage_progress_percentage + initialization_stage_progress_percentage >= 100:
            raise RuntimeError('Total optimization progress percentage is more than 100%')

        self.loading_stage_progress_percentage = loading_stage_progress_percentage
        self.initialization_stage_progress_percentage = initialization_stage_progress_percentage

        # set loading_stage_progress_percentage from the start as the model is already loaded at this point
        self.update_progress_callback(loading_stage_progress_percentage)

    def on_train_begin(self, logs=None):
        super().on_train_begin(logs)
        # Callback initialization takes place here after OTEProgressHook.before_run() is called
        train_percentage = 100 - self.loading_stage_progress_percentage - self.initialization_stage_progress_percentage
        loading_stage_steps = self.total_steps * self.loading_stage_progress_percentage / train_percentage
        initialization_stage_steps = self.total_steps * self.initialization_stage_progress_percentage / train_percentage
        self.total_steps += loading_stage_steps + initialization_stage_steps

        self.current_step = loading_stage_steps + initialization_stage_steps
        self.update_progress_callback(self.get_progress())

    def on_train_end(self, logs=None):
        super().on_train_end(logs)
        self.update_progress_callback(self.get_progress(), score=logs)

    def on_initialization_end(self):
        self.update_progress_callback(self.loading_stage_progress_percentage +
                                      self.initialization_stage_progress_percentage)