python类MaskedArray()的实例源码

def _fix_output(output, usemask=True, asrecarray=False):
    """
    Private function: return a recarray, a ndarray, a MaskedArray
    or a MaskedRecords depending on the input parameters
    """
    if not isinstance(output, MaskedArray):
        usemask = False
    if usemask:
        if asrecarray:
            output = output.view(MaskedRecords)
    else:
        output = ma.filled(output)
        if asrecarray:
            output = output.view(recarray)
    return output

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def _fix_output(output, usemask=True, asrecarray=False):
    """
    Private function: return a recarray, a ndarray, a MaskedArray
    or a MaskedRecords depending on the input parameters
    """
    if not isinstance(output, MaskedArray):
        usemask = False
    if usemask:
        if asrecarray:
            output = output.view(MaskedRecords)
    else:
        output = ma.filled(output)
        if asrecarray:
            output = output.view(recarray)
    return output

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def _fix_output(output, usemask=True, asrecarray=False):
    """
    Private function: return a recarray, a ndarray, a MaskedArray
    or a MaskedRecords depending on the input parameters
    """
    if not isinstance(output, MaskedArray):
        usemask = False
    if usemask:
        if asrecarray:
            output = output.view(MaskedRecords)
    else:
        output = ma.filled(output)
        if asrecarray:
            output = output.view(recarray)
    return output

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def get_gaussian_seeing_weighted_spec(self, x_c, y_c, radius, seeing=4):
        """
        Function to extract the spectrum of a circular aperture defined by x_c, y_c and radius in spaxel space.
        The spectrum is weighted by a 2d gaussian centered at the center of the aperture, with a std = seeing in spaxels
        :param x_c: x coordinate of the center of the aperture (spaxel)
        :param y_c: y coordiante of the center of the aperture (spaxel)
        :param radius: radius of the circular aperture
        :param seeing: standard deviation of the gaussian in spaxels
        :return: XSpectrum1D object
        """
        import scipy.ndimage.filters as fi
        new_3dmask = self.get_mini_cube_mask_from_ellipse_params(x_c, y_c, radius)
        w = self.wavelength
        n = len(w)
        fl = np.zeros(n)
        sig = np.zeros(n)
        self.cube.mask = new_3dmask
        for wv_ii in range(n):
            mask = new_3dmask[wv_ii]
            center = np.zeros(mask.shape)  ###Por alguna razon no funciona si cambio la asignacion a np.zeros_like(mask)
            center[y_c][x_c] = 1
            weigths = ma.MaskedArray(fi.gaussian_filter(center, seeing))
            weigths.mask = mask
            weigths = weigths / np.sum(weigths)
            fl[wv_ii] = np.sum(self.cube[wv_ii] * weigths)
            sig[wv_ii] = np.sqrt(np.sum(self.stat[wv_ii] * (weigths ** 2)))
        self.cube.mask = self.mask_init
        return XSpectrum1D.from_tuple((w, fl, sig))

