<commit_before># -*- coding: utf-8 -*-


import numpy as np
import hashlib

try:
    import contextlib2 as contextlib
    FileNotFoundError = IOError
except ImportError:
    import contextlib

try:
    import matplotlib.pyplot as plt
    HAS_MPL = True
except ImportError:
    HAS_MPL = False

try:
    import pyregion
    HAS_PYREGION = True
except ImportError:
    HAS_PYREGION = False

from srttools.imager import ScanSet
from srttools.scan import Scan
from srttools.calibration import CalibratorTable
from srttools.calibration import HAS_STATSM
from srttools.read_config import read_config
from srttools.imager import main_imager, main_preprocess, \
    _excluded_regions_from_args
from srttools.simulate import simulate_map
from srttools.global_fit import display_intermediate
from srttools.io import mkdir_p
from srttools.interactive_filter import intervals
import copy
import os
import glob
import astropy.units as u
import shutil
import pytest
from astropy import log
import subprocess as sp
import astropy
from astropy.logger import logging

try:
    from tqdm import tqdm
except ImportError:
    def tqdm(x):
        return x

@pytest.fixture()
def logger():
    logger = logging.getLogger('Some.Logger')
    logger.setLevel(logging.INFO)

    return logger


np.random.seed(1241347)


def _md5(file):
    with open(file, 'rb') as fobj:
        string = fobj.read()

    return hashlib.md5(string).hexdigest()


def _2d_gauss(x, y, sigma=2.5 / 60.):
    """A Gaussian beam"""
    return np.exp(-(x ** 2 + y ** 2) / (2 * sigma**2))


def gauss_src_func(x, y):
    return 25 * _2d_gauss(x, y, sigma=2.5 / 60)


def sim_config_file(filename, add_garbage=False, prefix=None):
    """Create a sample config file, to be modified by hand."""
    string0 = """
[local]
workdir : .
datadir : .
productdir : test_image

[analysis]
projection : ARC
interpolation : spline
prefix : test_
list_of_directories :
    gauss_ra
    gauss_dec
    defective
"""
    string1 = """
calibrator_directories :
    calibration
"""
    string2 = """

skydip_directories :
    gauss_skydip

noise_threshold : 5

pixel_size : 0.8

[debugging]

debug_file_format : png

"""
    if prefix is None:
        prefix = os.getcwd()
    import tempfile
    string = string0
    with open(filename, 'w') as fobj:
        print(string0, file=fobj)
        if add_garbage:
            for _ in range(100):
                garbage = '    ' + \
                    tempfile.NamedTemporaryFile(prefix=prefix).name[1:]
                print(garbage, file=fobj)
                string += garbage + '\n'
        print(string1, file=fobj)
        string += string1
        if add_garbage:
            for _ in range(100):
                garbage = '    ' + \
                    tempfile.NamedTemporaryFile(prefix=prefix).name[1:]
                print(garbage, file=fobj)
                string += garbage + '\n'
        print(string2, file=fobj)
        string += string2
    return string


def sim_map(obsdir_ra, obsdir_dec):
    simulate_map(count_map=gauss_src_func,
                 length_ra=30.,
                 length_dec=30.,
                 outdir=(obsdir_ra, obsdir_dec), mean_ra=180,
                 mean_dec=45, speed=1.5,
                 spacing=0.5, srcname='Dummy', channel_ratio=0.8,
                 baseline="flat")


