Source code for sunkit_spex.legacy.io

from collections import OrderedDict

import numpy as np
import scipy.io
import xarray

import astropy.units as u
from astropy.table import Table

from sunpy.data import manager
from sunpy.io.special.genx import read_genx

__all__ = ["load_chianti_continuum", "load_chianti_lines_lite", "load_xray_abundances", "read_abundance_genx"]


[docs] @manager.require( "chianti_lines", [ "https://soho.nascom.nasa.gov/solarsoft/packages/xray/dbase/chianti/chianti_lines_1_10_v71.sav", "https://lmsal.com/solarsoft/ssw/packages/xray/dbase/chianti/chianti_lines_1_10_v71.sav", ], "2046d818efec207a83e9c5cc6ba4a5fa8574bf8c2bd8a6bb9801e4b8a2a0c677", ) def load_chianti_lines_lite(): """ Read X-ray emission line info from an IDL sav file produced by CHIANTI. This function does not read all data in the file, but only that required to calculate the observed X-ray spectrum. Returns ------- line_intensities_at_source: `xarray.DataArray` Intensities of each of each line as a function of temperature and associated metadata and coordinates. Notes ----- CHIANTI File By default, this function uses the file located at https://hesperia.gsfc.nasa.gov/ssw/packages/xray/dbase/chianti/chianti_lines_1_10_v71.sav. To use a different file (created by CHIANTI and saved as a sav file) call this function in the following way: >>> from sunpy.data import manager # doctest: +SKIP >>> with manager.override_file("chianti_lines", uri=filename): # doctest: +SKIP ... line_info = load_chianti_lines_light() # doctest: +SKIP where filename is the location of the file to be read. Intensity Units The line intensities read from the CHIANTI file are in units of ph / cm**2 / s / sr. Therefore they are specific intensities, i.e. per steradian, or solid angle. Here, let us call these intensities, intensity_per_solid_angle. The solid angle is given by flare_area / observer_distance**2. Total integrated intensity can be rewritten in terms of volume EM and solid angle: intensity = intensity_per_solid_angle_per_volEM * volEM * solid_angle intensity = intensity_per_solid_angle / (colEM * flare_area) * (flare_area / observer_dist**2) * volEM intensity = intensity_per_solid_angle / colEM / observer_dist**2 * volEM i.e. flare area cancels. Therefore: intensity = intensity_per_solid_angle / colEM / observer_dist**2 * volEM, or, dividing both sides by volEM, intensity_per_EM = intensity_per_solid_angle / colEM / observer_dist**2 In this function, we normalize the intensity by colEM and scale it to the source, i.e. intensity_out = intensity_per_solid_angle / colEM * 4 * pi Therefore the intensity values output by this function must be multiplied by EM and divided by 4 pi observer_dist**2 to get physical values at the observer. """ # Read linefile linefile = manager.get("chianti_lines") contents = scipy.io.readsav(linefile) out = contents["out"] # Define units wvl_units = _clean_units(out["WVL_UNITS"]) int_units = _clean_units(out["INT_UNITS"]) energy_unit = u.keV # Extract line info and convert from wavelength to energy. line_intensities = [] line_elements = [] line_peak_energies = [] for j, lines in enumerate(out["lines"]): # Extract line element index and peak energy. line_elements.append(lines["IZ"] + 1) # TODO: Confirm lines["IZ"] is indeed atomic number - 1 line_peak_energies.append(u.Quantity(lines["WVL"], unit=wvl_units).to(energy_unit, equivalencies=u.spectral())) # Sort line info in ascending energy. ordd = np.argsort(np.array(line_peak_energies[j])) line_elements[j] = line_elements[j][ordd] line_peak_energies[j] = line_peak_energies[j][ordd] # Extract line intensities. line_intensities.append(_extract_line_intensities(lines["INT"][ordd]) * int_units) # If there is only one element in the line properties, unpack values. if len(out["lines"]) == 1: line_elements = line_elements[0] line_peak_energies = line_peak_energies[0] line_intensities = line_intensities[0] # Normalize line intensities by EM and integrate over whole sky to get intensity at source. # This means that physical intensities can be calculated by dividing by # 4 * pi * R**2 where R is the observer distance. line_colEMs = 10.0 ** _clean_array_dims(out["LOGEM_ISOTHERMAL"]) / u.cm**5 line_intensities /= line_colEMs line_intensities *= 4 * np.pi * u.sr # Put data into intuitive structure and return it. return xarray.DataArray( line_intensities.value, dims=["lines", "temperature"], coords={ "logT": ("temperature", _clean_array_dims(out["LOGT_ISOTHERMAL"])), "peak_energy": ("lines", line_peak_energies), "atomic_number": ("lines", line_elements), }, attrs={ "units": {"data": line_intensities.unit, "peak_energy": line_peak_energies.unit}, "file": linefile, "element_index": contents["zindex"], "chianti_doc": _clean_chianti_doc(contents["chianti_doc"]), }, )
[docs] @manager.require( "chianti_continuum", [ "https://soho.nascom.nasa.gov/solarsoft/packages/xray/dbase/chianti/chianti_cont_1_250_v71.sav", "https://lmsal.com/solarsoft/ssw/packages/xray/dbase/chianti/chianti_cont_1_250_v71.sav", ], "aadf4355931b4c241ac2cd5669e89928615dc1b55c9fce49a155b70915a454dd", ) def load_chianti_continuum(): """ Read X-ray continuum emission info from an IDL sav file produced by CHIANTI Returns ------- continuum_intensities: `xarray.DataArray` Continuum intensity as a function of element, temperature and energy/wavelength and associated metadata and coordinates. Notes ----- By default, this function uses the file located at https://hesperia.gsfc.nasa.gov/ssw/packages/xray/dbase/chianti/chianti_cont_1_250_v71.sav To use a different file call this function in the following way: >>> from sunpy.data import manager # doctest: +SKIP >>> with manager.override_file("chianti_continuum", uri=filename): # doctest: +SKIP ... line_info = load_chianti_lines_light() # doctest: +SKIP where filename is the location of the file to be read. """ # Define units intensity_unit = u.ph * u.cm**3 / (u.s * u.keV * u.sr) temperature_unit = u.K wave_unit = u.AA # Read file contfile = manager.get("chianti_continuum") contents = scipy.io.readsav(contfile) # Concatenate low and high wavelength intensity arrays. intensities = np.concatenate((contents["totcont_lo"], contents["totcont"]), axis=-1) # Integrate over sphere surface of radius equal to observer distance # to get intensity at source. This means that physical intensities can # be calculated by dividing by 4 * pi * R**2 where R is the observer distance. intensities *= 4 * np.pi intensity_unit *= u.sr # Put file data into intuitive structure and return data. return xarray.DataArray( intensities, dims=["element_index", "temperature", "wavelength"], coords={ "element_index": contents["zindex"], "temperature": contents["ctemp"], "wavelength": _clean_array_dims(contents["edge_str"]["WVL"]), }, attrs={ "units": {"data": intensity_unit, "temperature": temperature_unit, "wavelength": wave_unit}, "file": contfile, "wavelength_edges": _clean_array_dims(contents["edge_str"]["WVLEDGE"]) * wave_unit, "chianti_doc": _clean_chianti_doc(contents["chianti_doc"]), }, )
[docs] @manager.require( "xray_abundances", [ "https://soho.nascom.nasa.gov/solarsoft/packages/xray/dbase/chianti/xray_abun_file.genx", "https://lmsal.com/solarsoft/ssw/packages/xray/dbase/chianti/xray_abun_file.genx", ], "92c0e1f9a83da393cc38840752fda5a5b44c5b18a4946e5bf12c208771fe0fd3", ) def load_xray_abundances(abundance_type=None): """ Returns the abundances written in the xray_abun_file.genx The abundances are taken from CHIANTI and MEWE. The source filenames are: cosmic sun_coronal sun_coronal_ext sun_hybrid sun_hybrid_ext sun_photospheric mewe_cosmic mewe_solar The first six come from Chianti, the last two from Mewe. They are: cosmic sun_coronal sun_coronal_ext sun_hybrid sun_hybrid_ext sun_photospheric mewe_cosmic mewe_solar These abundances are used with CHIANTI_KEV. MEWE_KEV can only use the two mewe sourced abundance distributions unless using a heavily modified rel_abun structure for all of the elements. Parameters ---------- abundance_type: `str` Type of abundance to be read from file. Option are (From Chianti) 1. cosmic 2. sun_coronal - default abundance 3. sun_coronal_ext 4. sun_hybrid 5. sun_hybrid_ext 6. sun_photospheric 7. mewe_cosmic 8. mewe_solar - default for mewe_kev Returns ------- out: Array of 50 abundance levels for first 50 elements. """ # If kwargs not set, set defaults if abundance_type is None: abundance_type = "sun_coronal" xray_abundance_file = manager.get("xray_abundances") # Read file contents = read_abundance_genx(xray_abundance_file) # Restructure data into an easier form. try: contents.pop("header") except KeyError: pass n_elements = len(contents[list(contents.keys())[0]]) columns = [np.arange(1, n_elements + 1), *list(contents.values())] names = ["atomic number", *list(contents.keys())] return Table(columns, names=names)
[docs] def read_abundance_genx(filename): # Read file. contents = read_genx(filename) # Combine data and keys from each entry in file. output = OrderedDict() for arr in contents["SAVEGEN0"]: output[arr["FILNAM"]] = arr["ABUND"] # Add header data output["header"] = contents["HEADER"] output["header"]["CHIANTI VERSION"] = float(contents["SAVEGEN1"][:3]) return output
def _extract_line_intensities(lines_int_sorted): line_ints = np.empty((lines_int_sorted.shape[0], lines_int_sorted[0].shape[0]), dtype=float) for i in range(line_ints.shape[0]): line_ints[i, :] = lines_int_sorted[i] return line_ints def _clean_array_dims(arr, dtype=None): # Initialize a single array to hold contents of input arr. result = np.empty(list(arr.shape) + list(arr[0].shape)) # Combine arrays in arr into single array. for i in range(arr.shape[0]): result[i] = arr[i] # Remove redundant dimensions result = np.squeeze(result) # If result is now unsized, convert to scalar. if result.shape == (): result = result.item() if dtype is not None: dtype(result) return result def _clean_string_dims(arr): result = [str(s, "utf-8") for s in arr] if len(result) == 1: result = result[0] return result def _combine_strings(arr): result = [".".join([str(ss, "utf-8") for ss in s]) for s in arr] if len(result) == 1: result = result[0] return result def _clean_units(arr): result = [] for a in arr: unit = str(a, "utf-8") unit_components = unit.split() for i, component in enumerate(unit_components): # Remove plurals if component in ["photons", "Angstroms"]: component = component[:-1] # Insert ** for indices. component_minus_split = component.split("-") if len(component_minus_split) > 1: "**-".join(component_minus_split) component_plus_split = component.split("+") if len(component_plus_split) > 1: "**-".join(component_plus_split) unit_components[i] = component result.append("*".join(unit_components)) if len(result) == 1: result = result[0] return u.Unit(result) def _clean_chianti_doc(arr): chianti_doc = {} chianti_doc["ion_file"] = str(arr[0][0], "utf-8") chianti_doc["ion_ref"] = "{}.{}.{}".format( str(arr["ion_ref"][0][0], "utf-8"), str(arr["ion_ref"][0][1], "utf-8"), str(arr["ion_ref"][0][2], "utf-8") ) chianti_doc["version"] = str(arr[0][2], "utf-8") return chianti_doc