def _fix_output(output, usemask=True, asrecarray=False):
    """
    Private function: return a recarray, a ndarray, a MaskedArray
    or a MaskedRecords depending on the input parameters
    """
    if not isinstance(output, MaskedArray):
        usemask = False
    if usemask:
        if asrecarray:
            output = output.view(MaskedRecords)
    else:
        output = ma.filled(output)
        if asrecarray:
            output = output.view(recarray)
    return output

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def _fix_output(output, usemask=True, asrecarray=False):
    """
    Private function: return a recarray, a ndarray, a MaskedArray
    or a MaskedRecords depending on the input parameters
    """
    if not isinstance(output, MaskedArray):
        usemask = False
    if usemask:
        if asrecarray:
            output = output.view(MaskedRecords)
    else:
        output = ma.filled(output)
        if asrecarray:
            output = output.view(recarray)
    return output

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def set_smooth_metdata():
    @wrapt.decorator
    def func_wrapper(wrapped, instance, args, kwargs): 
        do_meta = from_args(wrapped, ("meta",), *args, **kwargs)["meta"]

        if do_meta is None:
            do_meta = True

        if not xarray_enabled() or not do_meta:
            return wrapped(*args, **kwargs)

        argvars = from_args(wrapped, ("field", "passes"), 
                            *args, **kwargs)

        field = argvars["field"]
        passes = argvars["passes"]

        result = wrapped(*args, **kwargs)

        outname = "smooth2d"
        outdimnames = ["dim_{}".format(i) for i in py3range(result.ndim)]
        outcoords = OrderedDict()
        outattrs = OrderedDict()

        if isinstance(field, DataArray):
            outname = "smooth_" + ucode(field.name)
            outdimnames = list(field.dims)
            outcoords.update(field.coords)
            outattrs.update(field.attrs)
            outattrs["passes"] = passes

        if isinstance(result, ma.MaskedArray):
            outattrs["_FillValue"] = result.fill_value
            outattrs["missing_value"] = result.fill_value

        return DataArray(result, name=outname, coords=outcoords, 
                       dims=outdimnames, attrs=outattrs)

    return func_wrapper

def _fix_output(output, usemask=True, asrecarray=False):
    """
    Private function: return a recarray, a ndarray, a MaskedArray
    or a MaskedRecords depending on the input parameters
    """
    if not isinstance(output, MaskedArray):
        usemask = False
    if usemask:
        if asrecarray:
            output = output.view(MaskedRecords)
    else:
        output = ma.filled(output)
        if asrecarray:
            output = output.view(recarray)
    return output

def test_view_simple_dtype(self):
        (mrec, a, b, arr) = self.data
        ntype = (np.float, 2)
        test = mrec.view(ntype)
        self.assertTrue(isinstance(test, ma.MaskedArray))
        assert_equal(test, np.array(list(zip(a, b)), dtype=np.float))
        self.assertTrue(test[3, 1] is ma.masked)

def test_null_types(self):
        """
        Test to validate that the numpy protocol handler can deal with null values.
        @since 3.3.0
         - updated 3.6.0: now numeric types used masked array
        @jira_ticket PYTHON-550
        @expected_result Numpy can handle non mapped types' null values.

        @test_category data_types:serialization
        """
        s = self.session
        s.row_factory = tuple_factory
        s.client_protocol_handler = NumpyProtocolHandler

        table = "%s.%s" % (self.keyspace_name, self.function_table_name)
        create_table_with_all_types(table, s, 10)

        begin_unset = max(s.execute('select primkey from %s' % (table,))[0]['primkey']) + 1
        keys_null = range(begin_unset, begin_unset + 10)

        # scatter some emptry rows in here
        insert = "insert into %s (primkey) values (%%s)" % (table,)
        execute_concurrent_with_args(s, insert, ((k,) for k in keys_null))

        result = s.execute("select * from %s" % (table,))[0]

        from numpy.ma import masked, MaskedArray
        result_keys = result.pop('primkey')
        mapped_index = [v[1] for v in sorted(zip(result_keys, count()))]

        had_masked = had_none = False
        for col_array in result.values():
            # these have to be different branches (as opposed to comparing against an 'unset value')
            # because None and `masked` have different identity and equals semantics
            if isinstance(col_array, MaskedArray):
                had_masked = True
                [self.assertIsNot(col_array[i], masked) for i in mapped_index[:begin_unset]]
                [self.assertIs(col_array[i], masked) for i in mapped_index[begin_unset:]]
            else:
                had_none = True
                [self.assertIsNotNone(col_array[i]) for i in mapped_index[:begin_unset]]
                [self.assertIsNone(col_array[i]) for i in mapped_index[begin_unset:]]
        self.assertTrue(had_masked)
        self.assertTrue(had_none)