class TestScanSet(object):
    @classmethod
    def setup_class(klass):
        import os

        klass.curdir = os.path.dirname(__file__)
        klass.datadir = os.path.join(klass.curdir, 'data')
        klass.sim_dir = os.path.join(klass.datadir, 'sim')
        klass.prodir_ra = os.path.join(klass.datadir,
                                       'sim', 'test_image', 'gauss_ra')
        klass.prodir_dec = os.path.join(klass.datadir,
                                        'sim', 'test_image', 'gauss_dec')

        klass.obsdir_ra = os.path.join(klass.datadir, 'sim', 'gauss_ra')
        klass.obsdir_dec = os.path.join(klass.datadir, 'sim', 'gauss_dec')
        klass.config_file = \
            os.path.abspath(os.path.join(klass.sim_dir, 'test_config_sim.ini'))
        klass.caldir = os.path.join(klass.datadir, 'sim', 'calibration')
        klass.simulated_flux = 0.25
        # First off, simulate a beamed observation  -------

        sim_config_file(klass.config_file, add_garbage=True,
                        prefix="./")

        if (not os.path.exists(klass.obsdir_ra)) or \
                (not os.path.exists(klass.obsdir_dec)):
            mkdir_p(klass.obsdir_ra)
            mkdir_p(klass.obsdir_dec)
            log.info('Fake map: Point-like (but Gaussian beam shape), '
                  '{} Jy.'.format(klass.simulated_flux))
            sim_map(klass.obsdir_ra, klass.obsdir_dec)

        defective_dir = os.path.join(klass.sim_dir, 'defective')
        if not os.path.exists(defective_dir):
            shutil.copytree(os.path.join(klass.datadir, 'calibrators'),
                            defective_dir)

        # Copy skydip scan
        skydip_dir = os.path.join(klass.datadir, 'gauss_skydip')
        new_skydip_dir = os.path.join(klass.sim_dir, 'gauss_skydip')
        if os.path.exists(skydip_dir) and not os.path.exists(new_skydip_dir):
            shutil.copytree(skydip_dir, new_skydip_dir)
        caltable = CalibratorTable()
        caltable.from_scans(glob.glob(os.path.join(klass.caldir,
                                                   '*.fits')), debug=True)

        caltable.update()
        klass.calfile = os.path.join(klass.datadir, 'calibrators.hdf5')
        caltable.write(klass.calfile, overwrite=True)

        klass.config = read_config(klass.config_file)
        klass.raonly = os.path.abspath(os.path.join(klass.datadir,
                                                    'test_raonly.ini'))
        klass.deconly = os.path.abspath(os.path.join(klass.datadir,
                                                     'test_deconly.ini'))

        if HAS_PYREGION:
            excluded_xy, excluded_radec = \
                _excluded_regions_from_args([os.path.join(klass.datadir,
                                                          "center.reg")])
        else:
            excluded_xy, excluded_radec = None, None

        klass.scanset = ScanSet(klass.config_file, nosub=False,
                                norefilt=False, plot=True,
                                debug=True, avoid_regions=excluded_radec)
        klass.scanset.write('test.hdf5', overwrite=True)

        klass.stdinfo = {}
        klass.stdinfo['FLAG'] = None
        klass.stdinfo['zap'] = intervals()
        klass.stdinfo['base'] = intervals()
        klass.stdinfo['fitpars'] = np.array([0, 0])

        def scan_no(scan_str):
            basename = os.path.splitext(os.path.basename(scan_str))[0]
            return int(basename.replace('Dec', '').replace('Ra', ''))

        klass.dec_scans = \
            dict([(scan_no(s), s)
                 for s in klass.scanset.scan_list if 'Dec' in s])
        klass.ra_scans = \
            dict([(scan_no(s), s)
                 for s in klass.scanset.scan_list if 'Ra' in s])
        klass.n_ra_scans = max(list(klass.ra_scans.keys()))
        klass.n_dec_scans = max(list(klass.dec_scans.keys()))

        if HAS_MPL:
            plt.ioff()

    def test_prepare(self):
        pass

    def test_script_is_installed(self):
        sp.check_call('SDTimage -h'.split(' '))

    def test_preprocess_single_files(self):
        files = glob.glob(os.path.join(self.obsdir_ra, '*.fits'))

        main_preprocess(files[:2] + ['--debug', '-c', self.config_file])
        for file in files[:2]:
            # I used debug_file_format : png in the config
            if HAS_MPL:
                f = os.path.basename(file).replace('.fits', '_0.png')

                assert os.path.exists(os.path.join(self.prodir_ra, f))

    def test_script_is_installed_prep(self):
        sp.check_call('SDTpreprocess -h'.split(' '))

    def test_preprocess_no_config(self):
        with pytest.raises(ValueError) as excinfo:
            main_preprocess([])
        assert "Please specify the config file!" in str(excinfo.value)

    def test_preprocess_invalid(self):
        with pytest.warns(UserWarning) as record:
            main_preprocess([self.config_file])<commit_msg>Fix test<commit_after># -*- coding: utf-8 -*-


import numpy as np
import hashlib

try:
    import contextlib2 as contextlib
    FileNotFoundError = IOError
