text:
import math
import numpy as np
import subprocess

# BSDFs need more samples and a bigger integration resolution
nsamplehisto = 256000000
res = 4096

# Not enough samples and too low resolution. Use these parameters only to see the BSDF plot
# nsamplehisto = 1000000
# res = 512

alphax = 0.25
alphay = 0.25
rho = 0.

stx = 0.

width = 0.001

thetao = 0.2
phio = 0.

# microfacet relative area
mra = 1.0

outfilenameWithoutExtension = "plotchisquaretestglitteringbsdf"
outfilename = outfilenameWithoutExtension + ".py"

process = subprocess.run(['glinttool', 'chisquaretestglitteringbsdf', '-nsamplehisto', str(nsamplehisto), '-res', str(res), '-alphax', str(alphax), '-alphay', str(alphay), '-rho', str(rho), '-stx', str(stx), '-dstdxx', str(width), '-dstdyy', str(width), '-thetao', str(thetao), '-phio', str(phio), '-mra', str(mra), 'dict_N768_nLevels8.exr',
                          outfilename], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
print(process.stdout)

# Memory error with res = 4096. Too heavy for matplotlib.
# process = subprocess.run(['python', outfilenameWithoutExtension + "_test1.py"],
#                          stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, check=True)
# print(process.stdout)

ext: py

reason: import numpy


text:
from morfist import MixedRandomForest, cross_validation
from sklearn.ensemble import RandomForestRegressor, RandomForestClassifier
from sklearn.model_selection import cross_val_score
from sklearn.datasets import load_boston, load_breast_cancer
import numpy as np

# Config
n_trees = 11

# Data
x_reg, y_reg = load_boston(return_X_y=True)
x_cls, y_cls = load_breast_cancer(return_X_y=True)
x_mix_1, y_mix_1 = x_reg, np.vstack([y_reg, y_reg < y_reg.mean()]).T
x_mix_2, y_mix_2 = x_cls, np.vstack([y_cls, y_cls]).T

def test_class():
	cls_rf = MixedRandomForest(
		n_estimators=n_trees,
		min_samples_leaf=1,
		class_targets=[0]
	)

	cls_skrf = RandomForestClassifier(n_estimators=n_trees)
	cls_scores = cross_validation(
		cls_rf,
		x_cls,
		y_cls,
		class_targets=[0],
		folds=10
	)

	scores = cross_val_score(
		cls_skrf,
		x_cls,
		y_cls
	)

	print('Classification: ')
	print('\tmorfist (accuracy): {}'.format(cls_scores.mean()))
	print('\tscikit-learn (accuracy): {}'.format(scores.mean()))

def test_reg():
	reg_rf = MixedRandomForest(
		n_estimators=n_trees,
		min_samples_leaf=5
	)

	reg_skrf = RandomForestRegressor(n_estimators=n_trees)
	reg_scores = cross_validation(
		reg_rf,
		x_reg,
		y_reg,
		folds=10
	)

	scores = cross_val_score(
		reg_skrf,
		x_reg,
		y_reg,
		scoring='neg_mean_squared_error'
	)

	print('Regression: ')
	print('\tmorfist (rmse): {}'.format(reg_scores.mean()))
	print('\tscikit-learn (rmse): {}'.format(np.sqrt(-scores.mean())))

def test_mix_1():
	mix_rf = MixedRandomForest(
		n_estimators=n_trees,
		min_samples_leaf=5,
		class_targets=[1]
	)

	mix_scores = cross_validation(
		mix_rf,
		x_mix_1,
		y_mix_1,
		folds=10,
		class_targets=[1]
	)
	print('Mixed output: ')
	print('\ttask 1 (original) (rmse): {}'.format(mix_scores[0]))
	print('\ttask 2 (additional) (accuracy): {}'.format(mix_scores[1]))

def test_mix_2():
	mix_rf = MixedRandomForest(
		n_estimators=n_trees,
		min_samples_leaf=1,
		class_targets=[0]
	)

	mix_scores = cross_validation(
		mix_rf,
		x_mix_2,
		y_mix_2,
		folds=10,
		class_targets=[0]
	)
	print('Mixed output: ')
	print('\ttask 1 (original) (accuracy): {}'.format(mix_scores[0]))
	print('\ttask 2 (additional) (rmse): {}'.format(mix_scores[1]))

if __name__ == '__main__':
	test_class()
	print('')
	test_reg()
	print('')
	test_mix_1()
	print('')
	test_mix_2()


text:
'''
MIT License

Copyright (c) 2018 Frank Lee

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
'''

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

import numpy as np
import random
from tqdm import tqdm
import matplotlib.pyplot as plt

def get_action():
    if np.random.binomial(1, 0.5) == 1:
        return -1
    return 1

w = np.zeros(10)
def update(state, delta):
    global w
    w[(state-1)//100] += delta
    return

def image(x, y, l, ylabel, name):
    plt.subplot(1, 1, 1)
    for i in range(len(y)):
        plt.plot(x, y[i], label=l[i])
    plt.xlabel('State')
    plt.ylabel(ylabel)
    plt.legend()
    plt.savefig(name)
    plt.close()

def get_value(state):
    global w
    i = (state - 1 ) // 100
    return w[i]

def step(state, action):
    r = random.randint(1,200)
    if r < 101:
        new_state = state - r
    else:
        new_state = state + r - 100
    if new_state < 1:
        return 0, -1
    elif new_state > 1000:
        return 1001, 1
    return new_state, 0

def is_end(state):
    if state == 0 or state == 1001:
        return True
    return False

def run(episodes):
    alpha = 2e-5
    gamma = 1.0
    states = np.zeros((1000, 1))
    for _ in tqdm(range(episodes)):
        reward = 0.0
        state = 500
        trajectory = [state]
        rewards = []
        while not is_end(state):
            states[state-1, 0] += 1
            action = get_action()
            new_state, reward = step(state, action)
            rewards.append(reward)
            trajectory.append(new_state)
            state = new_state
        g = 0.0
        gs = []
        for j in range(len(trajectory)-1):
            i = len(trajectory) - 1 - j - 1
            r = rewards[i]
            g = gamma * g + r
            gs.append(g)
        for i in range(len(trajectory) - 1):
            s = trajectory[i]
            d = 1
            delta = alpha * (gs[i] - get_value(s)) * d
            update(s, delta)
    return states

def main():
    true_value = (np.arange(-998, 1003, 2) / 1001.0)[0:-1]
    true_value = np.reshape(true_value, (1000, 1))
    episodes = 100000
    states = run(episodes)
    data = np.zeros((1000, 1))
    for i in range(1000):
        j = i // 100
        data[i, 0] = w[j]
    states /= np.sum(states)
    image(np.reshape(np.arange(1000) + 1, (1000, 1)), [data, true_value], ['Approximate MC value', 'True Value'], 'Value Scale', 'value.png')
    image(np.reshape(np.arange(1000) + 1, (1000, 1)), [states], ['State Distribution'], 'State Distribution', 'distribution.png')

if __name__ == "__main__":
    main()

max_stars_repo_name: FrankLeeC/ReinforcementLearning

max_issues_repo_name: FrankLeeC/ReinforcementLearning

max_forks_repo_name: FrankLeeC/ReinforcementLearning


text:
"""
Measure speed to report in the paper, because 0.03 and 0.06 m/s is really unrealiable.
This will involve me holding a stopwatch. Yeah.
"""

from autolab.data_collector import DataCollector
from dvrk.robot import *
import numpy as np
import pickle
import sys
import time
import tfx
import utilities

###########################
# ADJUST THESE AS NEEDED! #
###########################

VERSION_INPUT = '00'
RF_REGRESSOR  = pickle.load(open('config/mapping_results/random_forest_predictor_v'+VERSION_INPUT+'.p', 'r'))
PARAMETERS    = pickle.load(open('config/mapping_results/params_matrices_v'+VERSION_INPUT+'.p', 'r'))
ROTATION      = utilities.get_average_rotation(VERSION_INPUT)
TFX_ROTATION  = tfx.tb_angles(ROTATION[0], ROTATION[1], ROTATION[2])
HOME          = [0.00, 0.06, -0.13]
NUM_RUNS      = 10

##########################
# END OF `CONFIGURATION` #
##########################

def move(arm, pos, SPEED_CLASS):
    """ Handles the different speeds we're using. """
    if SPEED_CLASS == 'Slow':
        arm.move_cartesian_frame_linear_interpolation(tfx.pose(pos, TFX_ROTATION), 0.03)
    elif SPEED_CLASS == 'Medium':
        arm.move_cartesian_frame_linear_interpolation(tfx.pose(pos, TFX_ROTATION), 0.06)
    elif SPEED_CLASS == 'Fast':
        arm.move_cartesian_frame(tfx.pose(pos, TFX_ROTATION))
    else:
        raise ValueError()

def benchmark_time(arm, START, TARGET, SPEED_CLASS):
    """ This method returns an estimate of the speed. """
    times = []
    for i in range(NUM_RUNS):
        move(arm, START, 'Slow') # Just bring it slowly to the starting position.
        time.sleep(2)
        start_time = time.time()
        move(arm, TARGET, SPEED_CLASS) # This is what we measure.
        difference = time.time() - start_time
        times.append(difference)
        time.sleep(2)
    times = np.array(times)
    print("\ntimes (in seconds!!) for speed class {}:\n{}".format(SPEED_CLASS, times))
    print("mean(times): {:.2f}".format(np.mean(times)))
    print("std(times):  {:.2f}".format(np.std(times)))

if __name__ == "__main__":
    """ See the top of the file for program-wide arguments. """
    arm, _, d = utilities.initializeRobots(sleep_time=2)
    arm.close_gripper()
    START = [HOME[0]-0.03, HOME[1], HOME[2]]
    TARGET = [START[0]+0.1, START[1], START[2]]
    benchmark_time(arm, START, TARGET, SPEED_CLASS='Slow')
    benchmark_time(arm, START, TARGET, SPEED_CLASS='Medium')
    benchmark_time(arm, START, TARGET, SPEED_CLASS='Fast')

max_stars_repo_name: DanielTakeshi/debridement-code

max_issues_repo_name: DanielTakeshi/debridement-code

max_forks_repo_name: DanielTakeshi/debridement-code


text:
# -*- coding: utf-8 -*-
# Licensed under a 3-clause BSD style license - see LICENSE.rst

import re
from io import BytesIO, open
from collections import OrderedDict
import locale
import platform
from io import StringIO

import pathlib
import pytest
import numpy as np

from ... import ascii
from ....table import Table
from .... import table
from ....units import Unit
from ....table.table_helpers import simple_table

from .common import (raises, assert_equal, assert_almost_equal,
                     assert_true)
from .. import core
from ..ui import _probably_html, get_read_trace, cparser

# setup/teardown function to have the tests run in the correct directory
from .common import setup_function, teardown_function

try:
    import bz2  # pylint: disable=W0611
except ImportError:
    HAS_BZ2 = False
else:
    HAS_BZ2 = True

asciiIO = lambda x: BytesIO(x.encode('ascii'))

@pytest.mark.parametrize('fast_reader', [True, False, {'use_fast_converter': False},
                                         {'use_fast_converter': True}, 'force'])
def test_convert_overflow(fast_reader):
    """
    Test reading an extremely large integer, which falls through to
    string due to an overflow error (#2234). The C parsers used to
    return inf (kind 'f') for this.
    """
    expected_kind = 'U'
    dat = ascii.read(['a', '1' * 10000], format='basic',
                     fast_reader=fast_reader, guess=False)
    assert dat['a'].dtype.kind == expected_kind

def test_guess_with_names_arg():
    """
    Make sure reading a table with guess=True gives the expected result when
    the names arg is specified.
    """
    # This is a NoHeader format table and so `names` should replace
    # the default col0, col1 names.  It fails as a Basic format
    # table when guessing because the column names would be '1', '2'.
    dat = ascii.read(['1,2', '3,4'], names=('a', 'b'))
    assert len(dat) == 2
    assert dat.colnames == ['a', 'b']

    # This is a Basic format table and the first row
    # gives the column names 'c', 'd', which get replaced by 'a', 'b'
    dat = ascii.read(['c,d', '3,4'], names=('a', 'b'))
    assert len(dat) == 1
    assert dat.colnames == ['a', 'b']

    # This is also a Basic format table and the first row
    # gives the column names 'c', 'd', which get replaced by 'a', 'b'
    dat = ascii.read(['c d', 'e f'], names=('a', 'b'))
    assert len(dat) == 1
    assert dat.colnames == ['a', 'b']

def test_guess_with_format_arg():
    """
    When the format or Reader is explicitly given then disable the
    strict column name checking in guessing.
    """
    dat = ascii.read(['1,2', '3,4'], format='basic')
    assert len(dat) == 1
    assert dat.colnames == ['1', '2']

    dat = ascii.read(['1,2', '3,4'], names=('a', 'b'), format='basic')
    assert len(dat) == 1
    assert dat.colnames == ['a', 'b']

    dat = ascii.read(['1,2', '3,4'], Reader=ascii.Basic)
    assert len(dat) == 1
    assert dat.colnames == ['1', '2']

    dat = ascii.read(['1,2', '3,4'], names=('a', 'b'), Reader=ascii.Basic)
    assert len(dat) == 1
    assert dat.colnames == ['a', 'b']

    # For good measure check the same in the unified I/O interface
    dat = Table.read(['1,2', '3,4'], format='ascii.basic')
    assert len(dat) == 1
    assert dat.colnames == ['1', '2']

    dat = Table.read(['1,2', '3,4'], format='ascii.basic', names=('a', 'b'))
    assert len(dat) == 1
    assert dat.colnames == ['a', 'b']

def test_guess_with_delimiter_arg():
    """
    When the delimiter is explicitly given then do not try others in guessing.
    """
    fields = ['10.1E+19', '3.14', '2048', '-23']
    values = [1.01e20, 3.14, 2048, -23]

    # Default guess should recognise CSV with optional spaces
    t0 = ascii.read(asciiIO(', '.join(fields)), guess=True)
    for n, v in zip(t0.colnames, values):
        assert t0[n][0] == v

    # Forcing space as delimiter produces type str columns ('10.1E+19,')
    t1 = ascii.read(asciiIO(', '.join(fields)), guess=True, delimiter=' ')
    for n, v in zip(t1.colnames[:-1], fields[:-1]):
        assert t1[n][0] == v+','

def test_reading_mixed_delimiter_tabs_spaces():
    # Regression test for https://github.com/astropy/astropy/issues/6770
    dat = ascii.read('1 2\t3\n1 2\t3', format='no_header', names=list('abc'))
    assert len(dat) == 2

    Table.read(['1 2\t3', '1 2\t3'], format='ascii.no_header',
               names=['a', 'b', 'c'])
    assert len(dat) == 2

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_read_with_names_arg(fast_reader):
    """
    Test that a bad value of `names` raises an exception.
    """
    # CParser only uses columns in `names` and thus reports mismach in num_col
    with pytest.raises(ascii.InconsistentTableError):
        ascii.read(['c d', 'e f'], names=('a', ), guess=False, fast_reader=fast_reader)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_read_all_files(fast_reader):
    for testfile in get_testfiles():
        if testfile.get('skip'):
            print('\n\n******** SKIPPING {}'.format(testfile['name']))
            continue
        print('\n\n******** READING {}'.format(testfile['name']))
        for guess in (True, False):
            test_opts = testfile['opts'].copy()
            if 'guess' not in test_opts:
                test_opts['guess'] = guess
            if ('Reader' in test_opts and 'fast_{0}'.format(test_opts['Reader']._format_name)
                    in core.FAST_CLASSES):  # has fast version
                if 'Inputter' not in test_opts:  # fast reader doesn't allow this
                    test_opts['fast_reader'] = fast_reader
            table = ascii.read(testfile['name'], **test_opts)
            assert_equal(table.dtype.names, testfile['cols'])
            for colname in table.dtype.names:
                assert_equal(len(table[colname]), testfile['nrows'])

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_read_all_files_via_table(fast_reader):
    for testfile in get_testfiles():
        if testfile.get('skip'):
            print('\n\n******** SKIPPING {}'.format(testfile['name']))
            continue
        print('\n\n******** READING {}'.format(testfile['name']))
        for guess in (True, False):
            test_opts = testfile['opts'].copy()
            if 'guess' not in test_opts:
                test_opts['guess'] = guess
            if 'Reader' in test_opts:
                format = 'ascii.{0}'.format(test_opts['Reader']._format_name)
                del test_opts['Reader']
            else:
                format = 'ascii'
            if 'fast_{0}'.format(format) in core.FAST_CLASSES:
                test_opts['fast_reader'] = fast_reader
            table = Table.read(testfile['name'], format=format, **test_opts)
            assert_equal(table.dtype.names, testfile['cols'])
            for colname in table.dtype.names:
                assert_equal(len(table[colname]), testfile['nrows'])

def test_guess_all_files():
    for testfile in get_testfiles():
        if testfile.get('skip'):
            print('\n\n******** SKIPPING {}'.format(testfile['name']))
            continue
        if not testfile['opts'].get('guess', True):
            continue
        print('\n\n******** READING {}'.format(testfile['name']))
        for filter_read_opts in (['Reader', 'delimiter', 'quotechar'], []):
            # Copy read options except for those in filter_read_opts
            guess_opts = dict((k, v) for k, v in testfile['opts'].items()
                              if k not in filter_read_opts)
            table = ascii.read(testfile['name'], guess=True, **guess_opts)
            assert_equal(table.dtype.names, testfile['cols'])
            for colname in table.dtype.names:
                assert_equal(len(table[colname]), testfile['nrows'])

def test_daophot_indef():
    """Test that INDEF is correctly interpreted as a missing value"""
    table = ascii.read('t/daophot2.dat', Reader=ascii.Daophot)
    for colname in table.colnames:
        # Three columns have all INDEF values and are masked
        mask_value = colname in ('OTIME', 'MAG', 'MERR', 'XAIRMASS')
        assert np.all(table[colname].mask == mask_value)

def test_daophot_types():
    """
    Test specific data types which are different from what would be
    inferred automatically based only data values.  DAOphot reader uses
    the header information to assign types.
    """
    table = ascii.read('t/daophot2.dat', Reader=ascii.Daophot)
    assert table['LID'].dtype.char in 'fd'  # float or double
    assert table['MAG'].dtype.char in 'fd'  # even without any data values
    assert table['PIER'].dtype.char in 'US'  # string (data values are consistent with int)
    assert table['ID'].dtype.char in 'il'  # int or long

def test_daophot_header_keywords():
    table = ascii.read('t/daophot.dat', Reader=ascii.Daophot)
    expected_keywords = (('NSTARFILE', 'test.nst.1', 'filename', '%-23s'),
                         ('REJFILE', '"hello world"', 'filename', '%-23s'),
                         ('SCALE', '1.', 'units/pix', '%-23.7g'),)

    keywords = table.meta['keywords']  # Ordered dict of keyword structures
    for name, value, units, format_ in expected_keywords:
        keyword = keywords[name]
        assert_equal(keyword['value'], value)
        assert_equal(keyword['units'], units)
        assert_equal(keyword['format'], format_)

def test_daophot_multiple_aperture():
    table = ascii.read('t/daophot3.dat', Reader=ascii.Daophot)
    assert 'MAG5' in table.colnames  # MAG5 is one of the newly created column names
    assert table['MAG5'][4] == 22.13  # A sample entry in daophot3.dat file
    assert table['MERR2'][0] == 1.171
    assert np.all(table['RAPERT5'] == 23.3)  # assert all the 5th apertures are same 23.3

def test_daophot_multiple_aperture2():
    table = ascii.read('t/daophot4.dat', Reader=ascii.Daophot)
    assert 'MAG15' in table.colnames  # MAG15 is one of the newly created column name
    assert table['MAG15'][1] == -7.573  # A sample entry in daophot4.dat file
    assert table['MERR2'][0] == 0.049
    assert np.all(table['RAPERT5'] == 5.)  # assert all the 5th apertures are same 5.0

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_empty_table_no_header(fast_reader):
    with pytest.raises(ascii.InconsistentTableError):
        ascii.read('t/no_data_without_header.dat', Reader=ascii.NoHeader,
                   guess=False, fast_reader=fast_reader)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_wrong_quote(fast_reader):
    with pytest.raises(ascii.InconsistentTableError):
        ascii.read('t/simple.txt', guess=False, fast_reader=fast_reader)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_extra_data_col(fast_reader):
    with pytest.raises(ascii.InconsistentTableError):
        ascii.read('t/bad.txt', fast_reader=fast_reader)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_extra_data_col2(fast_reader):
    with pytest.raises(ascii.InconsistentTableError):
        ascii.read('t/simple5.txt', delimiter='|', fast_reader=fast_reader)

@raises(OSError)
def test_missing_file():
    ascii.read('does_not_exist')

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_set_names(fast_reader):
    names = ('c1', 'c2', 'c3', 'c4', 'c5', 'c6')
    data = ascii.read('t/simple3.txt', names=names, delimiter='|',
                      fast_reader=fast_reader)
    assert_equal(data.dtype.names, names)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_set_include_names(fast_reader):
    names = ('c1', 'c2', 'c3', 'c4', 'c5', 'c6')
    include_names = ('c1', 'c3')
    data = ascii.read('t/simple3.txt', names=names, include_names=include_names,
                      delimiter='|', fast_reader=fast_reader)
    assert_equal(data.dtype.names, include_names)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_set_exclude_names(fast_reader):
    exclude_names = ('Y', 'object')
    data = ascii.read('t/simple3.txt', exclude_names=exclude_names, delimiter='|',
                      fast_reader=fast_reader)
    assert_equal(data.dtype.names, ('obsid', 'redshift', 'X', 'rad'))

def test_include_names_daophot():
    include_names = ('ID', 'MAG', 'PIER')
    data = ascii.read('t/daophot.dat', include_names=include_names)
    assert_equal(data.dtype.names, include_names)

def test_exclude_names_daophot():
    exclude_names = ('ID', 'YCENTER', 'MERR', 'NITER', 'CHI', 'PERROR')
    data = ascii.read('t/daophot.dat', exclude_names=exclude_names)
    assert_equal(data.dtype.names, ('XCENTER', 'MAG', 'MSKY', 'SHARPNESS', 'PIER'))

def test_custom_process_lines():
    def process_lines(lines):
        bars_at_ends = re.compile(r'^\| | \|$', re.VERBOSE)
        striplines = (x.strip() for x in lines)
        return [bars_at_ends.sub('', x) for x in striplines if len(x) > 0]
    reader = ascii.get_reader(delimiter='|')
    reader.inputter.process_lines = process_lines
    data = reader.read('t/bars_at_ends.txt')
    assert_equal(data.dtype.names, ('obsid', 'redshift', 'X', 'Y', 'object', 'rad'))
    assert_equal(len(data), 3)

def test_custom_process_line():
    def process_line(line):
        line_out = re.sub(r'^\|\s*', '', line.strip())
        return line_out
    reader = ascii.get_reader(data_start=2, delimiter='|')
    reader.header.splitter.process_line = process_line
    reader.data.splitter.process_line = process_line
    data = reader.read('t/nls1_stackinfo.dbout')
    cols = get_testfiles('t/nls1_stackinfo.dbout')['cols']
    assert_equal(data.dtype.names, cols[1:])

def test_custom_splitters():
    reader = ascii.get_reader()
    reader.header.splitter = ascii.BaseSplitter()
    reader.data.splitter = ascii.BaseSplitter()
    f = 't/test4.dat'
    data = reader.read(f)
    testfile = get_testfiles(f)
    assert_equal(data.dtype.names, testfile['cols'])
    assert_equal(len(data), testfile['nrows'])
    assert_almost_equal(data.field('zabs1.nh')[2], 0.0839710433091)
    assert_almost_equal(data.field('p1.gamma')[2], 1.25997502704)
    assert_almost_equal(data.field('p1.ampl')[2], 0.000696444029148)
    assert_equal(data.field('statname')[2], 'chi2modvar')
    assert_almost_equal(data.field('statval')[2], 497.56468441)

def test_start_end():
    data = ascii.read('t/test5.dat', header_start=1, data_start=3, data_end=-5)
    assert_equal(len(data), 13)
    assert_equal(data.field('statname')[0], 'chi2xspecvar')
    assert_equal(data.field('statname')[-1], 'chi2gehrels')

def test_set_converters():
    converters = {'zabs1.nh': [ascii.convert_numpy('int32'),
                               ascii.convert_numpy('float32')],
                  'p1.gamma': [ascii.convert_numpy('str')]
                  }
    data = ascii.read('t/test4.dat', converters=converters)
    assert_equal(str(data['zabs1.nh'].dtype), 'float32')
    assert_equal(data['p1.gamma'][0], '1.26764500000')

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_from_string(fast_reader):
    f = 't/simple.txt'
    with open(f) as fd:
        table = fd.read()
    testfile = get_testfiles(f)
    data = ascii.read(table, fast_reader=fast_reader, **testfile['opts'])
    assert_equal(data.dtype.names, testfile['cols'])
    assert_equal(len(data), testfile['nrows'])

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_from_filelike(fast_reader):
    f = 't/simple.txt'
    testfile = get_testfiles(f)
    with open(f, 'rb') as fd:
        data = ascii.read(fd, fast_reader=fast_reader, **testfile['opts'])
    assert_equal(data.dtype.names, testfile['cols'])
    assert_equal(len(data), testfile['nrows'])

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_from_lines(fast_reader):
    f = 't/simple.txt'
    with open(f) as fd:
        table = fd.readlines()
    testfile = get_testfiles(f)
    data = ascii.read(table, fast_reader=fast_reader, **testfile['opts'])
    assert_equal(data.dtype.names, testfile['cols'])
    assert_equal(len(data), testfile['nrows'])

def test_comment_lines():
    table = ascii.get_reader(Reader=ascii.Rdb)
    data = table.read('t/apostrophe.rdb')
    assert_equal(table.comment_lines, ['# first comment', '  # second comment'])
    assert_equal(data.meta['comments'], ['first comment', 'second comment'])

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_fill_values(fast_reader):
    f = 't/fill_values.txt'
    testfile = get_testfiles(f)
    data = ascii.read(f, fill_values=('a', '1'), fast_reader=fast_reader,
                      **testfile['opts'])
    assert_true((data['a'].mask == [False, True]).all())
    assert_true((data['a'] == [1, 1]).all())
    assert_true((data['b'].mask == [False, True]).all())
    assert_true((data['b'] == [2, 1]).all())

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_fill_values_col(fast_reader):
    f = 't/fill_values.txt'
    testfile = get_testfiles(f)
    data = ascii.read(f, fill_values=('a', '1', 'b'), fast_reader=fast_reader,
                      **testfile['opts'])
    check_fill_values(data)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_fill_values_include_names(fast_reader):
    f = 't/fill_values.txt'
    testfile = get_testfiles(f)
    data = ascii.read(f, fill_values=('a', '1'), fast_reader=fast_reader,
                      fill_include_names=['b'], **testfile['opts'])
    check_fill_values(data)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_fill_values_exclude_names(fast_reader):
    f = 't/fill_values.txt'
    testfile = get_testfiles(f)
    data = ascii.read(f, fill_values=('a', '1'), fast_reader=fast_reader,
                      fill_exclude_names=['a'], **testfile['opts'])
    check_fill_values(data)

def check_fill_values(data):
    """compare array column by column with expectation """
    assert_true((data['a'].mask == [False, False]).all())
    assert_true((data['a'] == ['1', 'a']).all())
    assert_true((data['b'].mask == [False, True]).all())
    # Check that masked value is "do not care" in comparison
    assert_true((data['b'] == [2, -999]).all())
    data['b'].mask = False  # explicitly unmask for comparison
    assert_true((data['b'] == [2, 1]).all())

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_fill_values_list(fast_reader):
    f = 't/fill_values.txt'
    testfile = get_testfiles(f)
    data = ascii.read(f, fill_values=[('a', '42'), ('1', '42', 'a')],
                      fast_reader=fast_reader, **testfile['opts'])
    data['a'].mask = False  # explicitly unmask for comparison
    assert_true((data['a'] == [42, 42]).all())

def test_masking_Cds():
    f = 't/cds.dat'
    testfile = get_testfiles(f)
    data = ascii.read(f,
                           **testfile['opts'])
    assert_true(data['AK'].mask[0])
    assert_true(not data['Fit'].mask[0])

def test_null_Ipac():
    f = 't/ipac.dat'
    testfile = get_testfiles(f)
    data = ascii.read(f, **testfile['opts'])
    mask = np.array([(True, False, True, False, True),
                     (False, False, False, False, False)],
                    dtype=[(str('ra'), '|b1'),
                           (str('dec'), '|b1'),
                           (str('sai'), '|b1'),
                           (str('v2'), '|b1'),
                           (str('sptype'), '|b1')])
    assert np.all(data.mask == mask)

def test_Ipac_meta():
    keywords = OrderedDict((('intval', 1),
                            ('floatval', 2.3e3),
                            ('date', "Wed Sp 20 09:48:36 1995"),
                            ('key_continue', 'IPAC keywords can continue across lines')))
    comments = ['This is an example of a valid comment']
    f = 't/ipac.dat'
    testfile = get_testfiles(f)
    data = ascii.read(f, **testfile['opts'])
    assert data.meta['keywords'].keys() == keywords.keys()
    for data_kv, kv in zip(data.meta['keywords'].values(), keywords.values()):
        assert data_kv['value'] == kv
    assert data.meta['comments'] == comments

def test_set_guess_kwarg():
    """Read a file using guess with one of the typical guess_kwargs explicitly set."""
    data = ascii.read('t/space_delim_no_header.dat',
                      delimiter=',', guess=True)
    assert(data.dtype.names == ('1 3.4 hello',))
    assert(len(data) == 1)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_read_rdb_wrong_type(fast_reader):
    """Read RDB data with inconstent data type (except failure)"""
    table = """col1\tcol2
N\tN
1\tHello"""
    with pytest.raises(ValueError):
        ascii.read(table, Reader=ascii.Rdb, fast_reader=fast_reader)

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_default_missing(fast_reader):
    """Read a table with empty values and ensure that corresponding entries are masked"""
    table = '\n'.join(['a,b,c,d',
                       '1,3,,',
                       '2, , 4.0 , ss '])
    dat = ascii.read(table, fast_reader=fast_reader)
    assert dat.masked is True
    assert dat.pformat() == [' a   b   c   d ',
                             '--- --- --- ---',
                             '  1   3  --  --',
                             '  2  -- 4.0  ss']

    # Single row table with a single missing element
    table = """ a \n "" """
    dat = ascii.read(table, fast_reader=fast_reader)
    assert dat.pformat() == [' a ',
                             '---',
                             ' --']
    assert dat['a'].dtype.kind == 'i'

    # Same test with a fixed width reader
    table = '\n'.join([' a   b   c   d ',
                       '--- --- --- ---',
                       '  1   3        ',
                       '  2     4.0  ss'])
    dat = ascii.read(table, Reader=ascii.FixedWidthTwoLine)
    assert dat.masked is True
    assert dat.pformat() == [' a   b   c   d ',
                             '--- --- --- ---',
                             '  1   3  --  --',
                             '  2  -- 4.0  ss']

    dat = ascii.read(table, Reader=ascii.FixedWidthTwoLine, fill_values=None)
    assert dat.masked is False
    assert dat.pformat() == [' a   b   c   d ',
                             '--- --- --- ---',
                             '  1   3        ',
                             '  2     4.0  ss']

    dat = ascii.read(table, Reader=ascii.FixedWidthTwoLine, fill_values=[])
    assert dat.masked is False
    assert dat.pformat() == [' a   b   c   d ',
                             '--- --- --- ---',
                             '  1   3        ',
                             '  2     4.0  ss']

def get_testfiles(name=None):
    """Set up information about the columns, number of rows, and reader params to
    read a bunch of test files and verify columns and number of rows."""

    testfiles = [
        {'cols': ('agasc_id', 'n_noids', 'n_obs'),
         'name': 't/apostrophe.rdb',
         'nrows': 2,
         'opts': {'Reader': ascii.Rdb}},
        {'cols': ('agasc_id', 'n_noids', 'n_obs'),
         'name': 't/apostrophe.tab',
         'nrows': 2,
         'opts': {'Reader': ascii.Tab}},
        {'cols': ('Index',
                  'RAh',
                  'RAm',
                  'RAs',
                  'DE-',
                  'DEd',
                  'DEm',
                  'DEs',
                  'Match',
                  'Class',
                  'AK',
                  'Fit'),
         'name': 't/cds.dat',
         'nrows': 1,
         'opts': {'Reader': ascii.Cds}},
        # Test malformed CDS file (issues #2241 #467)
        {'cols': ('Index',
                  'RAh',
                  'RAm',
                  'RAs',
                  'DE-',
                  'DEd',
                  'DEm',
                  'DEs',
                  'Match',
                  'Class',
                  'AK',
                  'Fit'),
         'name': 't/cds_malformed.dat',
         'nrows': 1,
         'opts': {'Reader': ascii.Cds, 'data_start': 'guess'}},
        {'cols': ('a', 'b', 'c'),
         'name': 't/commented_header.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.CommentedHeader}},
        {'cols': ('a', 'b', 'c'),
         'name': 't/commented_header2.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.CommentedHeader, 'header_start': -1}},
        {'cols': ('col1', 'col2', 'col3', 'col4', 'col5'),
         'name': 't/continuation.dat',
         'nrows': 2,
         'opts': {'Inputter': ascii.ContinuationLinesInputter,
                  'Reader': ascii.NoHeader}},
        {'cols': ('ID',
                  'XCENTER',
                  'YCENTER',
                  'MAG',
                  'MERR',
                  'MSKY',
                  'NITER',
                  'SHARPNESS',
                  'CHI',
                  'PIER',
                  'PERROR'),
         'name': 't/daophot.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.Daophot}},
        {'cols': ('NUMBER',
                  'FLUX_ISO',
                  'FLUXERR_ISO',
                  'VALU-ES',
                  'VALU-ES_1',
                  'FLAG'),
         'name': 't/sextractor.dat',
         'nrows': 3,
         'opts': {'Reader': ascii.SExtractor}},
        {'cols': ('ra', 'dec', 'sai', 'v2', 'sptype'),
         'name': 't/ipac.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.Ipac}},
        {'cols': ('col0',
                  'objID',
                  'osrcid',
                  'xsrcid',
                  'SpecObjID',
                  'ra',
                  'dec',
                  'obsid',
                  'ccdid',
                  'z',
                  'modelMag_i',
                  'modelMagErr_i',
                  'modelMag_r',
                  'modelMagErr_r',
                  'expo',
                  'theta',
                  'rad_ecf_39',
                  'detlim90',
                  'fBlim90'),
         'name': 't/nls1_stackinfo.dbout',
         'nrows': 58,
         'opts': {'data_start': 2, 'delimiter': '|', 'guess': False}},
        {'cols': ('Index',
                  'RAh',
                  'RAm',
                  'RAs',
                  'DE-',
                  'DEd',
                  'DEm',
                  'DEs',
                  'Match',
                  'Class',
                  'AK',
                  'Fit'),
         'name': 't/no_data_cds.dat',
         'nrows': 0,
         'opts': {'Reader': ascii.Cds}},
        {'cols': ('ID',
                  'XCENTER',
                  'YCENTER',
                  'MAG',
                  'MERR',
                  'MSKY',
                  'NITER',
                  'SHARPNESS',
                  'CHI',
                  'PIER',
                  'PERROR'),
         'name': 't/no_data_daophot.dat',
         'nrows': 0,
         'opts': {'Reader': ascii.Daophot}},
        {'cols': ('NUMBER',
                  'FLUX_ISO',
                  'FLUXERR_ISO',
                  'VALUES',
                  'VALUES_1',
                  'FLAG'),
         'name': 't/no_data_sextractor.dat',
         'nrows': 0,
         'opts': {'Reader': ascii.SExtractor}},
        {'cols': ('ra', 'dec', 'sai', 'v2', 'sptype'),
         'name': 't/no_data_ipac.dat',
         'nrows': 0,
         'opts': {'Reader': ascii.Ipac}},
        {'cols': ('ra', 'v2'),
         'name': 't/ipac.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.Ipac, 'include_names': ['ra', 'v2']}},
        {'cols': ('a', 'b', 'c'),
         'name': 't/no_data_with_header.dat',
         'nrows': 0,
         'opts': {}},
        {'cols': ('agasc_id', 'n_noids', 'n_obs'),
         'name': 't/short.rdb',
         'nrows': 7,
         'opts': {'Reader': ascii.Rdb}},
        {'cols': ('agasc_id', 'n_noids', 'n_obs'),
         'name': 't/short.tab',
         'nrows': 7,
         'opts': {'Reader': ascii.Tab}},
        {'cols': ('test 1a', 'test2', 'test3', 'test4'),
         'name': 't/simple.txt',
         'nrows': 2,
         'opts': {'quotechar': "'"}},
        {'cols': ('top1', 'top2', 'top3', 'top4'),
         'name': 't/simple.txt',
         'nrows': 1,
         'opts': {'quotechar': "'", 'header_start': 1, 'data_start': 2}},
        {'cols': ('top1', 'top2', 'top3', 'top4'),
         'name': 't/simple.txt',
         'nrows': 1,
         'opts': {'quotechar': "'", 'header_start': 1}},
        {'cols': ('top1', 'top2', 'top3', 'top4'),
         'name': 't/simple.txt',
         'nrows': 2,
         'opts': {'quotechar': "'", 'header_start': 1, 'data_start': 1}},
        {'cols': ('obsid', 'redshift', 'X', 'Y', 'object', 'rad'),
         'name': 't/simple2.txt',
         'nrows': 3,
         'opts': {'delimiter': '|'}},
        {'cols': ('obsid', 'redshift', 'X', 'Y', 'object', 'rad'),
         'name': 't/simple3.txt',
         'nrows': 2,
         'opts': {'delimiter': '|'}},
        {'cols': ('col1', 'col2', 'col3', 'col4', 'col5', 'col6'),
         'name': 't/simple4.txt',
         'nrows': 3,
         'opts': {'Reader': ascii.NoHeader, 'delimiter': '|'}},
        {'cols': ('col1', 'col2', 'col3'),
         'name': 't/space_delim_no_header.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.NoHeader}},
        {'cols': ('col1', 'col2', 'col3'),
         'name': 't/space_delim_no_header.dat',
         'nrows': 2,
         'opts': {'Reader': ascii.NoHeader, 'header_start': None}},
        {'cols': ('obsid', 'offset', 'x', 'y', 'name', 'oaa'),
         'name': 't/space_delim_blank_lines.txt',
         'nrows': 3,
         'opts': {}},
        {'cols': ('zabs1.nh', 'p1.gamma', 'p1.ampl', 'statname', 'statval'),
         'name': 't/test4.dat',
         'nrows': 9,
         'opts': {}},
        {'cols': ('a', 'b', 'c'),
         'name': 't/fill_values.txt',
         'nrows': 2,
         'opts': {'delimiter': ','}},
        {'name': 't/whitespace.dat',
         'cols': ('quoted colname with tab\tinside', 'col2', 'col3'),
         'nrows': 2,
         'opts': {'delimiter': r'\s'}},
        {'name': 't/simple_csv.csv',
         'cols': ('a', 'b', 'c'),
         'nrows': 2,
         'opts': {'Reader': ascii.Csv}},
        {'name': 't/simple_csv_missing.csv',
         'cols': ('a', 'b', 'c'),
         'nrows': 2,
         'skip': True,
         'opts': {'Reader': ascii.Csv}},
        {'cols': ('cola', 'colb', 'colc'),
         'name': 't/latex1.tex',
         'nrows': 2,
         'opts': {'Reader': ascii.Latex}},
        {'cols': ('Facility', 'Id', 'exposure', 'date'),
         'name': 't/latex2.tex',
         'nrows': 3,
         'opts': {'Reader': ascii.AASTex}},
        {'cols': ('cola', 'colb', 'colc'),
         'name': 't/latex3.tex',
         'nrows': 2,
         'opts': {'Reader': ascii.Latex}},
        {'cols': ('Col1', 'Col2', 'Col3', 'Col4'),
         'name': 't/fixed_width_2_line.txt',
         'nrows': 2,
         'opts': {'Reader': ascii.FixedWidthTwoLine}},
    ]

    try:
        import bs4  # pylint: disable=W0611
        testfiles.append({'cols': ('Column 1', 'Column 2', 'Column 3'),
                          'name': 't/html.html',
                          'nrows': 3,
                          'opts': {'Reader': ascii.HTML}})
    except ImportError:
        pass

    if name is not None:
        return [x for x in testfiles if x['name'] == name][0]
    else:
        return testfiles

def test_header_start_exception():
    '''Check certain Readers throw an exception if ``header_start`` is set

    For certain Readers it does not make sense to set the ``header_start``, they
    throw an exception if you try.
    This was implemented in response to issue #885.
    '''
    for readerclass in [ascii.NoHeader, ascii.SExtractor, ascii.Ipac,
                        ascii.BaseReader, ascii.FixedWidthNoHeader,
                        ascii.Cds, ascii.Daophot]:
        with pytest.raises(ValueError):
            reader = ascii.core._get_reader(readerclass, header_start=5)

def test_csv_table_read():
    """
    Check for a regression introduced by #1935.  Pseudo-CSV file with
    commented header line.
    """
    lines = ['# a, b',
             '1, 2',
             '3, 4']
    t = ascii.read(lines)
    assert t.colnames == ['a', 'b']

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_overlapping_names(fast_reader):
    """
    Check that the names argument list can overlap with the existing column names.
    This tests the issue in #1991.
    """
    t = ascii.read(['a b', '1 2'], names=['b', 'a'], fast_reader=fast_reader)
    assert t.colnames == ['b', 'a']

def test_sextractor_units():
    """
    Make sure that the SExtractor reader correctly inputs descriptions and units.
    """
    table = ascii.read('t/sextractor2.dat', Reader=ascii.SExtractor, guess=False)
    expected_units = [None, Unit('pix'), Unit('pix'), Unit('mag'),
                      Unit('mag'), None, Unit('pix**2'), Unit('m**(-6)'),
                      Unit('mag * arcsec**(-2)')]
    expected_descrs = ['Running object number',
                       'Windowed position estimate along x',
                       'Windowed position estimate along y',
                       'Kron-like elliptical aperture magnitude',
                       'RMS error for AUTO magnitude',
                       'Extraction flags',
                       None,
                       'Barycenter position along MAMA x axis',
                       'Peak surface brightness above background']
    for i, colname in enumerate(table.colnames):
        assert table[colname].unit == expected_units[i]
        assert table[colname].description == expected_descrs[i]

def test_sextractor_last_column_array():
    """
    Make sure that the SExtractor reader handles the last column correctly when it is array-like.
    """
    table = ascii.read('t/sextractor3.dat', Reader=ascii.SExtractor, guess=False)
    expected_columns = ['X_IMAGE', 'Y_IMAGE', 'ALPHA_J2000', 'DELTA_J2000',
                        'MAG_AUTO', 'MAGERR_AUTO',
                        'MAG_APER', 'MAG_APER_1', 'MAG_APER_2', 'MAG_APER_3', 'MAG_APER_4', 'MAG_APER_5', 'MAG_APER_6',
                        'MAGERR_APER', 'MAGERR_APER_1', 'MAGERR_APER_2', 'MAGERR_APER_3', 'MAGERR_APER_4', 'MAGERR_APER_5', 'MAGERR_APER_6']
    expected_units = [Unit('pix'), Unit('pix'), Unit('deg'), Unit('deg'),
                      Unit('mag'), Unit('mag'),
                      Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'),
                      Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag'), Unit('mag')]
    expected_descrs = ['Object position along x', None,
                       'Right ascension of barycenter (J2000)',
                       'Declination of barycenter (J2000)',
                       'Kron-like elliptical aperture magnitude',
                       'RMS error for AUTO magnitude', ] + [
                       'Fixed aperture magnitude vector'] * 7 + [
                       'RMS error vector for fixed aperture mag.'] * 7
    for i, colname in enumerate(table.colnames):
        assert table[colname].name == expected_columns[i]
        assert table[colname].unit == expected_units[i]
        assert table[colname].description == expected_descrs[i]

def test_list_with_newlines():
    """
    Check that lists of strings where some strings consist of just a newline
    ("\n") are parsed correctly.
    """
    t = ascii.read(["abc", "123\n", "456\n", "\n", "\n"])
    assert t.colnames == ['abc']
    assert len(t) == 2
    assert t[0][0] == 123
    assert t[1][0] == 456

def test_commented_csv():
    """
    Check that Csv reader does not have ignore lines with the # comment
    character which is defined for most Basic readers.
    """
    t = ascii.read(['#a,b', '1,2', '#3,4'], format='csv')
    assert t.colnames == ['#a', 'b']
    assert len(t) == 2
    assert t['#a'][1] == '#3'

def test_meta_comments():
    """
    Make sure that line comments are included in the ``meta`` attribute
    of the output Table.
    """
    t = ascii.read(['#comment1', '#   comment2 \t', 'a,b,c', '1,2,3'])
    assert t.colnames == ['a', 'b', 'c']
    assert t.meta['comments'] == ['comment1', 'comment2']

def test_guess_fail():
    """
    Check the error message when guess fails
    """
    with pytest.raises(ascii.InconsistentTableError) as err:
        ascii.read('asfdasdf\n1 2 3', format='basic')
    assert "** To figure out why the table did not read, use guess=False and" in str(err.value)

    # Test the case with guessing enabled but for a format that has no free params
    with pytest.raises(ValueError) as err:
        ascii.read('asfdasdf\n1 2 3', format='ipac')
    assert 'At least one header line beginning and ending with delimiter required' in str(err.value)

    # Test the case with guessing enabled but with all params specified
    with pytest.raises(ValueError) as err:
        ascii.read('asfdasdf\n1 2 3', format='basic', quotechar='"', delimiter=' ', fast_reader=False)
    assert 'Number of header columns (1) inconsistent with data columns (3)' in str(err.value)

@pytest.mark.xfail('not HAS_BZ2')
def test_guessing_file_object():
    """
    Test guessing a file object.  Fixes #3013 and similar issue noted in #3019.
    """
    t = ascii.read(open('t/ipac.dat.bz2', 'rb'))
    assert t.colnames == ['ra', 'dec', 'sai', 'v2', 'sptype']

def test_pformat_roundtrip():
    """Check that the screen output of ``print tab`` can be read. See #3025."""
    """Read a table with empty values and ensure that corresponding entries are masked"""
    table = '\n'.join(['a,b,c,d',
                       '1,3,1.11,1',
                       '2, 2, 4.0 , ss '])
    dat = ascii.read(table)
    out = ascii.read(dat.pformat())
    assert len(dat) == len(out)
    assert dat.colnames == out.colnames
    for c in dat.colnames:
        assert np.all(dat[c] == out[c])

def test_ipac_abbrev():
    lines = ['| c1 | c2 | c3   |   c4 | c5| c6 | c7  | c8 | c9|c10|c11|c12|',
             '| r  | rE | rea  | real | D | do | dou | f  | i | l | da| c |',
             '  1    2    3       4     5   6    7     8    9   10  11  12 ']
    dat = ascii.read(lines, format='ipac')
    for name in dat.columns[0:8]:
        assert dat[name].dtype.kind == 'f'
    for name in dat.columns[8:10]:
        assert dat[name].dtype.kind == 'i'
    for name in dat.columns[10:12]:
        assert dat[name].dtype.kind in ('U', 'S')

def test_almost_but_not_quite_daophot():
    '''Regression test for #3319.
    This tables looks so close to a daophot table, that the daophot reader gets
    quite far before it fails with an AttributeError.

    Note that this table will actually be read as Commented Header table with
    the columns ['some', 'header', 'info'].
    '''
    lines = ["# some header info",
             "#F header info beginning with 'F'",
             "1 2 3",
             "4 5 6",
             "7 8 9"]
    dat = ascii.read(lines)
    assert len(dat) == 3

@pytest.mark.parametrize('fast', [False, 'force'])
def test_commented_header_comments(fast):
    """
    Test that comments in commented_header are as expected with header_start
    at different positions, and that the table round-trips.
    """
    comments = ['comment 1', 'comment 2', 'comment 3']
    lines = ['# a b',
             '# comment 1',
             '# comment 2',
             '# comment 3',
             '1 2',
             '3 4']
    dat = ascii.read(lines, format='commented_header', fast_reader=fast)
    assert dat.meta['comments'] == comments
    assert dat.colnames == ['a', 'b']

    out = StringIO()
    ascii.write(dat, out, format='commented_header', fast_writer=fast)
    assert out.getvalue().splitlines() == lines

    lines.insert(1, lines.pop(0))
    dat = ascii.read(lines, format='commented_header', header_start=1, fast_reader=fast)
    assert dat.meta['comments'] == comments
    assert dat.colnames == ['a', 'b']

    lines.insert(2, lines.pop(1))
    dat = ascii.read(lines, format='commented_header', header_start=2, fast_reader=fast)
    assert dat.meta['comments'] == comments
    assert dat.colnames == ['a', 'b']
    dat = ascii.read(lines, format='commented_header', header_start=-2, fast_reader=fast)
    assert dat.meta['comments'] == comments
    assert dat.colnames == ['a', 'b']

    lines.insert(3, lines.pop(2))
    dat = ascii.read(lines, format='commented_header', header_start=-1, fast_reader=fast)
    assert dat.meta['comments'] == comments
    assert dat.colnames == ['a', 'b']

    lines = ['# a b',
             '1 2',
             '3 4']
    dat = ascii.read(lines, format='commented_header', fast_reader=fast)
    assert 'comments' not in dat.meta
    assert dat.colnames == ['a', 'b']

def test_probably_html():
    """
    Test the routine for guessing if a table input to ascii.read is probably HTML
    """
    for table in ('t/html.html',
                  'http://blah.com/table.html',
                  'https://blah.com/table.html',
                  'file://blah/table.htm',
                  'ftp://blah.com/table.html',
                  'file://blah.com/table.htm',
                  ' <! doctype html > hello world',
                  'junk < table baz> <tr foo > <td bar> </td> </tr> </table> junk',
                  ['junk < table baz>', ' <tr foo >', ' <td bar> ', '</td> </tr>', '</table> junk'],
                  (' <! doctype html > ', ' hello world'),
                   ):
        assert _probably_html(table) is True

    for table in ('t/html.htms',
                  'Xhttp://blah.com/table.html',
                  ' https://blah.com/table.htm',
                  'fole://blah/table.htm',
                  ' < doctype html > hello world',
                  'junk < tble baz> <tr foo > <td bar> </td> </tr> </table> junk',
                  ['junk < table baz>', ' <t foo >', ' <td bar> ', '</td> </tr>', '</table> junk'],
                  (' <! doctype htm > ', ' hello world'),
                  [[1, 2, 3]],
                   ):
        assert _probably_html(table) is False

@pytest.mark.parametrize('fast_reader', [True, False, 'force'])
def test_data_header_start(fast_reader):
    tests = [(['# comment',
               '',
               ' ',
               'skip this line',  # line 0
               'a b',  # line 1
               '1 2'],  # line 2
              [{'header_start': 1},
               {'header_start': 1, 'data_start': 2}
               ]
               ),

             (['# comment',
               '',
               ' \t',
               'skip this line',  # line 0
               'a b',  # line 1
               '',
               ' \t',
               'skip this line',  # line 2
               '1 2'],  # line 3
              [{'header_start': 1, 'data_start': 3}]),

             (['# comment',
               '',
               ' ',
               'a b',  # line 0
               '',
               ' ',
               'skip this line',  # line 1
               '1 2'],  # line 2
              [{'header_start': 0, 'data_start': 2},
               {'data_start': 2}])]

    for lines, kwargs_list in tests:
        for kwargs in kwargs_list:

            t = ascii.read(lines, format='basic', fast_reader=fast_reader,
                           guess=True, **kwargs)
            assert t.colnames == ['a', 'b']
            assert len(t) == 1
            assert np.all(t['a'] == [1])
            # Sanity check that the expected Reader is being used
            assert get_read_trace()[-1]['kwargs']['Reader'] is (
                ascii.Basic if (fast_reader is False) else ascii.FastBasic)

def test_table_with_no_newline():
    """
    Test that an input file which is completely empty fails in the expected way.
    Test that an input file with one line but no newline succeeds.
    """
    # With guessing
    table = BytesIO()
    with pytest.raises(ascii.InconsistentTableError):
        ascii.read(table)

    # Without guessing
    table = BytesIO()
    with pytest.raises(ValueError) as err:
        ascii.read(table, guess=False, fast_reader=False, format='basic')
    assert 'No header line found' in str(err.value)

    table = BytesIO()
    with pytest.raises(ValueError) as err:
        ascii.read(table, guess=False, fast_reader=True, format='fast_basic')
    assert 'Inconsistent data column lengths' in str(err.value)

    # Put a single line of column names but with no newline
    for kwargs in [dict(),
                   dict(guess=False, fast_reader=False, format='basic'),
                   dict(guess=False, fast_reader=True, format='fast_basic')]:
        table = BytesIO()
        table.write(b'a b')
        t = ascii.read(table, **kwargs)
        assert t.colnames == ['a', 'b']
        assert len(t) == 0

def test_path_object():
    fpath = pathlib.Path('t/simple.txt')
    data = ascii.read(fpath)

    assert len(data) == 2
    assert sorted(list(data.columns)) == ['test 1a', 'test2', 'test3', 'test4']
    assert data['test2'][1] == 'hat2'

def test_column_conversion_error():
    """
    Test that context information (upstream exception message) from column
    conversion error is provided.
    """
    ipac = """\
| col0   |
| double |
 1  2
"""
    with pytest.raises(ValueError) as err:
        ascii.read(ipac, guess=False, format='ipac')
    assert 'Column col0 failed to convert:' in str(err.value)

    with pytest.raises(ValueError) as err:
        ascii.read(['a b', '1 2'], guess=False, format='basic', converters={'a': []})
    assert 'no converters' in str(err.value)

def test_non_C_locale_with_fast_reader():
    """Test code that forces "C" locale while calling fast reader (#4364)"""
    current = locale.setlocale(locale.LC_ALL)

    try:
        if platform.system() == 'Darwin':
            locale.setlocale(locale.LC_ALL, str('de_DE'))
        else:
            locale.setlocale(locale.LC_ALL, str('de_DE.utf8'))

        for fast_reader in (True, False, {'use_fast_converter': False}, {'use_fast_converter': True}):
            t = ascii.read(['a b', '1.5 2'], format='basic', guess=False,
                           fast_reader=fast_reader)
            assert t['a'].dtype.kind == 'f'
    except locale.Error as e:
        pytest.skip('Locale error: {}'.format(e))
    finally:
        locale.setlocale(locale.LC_ALL, current)

def test_no_units_for_char_columns():
    '''Test that a char column of a Table is assigned no unit and not
    a dimensionless unit.'''
    t1 = Table([["A"]], names="B")
    out = StringIO()
    ascii.write(t1, out, format="ipac")
    t2 = ascii.read(out.getvalue(), format="ipac", guess=False)
    assert t2["B"].unit is None

def test_initial_column_fill_values():
    """Regression test for #5336, #5338."""

    class TestHeader(ascii.BasicHeader):
        def _set_cols_from_names(self):
            self.cols = [ascii.Column(name=x) for x in self.names]
            # Set some initial fill values
            for col in self.cols:
                col.fill_values = {'--': '0'}

    class Tester(ascii.Basic):
        header_class = TestHeader

    reader = ascii.get_reader(Reader=Tester)

    assert reader.read("""# Column definition is the first uncommented line
# Default delimiter is the space character.
a b c
# Data starts after the header column definition, blank lines ignored
-- 2 3
4 5 6 """)['a'][0] is np.ma.masked

def test_latex_no_trailing_backslash():
    """
    Test that latex/aastex file with no trailing backslash can be read.
    """
    lines = r"""
\begin{table}
\begin{tabular}{ccc}
a & b & c \\
1 & 1.0 & c \\ % comment
3\% & 3.0 & e  % comment
\end{tabular}
\end{table}
"""
    dat = ascii.read(lines, format='latex')
    assert dat.colnames == ['a', 'b', 'c']
    assert np.all(dat['a'] == ['1', r'3\%'])
    assert np.all(dat['c'] == ['c', 'e'])

def text_aastex_no_trailing_backslash():
    lines = r"""
\begin{deluxetable}{ccc}
\tablehead{\colhead{a} & \colhead{b} & \colhead{c}}
\startdata
1 & 1.0 & c \\
2 & 2.0 & d \\ % comment
3\% & 3.0 & e  % comment
\enddata
\end{deluxetable}
"""
    dat = ascii.read(lines, format='aastex')
    assert dat.colnames == ['a', 'b', 'c']
    assert np.all(dat['a'] == ['1', r'3\%'])
    assert np.all(dat['c'] == ['c', 'e'])

@pytest.mark.parametrize('encoding', ['utf8', 'latin1', 'cp1252'])
def test_read_with_encoding(tmpdir, encoding):
    data = {
        'commented_header': u'# à b è \n 1 2 héllo',
        'csv': u'à,b,è\n1,2,héllo'
    }

    testfile = str(tmpdir.join('test.txt'))
    for fmt, content in data.items():
        with open(testfile, 'w', encoding=encoding) as f:
            f.write(content)

        table = ascii.read(testfile, encoding=encoding)
        assert table.pformat() == [' à   b    è  ',
                                   '--- --- -----',
                                   '  1   2 héllo']

        for guess in (True, False):
            table = ascii.read(testfile, format=fmt, fast_reader=False,
                               encoding=encoding, guess=guess)
            assert table['è'].dtype.kind == 'U'
            assert table.pformat() == [' à   b    è  ',
                                       '--- --- -----',
                                       '  1   2 héllo']

def test_unsupported_read_with_encoding(tmpdir):
    # Fast reader is not supported, make sure it raises an exception
    with pytest.raises(ascii.ParameterError):
        ascii.read('t/simple3.txt', guess=False, fast_reader='force',
                   encoding='latin1', format='fast_csv')

def test_read_chunks_input_types():
    """
    Test chunked reading for different input types: file path, file object,
    and string input.
    """
    fpath = 't/test5.dat'
    t1 = ascii.read(fpath, header_start=1, data_start=3, )

    for fp in (fpath, open(fpath, 'r'), open(fpath, 'r').read()):
        t_gen = ascii.read(fp, header_start=1, data_start=3,
                           guess=False, format='fast_basic',
                           fast_reader={'chunk_size': 400, 'chunk_generator': True})
        ts = list(t_gen)
        for t in ts:
            for col, col1 in zip(t.columns.values(), t1.columns.values()):
                assert col.name == col1.name
                assert col.dtype.kind == col1.dtype.kind

        assert len(ts) == 4
        t2 = table.vstack(ts)
        assert np.all(t1 == t2)

    for fp in (fpath, open(fpath, 'r'), open(fpath, 'r').read()):
        # Now read the full table in chunks
        t3 = ascii.read(fp, header_start=1, data_start=3,
                        fast_reader={'chunk_size': 300})
        assert np.all(t1 == t3)

@pytest.mark.parametrize('masked', [True, False])
def test_read_chunks_formats(masked):
    """
    Test different supported formats for chunked reading.
    """
    t1 = simple_table(size=102, cols=10, kinds='fS', masked=masked)
    for i, name in enumerate(t1.colnames):
        t1.rename_column(name, 'col{}'.format(i + 1))

    # TO DO commented_header does not currently work due to the special-cased
    # implementation of header parsing.

    for format in 'tab', 'csv', 'no_header', 'rdb', 'basic':
        out = StringIO()
        ascii.write(t1, out, format=format)
        t_gen = ascii.read(out.getvalue(), format=format,
                           fast_reader={'chunk_size': 400, 'chunk_generator': True})
        ts = list(t_gen)
        for t in ts:
            for col, col1 in zip(t.columns.values(), t1.columns.values()):
                assert col.name == col1.name
                assert col.dtype.kind == col1.dtype.kind

        assert len(ts) > 4
        t2 = table.vstack(ts)
        assert np.all(t1 == t2)

        # Now read the full table in chunks
        t3 = ascii.read(out.getvalue(), format=format, fast_reader={'chunk_size': 400})
        assert np.all(t1 == t3)

def test_read_chunks_chunk_size_too_small():
    fpath = 't/test5.dat'
    with pytest.raises(ValueError) as err:
        ascii.read(fpath, header_start=1, data_start=3,
                   fast_reader={'chunk_size': 10})
    assert 'no newline found in chunk (chunk_size too small?)' in str(err)

def test_read_chunks_table_changes():
    """Column changes type or size between chunks.  This also tests the case with
    no final newline.
    """
    col = ['a b c'] + ['1.12334 xyz a'] * 50 + ['abcdefg 555 abc'] * 50
    table = '\n'.join(col)
    t1 = ascii.read(table, guess=False)
    t2 = ascii.read(table, fast_reader={'chunk_size': 100})

    # This also confirms that the dtypes are exactly the same, i.e.
    # the string itemsizes are the same.
    assert np.all(t1 == t2)

def test_read_non_ascii():
    """Test that pure-Python reader is used in case the file contains non-ASCII characters
    in it.
    """
    table = Table.read(['col1, col2', '\u2119, \u01b4', '1, 2'], format='csv')
    assert np.all(table['col1'] == ['\u2119', '1'])
    assert np.all(table['col2'] == ['\u01b4', '2'])

@pytest.mark.parametrize('enable', [True, False, 'force'])
def test_kwargs_dict_guess(enable):
    """Test that fast_reader dictionary is preserved through guessing sequence.
    """
    # Fails for enable=(True, 'force') - #5578
    ascii.read('a\tb\n 1\t2\n3\t 4.0', fast_reader=dict(enable=enable))
    assert get_read_trace()[-1]['kwargs']['Reader'] is (
        ascii.Tab if (enable is False) else ascii.FastTab)
    for k in get_read_trace():
        if not k.get('status', 'Disabled').startswith('Disabled'):
            assert k.get('kwargs').get('fast_reader').get('enable') is enable

max_stars_repo_name: jbkalmbach/astropy

max_issues_repo_name: jbkalmbach/astropy

max_forks_repo_name: jbkalmbach/astropy


text:
import numpy as np
from myModules.dct2pack import dct2,idct2

def compress(image,low_compression,high_compression):
    #Image processing
    ###Compression
    #Convert from RGB to YUV color space
    h,w,_=image.shape
    YUV=np.zeros((h,w,3))
    YUV[:,:,0]=image[:,:,0]*0.299+image[:,:,1]*0.587+image[:,:,2]*0.114
    YUV[:,:,1]=image[:,:,0]*(-0.14713)+image[:,:,1]*(-0.28886)+image[:,:,2]*0.436
    YUV[:,:,2]=image[:,:,0]*0.615+image[:,:,1]*(-0.51499)+image[:,:,2]*(-0.10001)
    #Find DCT for each 8x8 block
    DCT_YUV=np.zeros((h,w,3))
    for i in range(0,h,8):
        for j in range(0,w,8):
            #Find DCT
            #Y(brightness) is dominant in determining the quality of visual effect so use low_compression here
            DCT_YUV[i:i+8,j:j+8,0]=np.round(dct2(YUV[i:i+8,j:j+8,0])/low_compression).astype(int)
            DCT_YUV[i:i+8,j:j+8,1]=np.round(dct2(YUV[i:i+8,j:j+8,1])/high_compression).astype(int)
            DCT_YUV[i:i+8,j:j+8,2]=np.round(dct2(YUV[i:i+8,j:j+8,2])/high_compression).astype(int)
    #RUN Length Encoding
    RLE_YUV=[[],[],[]] #[[integer,count,integer,count,...],...]
    #Zig-Zagging
    for i in range(0,h,8):
        for j in range(0,w,8):
            for c in [0,1,2]:
                ih,iw=i,j #index vertical, horizontal
                vh,vw=-1,1 #velocity vertical, horizontal
                data=DCT_YUV[i,j,c]
                cnt=0
                RLE_YUV[c].append(int(data))
                while True:
                    if DCT_YUV[ih,iw,c]==data:
                        cnt+=1
                    else:
                        RLE_YUV[c].append(cnt)
                        data=DCT_YUV[ih,iw,c]
                        RLE_YUV[c].append(int(data))
                        cnt=1
                    if ih==i+7 and iw==j+7:
                    #Traversed this 8x8 block
                        RLE_YUV[c].append(cnt)
                        break
                    if not (i<=ih+vh<i+8):
                    #Hit the upper or lower boundary
                        #Right
                        iw+=1
                        vw=-vw
                        vh=-vh
                    elif not (j<=iw+vw<j+8):
                    #Hit the left or right boundary
                        #Down
                        ih+=1
                        vw=-vw
                        vh=-vh
                    else:
                        ih+=vh
                        iw+=vw
    return RLE_YUV

def decompress(RLE_YUV,size,low_compression,high_compression):
    ###Decompression
    #file size (width,height), RLE_YUV and quantization tables are known beforehand
    h,w,_=size
    DCT_YUV_2=np.zeros((h,w,3))
    for c in [0,1,2]:
        RLE_ind=0
        for i in range(0,h,8):
            for j in range(0,w,8):
                ih,iw=i,j #vertical, horizontal index
                vh,vw=-1,1 #vertical, horizontal velocity
                cnt=0
                #Zig-Zagging
                while True:
                    if cnt==0:
                        data,cnt=RLE_YUV[c][RLE_ind],RLE_YUV[c][RLE_ind+1]
                        RLE_ind+=2
                    DCT_YUV_2[ih,iw,c]=data
                    cnt-=1
                    if ih==i+7 and iw==j+7:
                    #Reached the last pixel of this 8x8 block
                        break
                    if not (i<=ih+vh<i+8):
                    #Hit the upper or lower boundary
                        #Right
                        iw+=1
                        vw=-vw
                        vh=-vh
                    elif not (j<=iw+vw<j+8):
                    #Hit the left or right boundary
                        #Down
                        ih+=1
                        vw=-vw
                        vh=-vh
                    else:
                        ih+=vh
                        iw+=vw
    YUV_2=np.zeros((h,w,3))
    #iDCT
    for i in range(0,h,8):
        for j in range(0,w,8):
            YUV_2[i:i+8,j:j+8,0]=idct2(DCT_YUV_2[i:i+8,j:j+8,0]*low_compression)
            YUV_2[i:i+8,j:j+8,1]=idct2(DCT_YUV_2[i:i+8,j:j+8,1]*high_compression)
            YUV_2[i:i+8,j:j+8,2]=idct2(DCT_YUV_2[i:i+8,j:j+8,2]*high_compression)
    #Store decompressed image
    RGB=np.zeros((h,w,3))
    RGB[:,:,0]=YUV_2[:,:,0]+YUV_2[:,:,2]*1.13983
    RGB[:,:,1]=YUV_2[:,:,0]+YUV_2[:,:,1]*(-0.39465)+YUV_2[:,:,2]*(-0.58060)
    RGB[:,:,2]=YUV_2[:,:,0]+YUV_2[:,:,1]*2.03211
    #Regulate the result
    RGB=np.maximum(np.minimum(RGB,255),0).astype('uint8')

    return RGB

max_stars_repo_path: 4.Simple JPEG/myModules/JPEG.py

max_issues_repo_path: 4.Simple JPEG/myModules/JPEG.py

max_forks_repo_path: 4.Simple JPEG/myModules/JPEG.py


text:
#!/usr/bin/env python
# coding: utf-8

# # Desafio 6
#
# Neste desafio, vamos praticar _feature engineering_, um dos processos mais importantes e trabalhosos de ML. Utilizaremos o _data set_ [Countries of the world](https://www.kaggle.com/fernandol/countries-of-the-world), que contém dados sobre os 227 países do mundo com informações sobre tamanho da população, área, imigração e setores de produção.
#
# > Obs.: Por favor, não modifique o nome das funções de resposta.

# ## _Setup_ geral

# In[1]:

import pandas as pd
import numpy as np
import seaborn as sns
import sklearn as sk

from sklearn.datasets import fetch_20newsgroups
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, KBinsDiscretizer, OneHotEncoder
from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer

# In[2]:

# Algumas configurações para o matplotlib.
#%matplotlib inline
from IPython.core.pylabtools import figsize
figsize(12, 8)
sns.set()

# In[3]:

countries = pd.read_csv('countries.csv', decimal=',')

# In[4]:

new_column_names = [
    'Country', 'Region', 'Population', 'Area', 'Pop_density', 'Coastline_ratio',
    'Net_migration', 'Infant_mortality', 'GDP', 'Literacy', 'Phones_per_1000',
    'Arable', 'Crops', 'Other', 'Climate', 'Birthrate', 'Deathrate', 'Agriculture',
    'Industry', 'Service'
]

countries.columns = new_column_names

countries.head(5)

# ## Observações
#
# Esse _data set_ ainda precisa de alguns ajustes iniciais. Primeiro, note que as variáveis numéricas estão usando vírgula como separador decimal e estão codificadas como strings. Corrija isso antes de continuar: transforme essas variáveis em numéricas adequadamente.
#
# Além disso, as variáveis `Country` e `Region` possuem espaços a mais no começo e no final da string. Você pode utilizar o método `str.strip()` para remover esses espaços.

# ## Inicia sua análise a partir daqui

# In[5]:

# Sua análise começa aqui.
pd.DataFrame({
    'dtype': countries.dtypes,
    'nulls': countries.isna().sum(),
    'nulls (%)': countries.isna().mean()
})

# ## Questão 1
#
# Quais são as regiões (variável `Region`) presentes no _data set_? Retorne uma lista com as regiões únicas do _data set_ com os espaços à frente e atrás da string removidos (mas mantenha pontuação: ponto, hífen etc) e ordenadas em ordem alfabética.

# In[16]:

def q1():
    return sorted(map(lambda r: r.strip(), countries['Region'].unique()))

q1()

# ## Questão 2
#
# Discretizando a variável `Pop_density` em 10 intervalos com `KBinsDiscretizer`, seguindo o encode `ordinal` e estratégia `quantile`, quantos países se encontram acima do 90º percentil? Responda como um único escalar inteiro.

# In[7]:

def q2():
    kbins = KBinsDiscretizer(n_bins=10, encode='ordinal', strategy='quantile')
    pop_density_discrete = kbins.fit_transform(countries['Pop_density'].values.reshape(-1, 1))
    return int((pop_density_discrete==9).sum())

q2()

# # Questão 3
#
# Se codificarmos as variáveis `Region` e `Climate` usando _one-hot encoding_, quantos novos atributos seriam criados? Responda como um único escalar.

# In[19]:

def q3():
    ohe = OneHotEncoder()
    return int(ohe.fit_transform(countries[['Region', 'Climate']].fillna({'Climate': 0})).shape[1])

q3()

# ## Questão 4
#
# Aplique o seguinte _pipeline_:
#
# 1. Preencha as variáveis do tipo `int64` e `float64` com suas respectivas medianas.
# 2. Padronize essas variáveis.
#
# Após aplicado o _pipeline_ descrito acima aos dados (somente nas variáveis dos tipos especificados), aplique o mesmo _pipeline_ (ou `ColumnTransformer`) ao dado abaixo. Qual o valor da variável `Arable` após o _pipeline_? Responda como um único float arredondado para três casas decimais.

# In[9]:

test_country = [
    'Test Country', 'NEAR EAST', -0.19032480757326514,
    -0.3232636124824411, -0.04421734470810142, -0.27528113360605316,
    0.13255850810281325, -0.8054845935643491, 1.0119784924248225,
    0.6189182532646624, 1.0074863283776458, 0.20239896852403538,
    -0.043678728558593366, -0.13929748680369286, 1.3163604645710438,
    -0.3699637766938669, -0.6149300604558857, -0.854369594993175,
    0.263445277972641, 0.5712416961268142
]

# In[10]:

def q4():
    numeric_features = countries.select_dtypes('number').columns
    numeric_transformer = Pipeline(steps=[
        ('imputer', SimpleImputer(strategy='median')),
        ('scaler', StandardScaler())
    ])

    preprocessor = ColumnTransformer(
        transformers=[('num', numeric_transformer, numeric_features)],
        remainder='drop'
    )

    preprocessor.fit(countries)

    test = pd.DataFrame([test_country], columns=countries.columns)
    arable = preprocessor.transform(test)[0][numeric_features.get_loc('Arable')]
    return float(round(arable, 3))

q4()

# ## Questão 5
#
# Descubra o número de _outliers_ da variável `Net_migration` segundo o método do _boxplot_, ou seja, usando a lógica:
#
# $$x \notin [Q1 - 1.5 \times \text{IQR}, Q3 + 1.5 \times \text{IQR}] \Rightarrow x \text{ é outlier}$$
#
# que se encontram no grupo inferior e no grupo superior.
#
# Você deveria remover da análise as observações consideradas _outliers_ segundo esse método? Responda como uma tupla de três elementos `(outliers_abaixo, outliers_acima, removeria?)` ((int, int, bool)).

# In[12]:

def q5():
    net_migration = countries['Net_migration'].dropna()

    q1, q3 = np.quantile(net_migration, [0.25, 0.75], axis=0)
    iqr = q3 - q1

    lower_whisker = q1 - 1.5*iqr
    upper_whisker = q3 + 1.5*iqr

    outliers_abaixo = int((net_migration < lower_whisker).sum())
    outliers_acima  = int((net_migration > upper_whisker).sum())

    total_outliers = outliers_abaixo + outliers_acima
    return (outliers_abaixo, outliers_acima, (total_outliers / countries.shape[0]) <= 0.1)

q5()

# ## Questão 6
# Para as questões 6 e 7 utilize a biblioteca `fetch_20newsgroups` de datasets de test do `sklearn`
#
# Considere carregar as seguintes categorias e o dataset `newsgroups`:
#
# ```
# categories = ['sci.electronics', 'comp.graphics', 'rec.motorcycles']
# newsgroup = fetch_20newsgroups(subset="train", categories=categories, shuffle=True, random_state=42)
# ```
#
#
# Aplique `CountVectorizer` ao _data set_ `newsgroups` e descubra o número de vezes que a palavra _phone_ aparece no corpus. Responda como um único escalar.

# In[13]:

categories = ['sci.electronics', 'comp.graphics', 'rec.motorcycles']
newsgroup = fetch_20newsgroups(subset='train', categories=categories, shuffle=True, random_state=42)

# In[14]:

def q6():
    vectorizer = CountVectorizer().fit(newsgroup.data)
    data = vectorizer.transform(newsgroup.data)
    return int(data[:, vectorizer.vocabulary_['phone']].sum())

q6()

# ## Questão 7
#
# Aplique `TfidfVectorizer` ao _data set_ `newsgroups` e descubra o TF-IDF da palavra _phone_. Responda como um único escalar arredondado para três casas decimais.

# In[15]:

def q7():
    vectorizer = TfidfVectorizer().fit(newsgroup.data)
    data = vectorizer.transform(newsgroup.data)
    return float(data[:, vectorizer.vocabulary_['phone']].sum().round(3))

q7()

# In[ ]:


text:
#!/usr/bin/env python

import sys, copy, random
import json
import numpy as np
from randomPlayer import RandomPlayer
import play

class MCPlayer:
    def __init__(self, nPlay):
        self._nPlay = nPlay
        self._rp = RandomPlayer()
        self._rand = random.Random()
        self._epsSame = 1e-6

    def __str__(self):
        return "%s(nPlay=%d)" % (self.__class__.__name__, self._nPlay)

    def setSeed(self, seed):
        self._rp.setSeed(seed)
        if seed is not None:
            self._rand.seed(seed+1)
        else:
            self._rand.seed(seed)

    def move(self, ttt):
        bestQ = -1e99
        qs = []
        vm = ttt.validMoves()
        for m in vm:
            q = self.moveQuality(ttt, m)
            # print ("MOVE:", q)
            qs.append(q)
            if q > bestQ:
                bestQ = q

        bestMoves = []
        for iMove, q in enumerate(qs):
            if abs(q-bestQ) < self._epsSame:
                bestMoves.append(vm[iMove])

        return self._rand.choice(bestMoves)

    def moveQuality(self, ttt, m):
        scores = []
        for _ in range(self._nPlay):
            scores.append(play.simGame(self._rp, self._rp, ttt, m))

        scores = np.array(scores)
        return scores.mean()

if __name__ == "__main__":
    from ticTacToe import TicTacToe
    from mmPlayer import MMPlayer
    play.play(TicTacToe, MCPlayer(1), MCPlayer(100), True)


text:
import argparse
from pathlib import Path
from PIL import Image
from fall_prediction import Fall_prediction
import numpy as np
import json
import time
import yaml

class JsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.integer):
            return int(obj)
        if isinstance(obj, np.floating):
            return float(obj)
        if isinstance(obj, np.ndarray):
            return obj.tolist()

        return super(JsonEncoder, self).default(obj)

parser = argparse.ArgumentParser()
parser.add_argument("--image_1", type=Path, help='Path to the First Image', required=True)
parser.add_argument("--image_2", type=Path, help='Path to the Second Image', required=True)
parser.add_argument("--image_3", type=Path, help='Path to the Third Image')

p = parser.parse_args()
img1 = Image.open(p.image_1)
img2 = Image.open(p.image_2)
img3 = Image.open(p.image_3) if p.image_3 else None

response = Fall_prediction(img1, img2, img3)

if response:

    print("There is", response['category'])
    print("Confidence :", response['confidence'])
    print("Angle : ", response['angle'])
    print("Keypoint_corr :", response['keypoint_corr'])

    time_str = time.strftime("%Y%m%d-%H%M%S")
    json_str = json.dumps(response, cls=JsonEncoder)

    with open(f"tmp/{time_str}.yaml","w",) as file:
        yaml.dump(json.loads(json_str), file)
else:
     print("There is no fall detetcion...")

max_stars_repo_name: bhavikamkanani/fall-detection

max_issues_repo_name: bhavikamkanani/fall-detection

max_forks_repo_name: ivelin/fall-detection


text:
import torch
import torch.nn as nn
import torch.nn.functional as F
import dgl
import dgl.function as fn
import copy
import numpy as np
from functools import partial
from dgl.nn.pytorch.conv import RelGraphConv
from basemodel import GraphAdjModel
from utils import map_activation_str_to_layer, split_and_batchify_graph_feats

class RGINLayer(nn.Module):
    def __init__(self, in_feat, out_feat, num_rels, regularizer="basis", num_bases=None, act_func="relu", dropout=0.0):
        super(RGINLayer, self).__init__()
        self.rgc_layer = RelGraphConv(
            in_feat=in_feat, out_feat=out_feat, num_rels=num_rels,
            regularizer=regularizer, num_bases=num_bases,
            activation=None, self_loop=True, dropout=0.0)
        self.mlp = nn.Sequential(
            nn.Linear(out_feat, out_feat),
            # nn.BatchNorm1d(out_feat),
            map_activation_str_to_layer(act_func),
            nn.Linear(out_feat, out_feat),
            map_activation_str_to_layer(act_func))
        self.drop = nn.Dropout(dropout)

        # init
        if hasattr(self.rgc_layer, "weight") and self.rgc_layer.weight is not None:
            nn.init.normal_(self.rgc_layer.weight, 0.0, 1/(out_feat)**0.5)
        if hasattr(self.rgc_layer, "w_comp") and self.rgc_layer.w_comp is not None:
            nn.init.normal_(self.rgc_layer.w_comp, 0.0, 1/(out_feat)**0.5)
        if hasattr(self.rgc_layer, "loop_weight") and self.rgc_layer.loop_weight is not None:
            nn.init.normal_(self.rgc_layer.loop_weight, 0.0, 1/(out_feat)**0.5)
        if hasattr(self.rgc_layer, "h_bias") and self.rgc_layer.h_bias is not None:
            nn.init.zeros_(self.rgc_layer.h_bias)
        for m in self.mlp.modules():
            if isinstance(m, nn.Linear):
                nn.init.normal_(m.weight, 0.0, 1/(out_feat)**0.5)
                if hasattr(m, "bias") and m.bias is not None:
                    nn.init.zeros_(m.bias)
            elif isinstance(m, nn.BatchNorm1d):
                nn.init.ones_(m.weight)
                nn.init.zeros_(m.bias)

    def forward(self, g, x, etypes):
        g = self.rgc_layer(g, x, etypes, norm=None)
        g = self.mlp(g)
        g = self.drop(g)
        return g

class RGIN(GraphAdjModel):
    def __init__(self, config):
        super(RGIN, self).__init__(config)

        # create networks
        p_emb_dim, g_emb_dim = self.get_emb_dim()
        self.g_net, g_dim = self.create_net(
            name="graph", input_dim=g_emb_dim, hidden_dim=config["rgin_hidden_dim"],
            num_layers=config["rgin_graph_num_layers"],
            num_rels=self.max_ngel, num_bases=config["rgin_num_bases"], regularizer=config["rgin_regularizer"],
            act_func=self.act_func, dropout=self.dropout)
        self.p_net, p_dim = (self.g_net, g_dim) if self.share_arch else self.create_net(
            name="pattern", input_dim=p_emb_dim, hidden_dim=config["rgin_hidden_dim"],
            num_layers=config["rgin_pattern_num_layers"],
            num_rels=self.max_npel, num_bases=config["rgin_num_bases"], regularizer=config["rgin_regularizer"],
            act_func=self.act_func, dropout=self.dropout)

        if self.add_enc:
            p_enc_dim, g_enc_dim = self.get_enc_dim()
            p_dim += p_enc_dim
            g_dim += g_enc_dim
        if self.add_degree:
            p_dim += 1
            g_dim += 1
        self.predict_net = self.create_predict_net(config["predict_net"],
            pattern_dim=p_dim, graph_dim=g_dim, hidden_dim=config["predict_net_hidden_dim"],
            num_heads=config["predict_net_num_heads"], recurrent_steps=config["predict_net_recurrent_steps"],
            mem_len=config["predict_net_mem_len"], mem_init=config["predict_net_mem_init"])

    def create_net(self, name, input_dim, **kw):
        num_layers = kw.get("num_layers", 1)
        hidden_dim = kw.get("hidden_dim", 64)
        num_rels = kw.get("num_rels", 1)
        num_bases = kw.get("num_bases", 8)
        regularizer = kw.get("regularizer", "basis")
        act_func = kw.get("act_func", "relu")
        dropout = kw.get("dropout", 0.0)

        rgins = nn.ModuleList()
        for i in range(num_layers):
            rgins.add_module("%s_rgi%d" % (name, i), RGINLayer(
                in_feat=hidden_dim if i > 0 else input_dim, out_feat=hidden_dim, num_rels=num_rels,
                regularizer=regularizer, num_bases=num_bases,
                act_func=act_func, dropout=dropout))

        return rgins, hidden_dim

    def increase_input_size(self, config):
        old_p_enc_dim, old_g_enc_dim = self.get_enc_dim()
        old_max_npel, old_max_ngel = self.max_npel, self.max_ngel
        super(RGIN, self).increase_input_size(config)
        new_p_enc_dim, new_g_enc_dim = self.get_enc_dim()
        new_max_npel, new_max_ngel = self.max_npel, self.max_ngel

        # increase networks
        if new_max_ngel != old_max_ngel:
            for g_rgin in self.g_net:
                num_bases = g_rgin.rgc_layer.num_bases
                device = g_rgin.rgc_layer.weight.device
                regularizer = g_rgin.rgc_layer.regularizer
                if regularizer == "basis":
                    if num_bases < old_max_ngel:
                        new_w_comp = nn.Parameter(
                            torch.zeros((new_max_ngel, num_bases), dtype=torch.float32, device=device, requires_grad=True))
                        with torch.no_grad():
                            new_w_comp[:old_max_ngel].data.copy_(g_rgin.rgc_layer.w_comp)
                    else:
                        new_w_comp = nn.Parameter(
                            torch.zeros((new_max_ngel, num_bases), dtype=torch.float32, device=device, requires_grad=True))
                        with torch.no_grad():
                            ind = np.diag_indices(num_bases)
                            new_w_comp[ind[0], ind[1]] = 1.0
                    del g_rgin.rgc_layer.w_comp
                    g_rgin.rgc_layer.w_comp = new_w_comp
                elif regularizer == "bdd":
                    new_weight = nn.Parameter(
                        torch.zeros((new_max_ngel, g_rgin.rgc_layer.weight.size(1)),
                            dtype=torch.float32, device=device, requires_grad=True))
                    with torch.no_grad():
                        new_weight[:old_max_ngel].data.copy_(g_rgin.rgc_layer.weight)
                    del g_rgin.rgc_layer.weight
                    g_rgin.rgc_layer.weight = new_weight
                else:
                    raise NotImplementedError
        if self.share_arch:
            del self.p_net
            self.p_net = self.g_net
        elif new_max_npel != old_max_npel:
            for p_rgin in self.p_net:
                num_bases = p_rgin.rgc_layer.num_bases
                device = p_rgin.rgc_layer.weight.device
                regularizer = p_rgin.rgc_layer.regularizer
                if regularizer == "basis":
                    if num_bases < old_max_npel:
                        new_w_comp = nn.Parameter(
                            torch.zeros((new_max_npel, num_bases), dtype=torch.float32, device=device, requires_grad=True))
                        with torch.no_grad():
                            new_w_comp[:old_max_npel].data.copy_(p_rgin.rgc_layer.w_comp)
                    else:
                        new_w_comp = nn.Parameter(
                            torch.zeros((new_max_npel, num_bases), dtype=torch.float32, device=device, requires_grad=True))
                        with torch.no_grad():
                            ind = np.diap_indices(num_bases)
                            new_w_comp[ind[0], ind[1]] = 1.0
                    del p_rgin.rgc_layer.w_comp
                    p_rgin.rgc_layer.w_comp = new_w_comp
                elif regularizer == "bdd":
                    new_weight = nn.Parameter(
                        torch.zeros((max_npel, p_rgin.rgc_layer.weight.size(1)), dtype=torch.float32, device=device, requires_grad=True))
                    with torch.no_grad():
                        new_weight[:old_max_npel].data.copy_(p_rgin.rgc_layer.weight)
                    del p_rgin.rgc_layer.weight
                    p_rgin.rgc_layer.weight = new_weight
                else:
                    raise NotImplementedError

        # increase predict network
        if self.add_enc and (new_g_enc_dim != old_g_enc_dim or new_p_enc_dim != old_p_enc_dim):
            self.predict_net.increase_input_size(
                self.predict_net.pattern_dim+new_p_enc_dim-old_p_enc_dim,
                self.predict_net.graph_dim+new_g_enc_dim-old_g_enc_dim)

    def increase_net(self, config):
        p_emb_dim, g_emb_dim = self.get_emb_dim()
        g_net, g_dim = self.create_net(
            name="graph", input_dim=g_emb_dim, hidden_dim=config["rgin_hidden_dim"],
            num_layers=config["rgin_graph_num_layers"],
            num_rels=self.max_ngel, num_bases=config["rgin_num_bases"], regularizer=config["rgin_regularizer"],
            act_func=self.act_func, dropout=self.dropout)
        assert len(g_net) >= len(self.g_net)
        with torch.no_grad():
            for old_g_rgin, new_g_rgin in zip(self.g_net, g_net):
                new_g_rgin.load_state_dict(old_g_rgin.state_dict())
        del self.g_net
        self.g_net = g_net

        if self.share_arch:
            self.p_net = self.g_net
        else:
            p_net, p_dim = self.create_net(
                name="pattern", input_dim=p_emb_dim, hidden_dim=config["rgin_hidden_dim"],
                num_layers=config["rgin_pattern_num_layers"],
                num_rels=self.max_npel, num_bases=config["rgin_num_bases"], regularizer=config["rgin_regularizer"],
                act_func=self.act_func, dropout=self.dropout)
            assert len(p_net) >= len(self.p_net)
            with torch.no_grad():
                for old_p_rgin, new_p_rgin in zip(self.p_net, p_net):
                    new_p_rgin.load_state_dict(old_p_rgin.state_dict())
            del self.p_net
            self.p_net = p_net

    def forward(self, pattern, pattern_len, graph, graph_len):
        bsz = pattern_len.size(0)

        gate = self.get_filter_gate(pattern, pattern_len, graph, graph_len)
        zero_mask = (gate == 0) if gate is not None else None
        pattern_emb, graph_emb = self.get_emb(pattern, pattern_len, graph, graph_len)
        if zero_mask is not None:
            graph_emb.masked_fill_(zero_mask, 0.0)

        pattern_output = pattern_emb
        for p_rgin in self.p_net:
            o = p_rgin(pattern, pattern_output, pattern.edata["label"])
            pattern_output = o + pattern_output

        graph_output = graph_emb
        for g_rgin in self.g_net:
            o = g_rgin(graph, graph_output, graph.edata["label"])
            graph_output = o + graph_output
            if zero_mask is not None:
                graph_output.masked_fill_(zero_mask, 0.0)

        if self.add_enc and self.add_degree:
            pattern_enc, graph_enc = self.get_enc(pattern, pattern_len, graph, graph_len)
            if zero_mask is not None:
                graph_enc.masked_fill_(zero_mask, 0.0)
            pattern_output = torch.cat([pattern_enc, pattern_output, pattern.ndata["indeg"].unsqueeze(-1)], dim=1)
            graph_output = torch.cat([graph_enc, graph_output, graph.ndata["indeg"].unsqueeze(-1)], dim=1)
        elif self.add_enc:
            pattern_enc, graph_enc = self.get_enc(pattern, pattern_len, graph, graph_len)
            if zero_mask is not None:
                graph_enc.masked_fill_(zero_mask, 0.0)
            pattern_output = torch.cat([pattern_enc, pattern_output], dim=1)
            graph_output = torch.cat([graph_enc, graph_output], dim=1)
        elif self.add_degree:
            pattern_output = torch.cat([pattern_output, pattern.ndata["indeg"].unsqueeze(-1)], dim=1)
            graph_output = torch.cat([graph_output, graph.ndata["indeg"].unsqueeze(-1)], dim=1)

        pred = self.predict_net(
            split_and_batchify_graph_feats(pattern_output, pattern_len)[0], pattern_len,
            split_and_batchify_graph_feats(graph_output, graph_len)[0], graph_len)

        return pred

max_stars_repo_name: HKUST-KnowComp/NeuralSubIsoCnt

max_issues_repo_name: HKUST-KnowComp/NeuralSubIsoCnt

max_forks_repo_name: HKUST-KnowComp/NeuralSubIsoCnt