def load_data(self):
        hdulist = fits.open(self.filename)
        print("MuseCube: Loading the cube fluxes and variances...")

        # import pdb; pdb.set_trace()
        self.cube = ma.MaskedArray(hdulist[1].data)
        self.stat = ma.MaskedArray(hdulist[2].data)

        print("MuseCube: Defining master masks (this may take a while but it is for the greater good).")
        # masking
        self.mask_init = np.isnan(self.cube) | np.isnan(self.stat)
        self.cube.mask = self.mask_init
        self.stat.mask = self.mask_init

        # for ivar weighting ; consider creating it in init ; takes long
        # self.flux_over_ivar = self.cube / self.stat

        self.header_1 = hdulist[1].header  # Necesito el header para crear una buena copia del white.
        self.header_0 = hdulist[0].header

        if self.filename_white is None:
            print("MuseCube: No white image given, creating one.")

            w_data = copy.deepcopy(self.create_white(save=False).data)

            w_header_0 = copy.deepcopy(self.header_0)
            w_header_1 = copy.deepcopy(self.header_1)

            # These loops remove the third dimension from the header's keywords. This is neccesary in order to
            # create the white image and preserve the cube astrometry
            for i in w_header_0.keys():
                if '3' in i:
                    del w_header_0[i]
            for i in w_header_1.keys():
                if '3' in i:
                    del w_header_1[i]

            # prepare the header
            hdu = fits.HDUList()
            hdu_0 = fits.PrimaryHDU(header=w_header_0)
            hdu_1 = fits.ImageHDU(data=w_data, header=w_header_1)
            hdu.append(hdu_0)
            hdu.append(hdu_1)
            hdu.writeto('new_white.fits', clobber=True)
            self.filename_white = 'new_white.fits'
            print("MuseCube: `new_white.fits` image saved to disk.")

def spatial_smooth(self, npix, output="smoothed.fits", test=False, **kwargs):
        """Applies Gaussian filter of std=npix in both spatial directions
        and writes it to disk as a new MUSE Cube.
        Notes: the STAT cube is not touched.

        Parameters
        ----------
        npix : int
            Std of Gaussian kernel in spaxel units.
        output : str, optional
            Name of the output file
        test : bool, optional
            Whether to check for flux being conserved

        **kwargs are passed down to scipy.ndimage.gaussian_filter()

        Return
        ------
        Writes a new file to disk.

        """
        if not isinstance(npix, int):
            raise ValueError("npix must be integer.")

        cube_new = copy.deepcopy(self.cube)
        ntot = len(self.cube)
        for wv_ii in range(ntot):
            print('{}/{}'.format(wv_ii + 1, ntot))
            image_aux = self.cube[wv_ii, :, :]
            smooth_ii = ma.MaskedArray(ndimage.gaussian_filter(image_aux, sigma=npix, **kwargs))
            smooth_ii.mask = image_aux.mask | np.isnan(smooth_ii)

            # test the fluxes are conserved
            if test:
                gd_pix = ~smooth_ii.mask
                try:
                    med_1 = np.nansum(smooth_ii[gd_pix])
                    med_2 = np.nansum(image_aux[gd_pix])
                    print(med_1, med_2, (med_1 - med_2) / med_1)
                    np.testing.assert_allclose(med_1, med_2, decimal=4)
                except AssertionError:
                    import pdb
                    pdb.set_trace()
            cube_new[wv_ii, :, :] = smooth_ii
            # import pdb; pdb.set_trace()

        hdulist = fits.open(self.filename)
        hdulist[1].data = cube_new.data
        prihdr = hdulist[0].header
        comment = 'Spatially smoothed with a Gaussian kernel of sigma={} spaxels (by MuseCube)'.format(npix)
        print(comment)
        prihdr['history'] = comment
        hdulist.writeto(output, clobber=True)
        print("MuseCube: new smoothed cube written to {}".format(output))