except ImportError:
    import contextlib

try:
    import matplotlib.pyplot as plt
    HAS_MPL = True
except ImportError:
    HAS_MPL = False

try:
    import pyregion
    HAS_PYREGION = True
except ImportError:
    HAS_PYREGION = False

from srttools.imager import ScanSet
from srttools.scan import Scan
from srttools.calibration import CalibratorTable
from srttools.calibration import HAS_STATSM
from srttools.read_config import read_config
from srttools.imager import main_imager, main_preprocess, \
    _excluded_regions_from_args
from srttools.simulate import simulate_map
from srttools.global_fit import display_intermediate
from srttools.io import mkdir_p
from srttools.interactive_filter import intervals
import copy
import os
import glob
import astropy.units as u
import shutil
import pytest
from astropy import log
import subprocess as sp
import astropy
from astropy.logger import logging

try:
    from tqdm import tqdm
except ImportError:
    def tqdm(x):
        return x

@pytest.fixture()
def logger():
    logger = logging.getLogger('Some.Logger')
    logger.setLevel(logging.INFO)

    return logger


np.random.seed(1241347)


def _md5(file):
    with open(file, 'rb') as fobj:
        string = fobj.read()

    return hashlib.md5(string).hexdigest()


def _2d_gauss(x, y, sigma=2.5 / 60.):
    """A Gaussian beam"""
    return np.exp(-(x ** 2 + y ** 2) / (2 * sigma**2))


def gauss_src_func(x, y):
    return 25 * _2d_gauss(x, y, sigma=2.5 / 60)


def sim_config_file(filename, add_garbage=False, prefix=None):
    """Create a sample config file, to be modified by hand."""
    string0 = """
[local]
workdir : .
datadir : .
productdir : test_image

[analysis]
projection : ARC
interpolation : spline
prefix : test_
list_of_directories :
    gauss_ra
    gauss_dec
    defective
"""
    string1 = """
calibrator_directories :
    calibration
"""
    string2 = """

skydip_directories :
    gauss_skydip

noise_threshold : 5

pixel_size : 0.8

[debugging]

debug_file_format : png

"""
    if prefix is None:
        prefix = os.getcwd()
    import tempfile
    string = string0
    with open(filename, 'w') as fobj:
        print(string0, file=fobj)
        if add_garbage:
            for _ in range(100):
                garbage = '    ' + \
                    tempfile.NamedTemporaryFile(prefix=prefix).name[1:]
                print(garbage, file=fobj)
                string += garbage + '\n'
        print(string1, file=fobj)
        string += string1
        if add_garbage:
            for _ in range(100):
                garbage = '    ' + \
                    tempfile.NamedTemporaryFile(prefix=prefix).name[1:]
                print(garbage, file=fobj)
                string += garbage + '\n'
        print(string2, file=fobj)
        string += string2
    return string


def sim_map(obsdir_ra, obsdir_dec):
    simulate_map(count_map=gauss_src_func,
                 length_ra=30.,
                 length_dec=30.,
                 outdir=(obsdir_ra, obsdir_dec), mean_ra=180,
                 mean_dec=45, speed=1.5,
                 spacing=0.5, srcname='Dummy', channel_ratio=0.8,
                 baseline="flat")