def spatial_smooth(self, npix, output="smoothed.fits", test=False, **kwargs):
        """Applies Gaussian filter of std=npix in both spatial directions
        and writes it to disk as a new MUSE Cube.
        Notes: the STAT cube is not touched.

        Parameters
        ----------
        npix : int
            Std of Gaussian kernel in spaxel units.
        output : str, optional
            Name of the output file
        test : bool, optional
            Whether to check for flux being conserved

        **kwargs are passed down to scipy.ndimage.gaussian_filter()

        Return
        ------
        Writes a new file to disk.

        """
        if not isinstance(npix, int):
            raise ValueError("npix must be integer.")

        cube_new = ma.MaskedArray(copy.deepcopy(self.cube))
        cube_new.mask = np.isnan(self.cube) | np.isnan(self.stat)
        ntot = len(self.cube)
        for wv_ii in range(ntot):
            print('{}/{}'.format(wv_ii + 1, ntot))
            image_aux = cube_new[wv_ii, :, :]
            smooth_ii = ma.MaskedArray(ndimage.gaussian_filter(image_aux, sigma=npix, **kwargs))
            smooth_ii.mask = image_aux.mask | np.isnan(smooth_ii)

            # test the fluxes are conserved
            if test:
                gd_pix = ~smooth_ii.mask
                try:
                    med_1 = np.nansum(smooth_ii[gd_pix])
                    med_2 = np.nansum(image_aux[gd_pix])
                    print(med_1, med_2, (med_1 - med_2) / med_1)
                    np.testing.assert_allclose(med_1, med_2, decimal=4)
                except AssertionError:
                    import pdb
                    pdb.set_trace()
            cube_new[wv_ii, :, :] = smooth_ii
            # import pdb; pdb.set_trace()

        hdulist = fits.open(self.filename)
        hdulist[1].data = cube_new.data
        prihdr = hdulist[0].header
        comment = 'Spatially smoothed with a Gaussian kernel of sigma={} spaxels (by MuseCube)'.format(npix)
        print(comment)
        prihdr['history'] = comment
        hdulist.writeto(output, clobber=True)
        print("MuseCube: new smoothed cube written to {}".format(output))

def test_null_types(self):
        """
        Test to validate that the numpy protocol handler can deal with null values.
        @since 3.3.0
         - updated 3.6.0: now numeric types used masked array
        @jira_ticket PYTHON-550
        @expected_result Numpy can handle non mapped types' null values.

        @test_category data_types:serialization
        """
        s = self.session
        s.client_protocol_handler = NumpyProtocolHandler

        table = "%s.%s" % (self.keyspace_name, self.function_table_name)
        create_table_with_all_types(table, s, 10)

        begin_unset = max(s.execute('select primkey from %s' % (table,))[0]['primkey']) + 1
        keys_null = range(begin_unset, begin_unset + 10)

        # scatter some emptry rows in here
        insert = "insert into %s (primkey) values (%%s)" % (table,)
        execute_concurrent_with_args(s, insert, ((k,) for k in keys_null))

        result = s.execute("select * from %s" % (table,))[0]

        from numpy.ma import masked, MaskedArray
        result_keys = result.pop('primkey')
        mapped_index = [v[1] for v in sorted(zip(result_keys, count()))]

        had_masked = had_none = False
        for col_array in result.values():
            # these have to be different branches (as opposed to comparing against an 'unset value')
            # because None and `masked` have different identity and equals semantics
            if isinstance(col_array, MaskedArray):
                had_masked = True
                [self.assertIsNot(col_array[i], masked) for i in mapped_index[:begin_unset]]
                [self.assertIs(col_array[i], masked) for i in mapped_index[begin_unset:]]
            else:
                had_none = True
                [self.assertIsNotNone(col_array[i]) for i in mapped_index[:begin_unset]]
                [self.assertIsNone(col_array[i]) for i in mapped_index[begin_unset:]]
        self.assertTrue(had_masked)
        self.assertTrue(had_none)