class TestScanSet(object):
    @classmethod
    def setup_class(klass):
        import os

        klass.curdir = os.path.dirname(__file__)
        klass.datadir = os.path.join(klass.curdir, 'data')
        klass.sim_dir = os.path.join(klass.datadir, 'sim')
        klass.prodir_ra = os.path.join(klass.datadir,
                                       'sim', 'test_image', 'gauss_ra')
        klass.prodir_dec = os.path.join(klass.datadir,
                                        'sim', 'test_image', 'gauss_dec')

        klass.obsdir_ra = os.path.join(klass.datadir, 'sim', 'gauss_ra')
        klass.obsdir_dec = os.path.join(klass.datadir, 'sim', 'gauss_dec')
        klass.config_file = \
            os.path.abspath(os.path.join(klass.sim_dir, 'test_config_sim.ini'))
        klass.caldir = os.path.join(klass.datadir, 'sim', 'calibration')
        klass.simulated_flux = 0.25
        # First off, simulate a beamed observation  -------

        sim_config_file(klass.config_file, add_garbage=True,
                        prefix="./")

        if (not os.path.exists(klass.obsdir_ra)) or \
                (not os.path.exists(klass.obsdir_dec)):
            mkdir_p(klass.obsdir_ra)
            mkdir_p(klass.obsdir_dec)
            log.info('Fake map: Point-like (but Gaussian beam shape), '
                  '{} Jy.'.format(klass.simulated_flux))
            sim_map(klass.obsdir_ra, klass.obsdir_dec)

        defective_dir = os.path.join(klass.sim_dir, 'defective')
        if not os.path.exists(defective_dir):
            shutil.copytree(os.path.join(klass.datadir, 'calibrators'),
                            defective_dir)

        # Copy skydip scan
        skydip_dir = os.path.join(klass.datadir, 'gauss_skydip')
        new_skydip_dir = os.path.join(klass.sim_dir, 'gauss_skydip')
        if os.path.exists(skydip_dir) and not os.path.exists(new_skydip_dir):
            shutil.copytree(skydip_dir, new_skydip_dir)
        caltable = CalibratorTable()
        caltable.from_scans(glob.glob(os.path.join(klass.caldir,
                                                   '*.fits')), debug=True)

        caltable.update()
        klass.calfile = os.path.join(klass.datadir, 'calibrators.hdf5')
        caltable.write(klass.calfile, overwrite=True)

        klass.config = read_config(klass.config_file)
        klass.raonly = os.path.abspath(os.path.join(klass.datadir,
                                                    'test_raonly.ini'))
        klass.deconly = os.path.abspath(os.path.join(klass.datadir,
                                                     'test_deconly.ini'))

        if HAS_PYREGION:
            excluded_xy, excluded_radec = \
                _excluded_regions_from_args([os.path.join(klass.datadir,
                                                          "center.reg")])
        else:
            excluded_xy, excluded_radec = None, None

        klass.scanset = ScanSet(klass.config_file, nosub=False,
                                norefilt=False, plot=True,
                                debug=True, avoid_regions=excluded_radec)
        klass.scanset.write('test.hdf5', overwrite=True)

        klass.stdinfo = {}
        klass.stdinfo['FLAG'] = None
        klass.stdinfo['zap'] = intervals()
        klass.stdinfo['base'] = intervals()
        klass.stdinfo['fitpars'] = np.array([0, 0])

        def scan_no(scan_str):
            basename = os.path.splitext(os.path.basename(scan_str))[0]
            return int(basename.replace('Dec', '').replace('Ra', ''))

        klass.dec_scans = \
            dict([(scan_no(s), s)
                 for s in klass.scanset.scan_list if 'Dec' in s])
        klass.ra_scans = \
            dict([(scan_no(s), s)
                 for s in klass.scanset.scan_list if 'Ra' in s])
        klass.n_ra_scans = max(list(klass.ra_scans.keys()))
        klass.n_dec_scans = max(list(klass.dec_scans.keys()))

        if HAS_MPL:
            plt.ioff()

    def test_prepare(self):
        pass

    def test_script_is_installed(self):
        sp.check_call('SDTimage -h'.split(' '))

    def test_preprocess_single_files(self):
        files = glob.glob(os.path.join(self.obsdir_ra, '*.fits'))

        main_preprocess(files[:2] + ['--debug', '-c', self.config_file,
                                     '--plot'])
        for file in files[:2]:
            # I used debug_file_format : png in the config
            if HAS_MPL:
                f = os.path.basename(file).replace('.fits', '_0.png')

                assert os.path.exists(os.path.join(self.prodir_ra, f))

    def test_script_is_installed_prep(self):
        sp.check_call('SDTpreprocess -h'.split(' '))

    def test_preprocess_no_config(self):
        with pytest.raises(ValueError) as excinfo:
            main_preprocess([])
        assert "Please specify the config file!" in str(excinfo.value)

    def test_preprocess_invalid(self):
        with pytest.warns(UserWarning) as record:
            main_preprocess([self.config_file])
{'fraction_non_alphanumeric': 0.1123792270531401, 'fraction_numerical': 0.008816425120772946, 'mean_word_length': 2.1665391969407266, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 21, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1529415', 'n_tokens_mistral': 5498, 'n_tokens_neox': 5406, 'n_words': 1031}