Skip to content

Sky parser

C pynasonde.digisonde.parsers.sky — SkyExtractor

Parses sky images and associated metadata.

pynasonde.digisonde.parsers.sky

Skymap/sky-format file parser utilities for Digisonde.

This module contains :class:SkyExtractor, a compact parser for skymap/sky-format outputs (both skymap and velocity-style datasets). The extractor reads simple text files, parses nested fixed-format blocks and exposes helpers to convert parsed content into pandas DataFrames suitable for plotting with :class:SkySummaryPlots.

SkyExtractor

Bases: object

Parser for SKY-format files.

The :class:SkyExtractor reads text files produced by Digisonde's skymap/sky output. It parses nested blocks (data header, frequency headers, source sky-data) into a lightweight dictionary (:attr:sky_struct) and provides :meth:to_pandas to convert the parsed records into a flattened pandas.DataFrame.

Source code in pynasonde/digisonde/parsers/sky.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
class SkyExtractor(object):
    """Parser for SKY-format files.

    The :class:`SkyExtractor` reads text files produced by Digisonde's
    skymap/sky output. It parses nested blocks (data header, frequency
    headers, source sky-data) into a lightweight dictionary
    (:attr:`sky_struct`) and provides :meth:`to_pandas` to convert the
    parsed records into a flattened pandas.DataFrame.
    """

    def __init__(
        self,
        filename: str,
        extract_time_from_name: bool = False,
        extract_stn_from_name: bool = False,
        n_fft: int = 2048,
        delta_freq: float = 50,  # in Hz
    ) -> None:
        """Create a SkyExtractor.

        Parameters:
            filename: str
                Path to the SKY-format file to parse.
            extract_time_from_name: bool, optional
                If True, attempt to parse a timestamp token from the
                filename (default False).
            extract_stn_from_name: bool, optional
                If True, attempt to determine station code and local
                timezone information (default False).
            n_fft: int, optional
                FFT length used to compute Doppler frequency scaling.
            delta_freq: float, optional
                Frequency step in Hz used by :meth:`get_doppler_freq`.
        """
        # Initialize the data structure to hold extracted data
        self.filename = filename
        self.n_fft = n_fft
        self.delta_freq = delta_freq
        self.l0 = n_fft // 2
        if extract_time_from_name:
            date = self.filename.split("_")[-1].replace(".SKY", "").replace(".sky", "")
            self.date = dt.datetime(int(date[:4]), 1, 1) + dt.timedelta(
                int(date[4:7]) - 1
            )
            self.date = self.date.replace(
                hour=int(date[7:9]), minute=int(date[9:11]), second=int(date[11:13])
            )
            logger.info(f"Date: {self.date}")
        if extract_stn_from_name:
            self.stn_code = self.filename.split("/")[-1].split("_")[0]
            self.stn_info = get_digisonde_info(self.stn_code)
            self.local_timezone_converter = TimeZoneConversion(
                lat=self.stn_info["LAT"], long=self.stn_info["LONG"]
            )
            self.local_time = self.local_timezone_converter.utc_to_local_time(
                [self.date]
            )[0]
            logger.info(f"Station code: {self.stn_code}; {self.stn_info}")
        return

    def read_file(self) -> List[str]:
        """Read the input file and return a list of lines.

        Returns:
            List of lines including their trailing newline characters.
        """
        with open(self.filename, "r") as f:
            return f.readlines()

    def parse_line(self, sky_arch_list: List[str], _i: int):
        """Parse a single line from the raw lines list.

        Parameters:
            sky_arch_list:  List of file lines as returned by :meth:`read_file`.
            _i: Index of the line to parse.

        Returns:
            (indent_level, token_list) where ``indent_level`` is the
                number of leading spaces and ``token_list`` is the
                whitespace-split tokens from the line with 'D' markers
                removed when present.
        """
        sky_arch = sky_arch_list[_i]
        if "D" in sky_arch:
            sky_arch = sky_arch.replace("D", "")
        line_indent = get_indent(sky_arch)
        sky_arch = sky_arch.strip()
        sky_arch = list(filter(None, sky_arch.split()))
        return line_indent, sky_arch

    def parse_data_header(self, sky_arch):
        """Parse a data-header line into a dictionary of fields.

        Parameters:
            sky_arch: Tokenized line (as returned by :meth:`parse_line`).

        Returns:
            Parsed header fields including type, version, number of
                spectrums and other control fields.
        """
        parsed_data_header = dict(
            # Data Format Type (1 Skymap) (2 Velocity) (3 Quality control)
            type=int(sky_arch[0]),
            # Program version Number
            version=float(sky_arch[1]),
            # 57Z1 Digisonde Preface Header
            digi_preface_no=sky_arch[2],
            # Number of Frequencies/Height Spectrums
            n_spectrums=int(sky_arch[3]),
            # Number of Rows in data.
            n_rows_data=int(sky_arch[4]),
            # Others / not eligible
            others=int(sky_arch[6]),
        )
        return parsed_data_header

    def parse_freq_header(self, sky_arch: List[str]):
        """Parse a frequency/height header line into a dict of fields.

        Parameters:
            sky_arch: Tokenized line (as returned by :meth:`parse_line`).

        Returns:
            Parsed frequency/height header fields including sampling
                frequency, group range, polarization and number of sources.
        """
        parsed_freq_header = dict(
            # Frequcncy number (DGS) / Height spectrum number (DPS)
            frq_height_num=int(sky_arch[0]),
            # Maximum Main reciever array lobe zenith angle
            zenith_angle=float(sky_arch[1]),
            # Sampling Frequency in MHz
            sampl_freq=float(sky_arch[2]),
            # Group Range in KM.
            group_range=float(sky_arch[3]),
            # Gain Amplitude number
            gain_ampl=float(sky_arch[4]),
            # Height Spectrum Amplitude number
            height_spctrum_ampl=float(sky_arch[5]),
            # Maximum Height Spectrum Amplitude number
            max_height_spctrum_ampl=float(sky_arch[6]),
            # Number of sources, Maximum set for each line is for 26 sources
            # more sources are wrapped around below the first 26 as shown for
            # height spectrum No.2 below.
            n_sources=int(sky_arch[7]),
            # Height Spectrum cleaning threshold
            height_spctrum_cl_th=float(sky_arch[8]),
            # Spectral line cleaning threshold
            spect_line_cl_th=float(sky_arch[9]),
            # Polarization Identifier 0/1=O/X-mode
            polarization=int(sky_arch[10]),
            # Skymap dataset
            sky_data=None,
        )
        return parsed_freq_header

    def extract(self) -> SimpleNamespace:
        """Parse the SKY file into :attr:`sky_struct`.

        The parser walks the file line-by-line, identifies data headers
        (indentation 1 or 2), then parses frequency headers and nested
        sky-data blocks for each detected source. The final structure is
        wrapped into :attr:`sky` (a namespace) and returned.

        Returns:
            Namespace wrapper around the parsed ``sky_struct`` dictionary.
        """
        self.sky_struct = dict(dataset=[])
        # Read file lines and set the datastructure by indent
        sky_arch_list = self.read_file()
        _i = 0
        while _i < len(sky_arch_list):
            line_indent, sky_arch = self.parse_line(sky_arch_list, _i)
            # print(">", _i, line_indent, sky_arch)
            if line_indent in [1, 2]:  # Consider it first 'Data Header'
                ds = dict(
                    data_header=self.parse_data_header(sky_arch),
                    freq_headers=[],
                )
                _i_nest, n_spectrums = 0, ds["data_header"]["n_spectrums"]
                # print("n_spectrums", ds["data_header"]["n_spectrums"])
                while _i_nest < n_spectrums:
                    # add next line to above to this nest (previous while loop)
                    _i += 1  # you need to add this previous to work on sky_arch
                    line_indent, sky_arch = self.parse_line(sky_arch_list, _i)
                    # print(">>", _i, line_indent, sky_arch)
                    if line_indent == 4:  # Frequency header
                        fh = self.parse_freq_header(sky_arch)
                        # Nesting 2nd layer for datasets from sources 'n_sources'
                        n_sources = fh["n_sources"]
                        # print("n_sources", n_sources)
                        if n_sources > 0:  # Check if data esists
                            fh["sky_data"], _i = self.parse_sky_data(
                                sky_arch_list, _i, n_sources
                            )
                        ds["freq_headers"].append(fh)
                    _i_nest += 1
                self.sky_struct["dataset"].append(ds)
            # Loop add next line
            _i += 1
        self.sky = to_namespace(self.sky_struct)
        return self.sky

    def parse_sky_data(self, sky_arch_list, _i, n_sources):
        """Parse a nested sky-data block belonging to a frequency header.

        The sky-data block encodes coordinates and spectral arrays across
        multiple following lines. Depending on ``n_sources`` the block may
        wrap to multiple groups of five lines.

        Parameters:
            sky_arch_list: list[str]
                Full list of file lines.
            _i: int
                Current index in the file (frequency header line index).
            n_sources: int
                Number of sky sources to extract; determines how many lines
                to consume.

        Returns:
            (data_dict, new_index) where ``data_dict`` contains keys
                ``y_coords``, ``x_coords``, ``spect_amp``, ``spect_dop`` and
                ``rms_error`` and ``new_index`` points to the last consumed
                input index.
        """
        # if n_sources <= 26 then they are in 1 line otherwise check on _i+5 lines
        import re

        y_coords = re.findall(r"-?\d+\.\d+", sky_arch_list[_i + 1].strip())
        y_coords = [float(y) for y in y_coords]
        x_coords = re.findall(r"-?\d+\.\d+", sky_arch_list[_i + 2].strip())
        x_coords = [float(x) for x in x_coords]
        spect_amp = re.findall(r"-?\d+(?:\.\d+)?", sky_arch_list[_i + 3].strip())
        spect_amp = [float(a) for a in spect_amp]
        spect_dop = re.findall(r"-?\d+(?:\.\d+)?", sky_arch_list[_i + 4].strip())
        spect_dop = [float(d) for d in spect_dop]
        rms_error = re.findall(r"-?\d+(?:\.\d+)?", sky_arch_list[_i + 5].strip())
        rms_error = [float(r) for r in rms_error]
        data = dict(
            y_coords=y_coords,  # Y coordinate
            x_coords=x_coords,  # X coordinate
            spect_amp=spect_amp,  # Spectral Amplitude number
            spect_dop=spect_dop,  # Spectral Doppler line number
            rms_error=rms_error,  # Least Squares Fit RMS error
        )
        _i = _i + 5 if n_sources <= 26 else _i + (5 * (int(n_sources / 26) + 1))
        return data, _i

    def get_doppler_freq(self, L: float) -> float:
        """Convert a Doppler bin index to Doppler frequency in MHz.

        Parameters:
            L: float
                Doppler bin index (integer or float) returned by the skymap
                processing.

        Returns:
            Doppler frequency in MHz computed as ``L * delta_freq / n_fft``.
        """
        return L * self.delta_freq / self.n_fft

    def to_pandas(self) -> pd.DataFrame:
        """Flatten parsed sky records into a pandas DataFrame.

        Returns:
            One row per sky-source with columns for coordinates,
                amplitudes, Doppler values (and Doppler frequency), RMS error
                and timestamps where available.
        """
        df = pd.DataFrame()
        for ds in self.sky.dataset:
            for fh in ds.freq_headers:
                d = []
                if fh.sky_data:
                    d.extend(
                        [
                            dict(
                                zenith_angle=fh.zenith_angle,
                                sampl_freq=fh.sampl_freq,
                                group_range=fh.group_range,
                                gain_ampl=fh.gain_ampl,
                                height_spctrum_ampl=fh.height_spctrum_ampl,
                                max_height_spctrum_ampl=fh.max_height_spctrum_ampl,
                                n_sources=fh.n_sources,
                                height_spctrum_cl_th=fh.height_spctrum_cl_th,
                                spect_line_cl_th=fh.spect_line_cl_th,
                                polarization=fh.polarization,
                                x_coord=fh.sky_data.x_coords[i],
                                y_coord=fh.sky_data.y_coords[i],
                                spect_amp=fh.sky_data.spect_amp[i],
                                spect_dop=fh.sky_data.spect_dop[i],
                                spect_dop_freq=self.get_doppler_freq(
                                    fh.sky_data.spect_dop[i]
                                ),
                                rms_error=fh.sky_data.rms_error[i],
                                datetime=self.date if hasattr(self, "date") else None,
                                local_datetime=(
                                    self.local_time
                                    if hasattr(self, "local_time")
                                    else None
                                ),
                            )
                            for i in range(len(fh.sky_data.y_coords))
                        ]
                    )
                d = pd.DataFrame.from_records(d)
                df = pd.concat([df, d])
        return df

__init__(filename, extract_time_from_name=False, extract_stn_from_name=False, n_fft=2048, delta_freq=50)

Create a SkyExtractor.

Parameters:

Name Type Description Default
filename str

str Path to the SKY-format file to parse.

 required
extract_time_from_name bool

bool, optional If True, attempt to parse a timestamp token from the filename (default False).

 False
extract_stn_from_name bool

bool, optional If True, attempt to determine station code and local timezone information (default False).

 False
n_fft int

int, optional FFT length used to compute Doppler frequency scaling.

 2048
delta_freq float

float, optional Frequency step in Hz used by :meth:get_doppler_freq.

 50
Source code in pynasonde/digisonde/parsers/sky.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    filename: str,
    extract_time_from_name: bool = False,
    extract_stn_from_name: bool = False,
    n_fft: int = 2048,
    delta_freq: float = 50,  # in Hz
) -> None:
    """Create a SkyExtractor.

    Parameters:
        filename: str
            Path to the SKY-format file to parse.
        extract_time_from_name: bool, optional
            If True, attempt to parse a timestamp token from the
            filename (default False).
        extract_stn_from_name: bool, optional
            If True, attempt to determine station code and local
            timezone information (default False).
        n_fft: int, optional
            FFT length used to compute Doppler frequency scaling.
        delta_freq: float, optional
            Frequency step in Hz used by :meth:`get_doppler_freq`.
    """
    # Initialize the data structure to hold extracted data
    self.filename = filename
    self.n_fft = n_fft
    self.delta_freq = delta_freq
    self.l0 = n_fft // 2
    if extract_time_from_name:
        date = self.filename.split("_")[-1].replace(".SKY", "").replace(".sky", "")
        self.date = dt.datetime(int(date[:4]), 1, 1) + dt.timedelta(
            int(date[4:7]) - 1
        )
        self.date = self.date.replace(
            hour=int(date[7:9]), minute=int(date[9:11]), second=int(date[11:13])
        )
        logger.info(f"Date: {self.date}")
    if extract_stn_from_name:
        self.stn_code = self.filename.split("/")[-1].split("_")[0]
        self.stn_info = get_digisonde_info(self.stn_code)
        self.local_timezone_converter = TimeZoneConversion(
            lat=self.stn_info["LAT"], long=self.stn_info["LONG"]
        )
        self.local_time = self.local_timezone_converter.utc_to_local_time(
            [self.date]
        )[0]
        logger.info(f"Station code: {self.stn_code}; {self.stn_info}")
    return

read_file()

Read the input file and return a list of lines.

Returns:

Type Description
List[str]

List of lines including their trailing newline characters.
Source code in pynasonde/digisonde/parsers/sky.py 
 95
 96
 97
 98
 99
100
101
102
def read_file(self) -> List[str]:
    """Read the input file and return a list of lines.

    Returns:
        List of lines including their trailing newline characters.
    """
    with open(self.filename, "r") as f:
        return f.readlines()

parse_line(sky_arch_list, _i)

Parse a single line from the raw lines list.

Parameters:

Name Type Description Default
sky_arch_list List[str]

List of file lines as returned by :meth:read_file.

 required
_i int

Index of the line to parse.

 required

Returns:

Type Description

(indent_level, token_list) where indent_level is the number of leading spaces and token_list is the whitespace-split tokens from the line with 'D' markers removed when present.
Source code in pynasonde/digisonde/parsers/sky.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def parse_line(self, sky_arch_list: List[str], _i: int):
    """Parse a single line from the raw lines list.

    Parameters:
        sky_arch_list:  List of file lines as returned by :meth:`read_file`.
        _i: Index of the line to parse.

    Returns:
        (indent_level, token_list) where ``indent_level`` is the
            number of leading spaces and ``token_list`` is the
            whitespace-split tokens from the line with 'D' markers
            removed when present.
    """
    sky_arch = sky_arch_list[_i]
    if "D" in sky_arch:
        sky_arch = sky_arch.replace("D", "")
    line_indent = get_indent(sky_arch)
    sky_arch = sky_arch.strip()
    sky_arch = list(filter(None, sky_arch.split()))
    return line_indent, sky_arch

parse_data_header(sky_arch)

Parse a data-header line into a dictionary of fields.

Parameters:

Name Type Description Default
sky_arch

Tokenized line (as returned by :meth:parse_line).

 required

Returns:

Type Description

Parsed header fields including type, version, number of spectrums and other control fields.
Source code in pynasonde/digisonde/parsers/sky.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def parse_data_header(self, sky_arch):
    """Parse a data-header line into a dictionary of fields.

    Parameters:
        sky_arch: Tokenized line (as returned by :meth:`parse_line`).

    Returns:
        Parsed header fields including type, version, number of
            spectrums and other control fields.
    """
    parsed_data_header = dict(
        # Data Format Type (1 Skymap) (2 Velocity) (3 Quality control)
        type=int(sky_arch[0]),
        # Program version Number
        version=float(sky_arch[1]),
        # 57Z1 Digisonde Preface Header
        digi_preface_no=sky_arch[2],
        # Number of Frequencies/Height Spectrums
        n_spectrums=int(sky_arch[3]),
        # Number of Rows in data.
        n_rows_data=int(sky_arch[4]),
        # Others / not eligible
        others=int(sky_arch[6]),
    )
    return parsed_data_header

parse_freq_header(sky_arch)

Parse a frequency/height header line into a dict of fields.

Parameters:

Name Type Description Default
sky_arch List[str]

Tokenized line (as returned by :meth:parse_line).

 required

Returns:

Type Description

Parsed frequency/height header fields including sampling frequency, group range, polarization and number of sources.
Source code in pynasonde/digisonde/parsers/sky.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def parse_freq_header(self, sky_arch: List[str]):
    """Parse a frequency/height header line into a dict of fields.

    Parameters:
        sky_arch: Tokenized line (as returned by :meth:`parse_line`).

    Returns:
        Parsed frequency/height header fields including sampling
            frequency, group range, polarization and number of sources.
    """
    parsed_freq_header = dict(
        # Frequcncy number (DGS) / Height spectrum number (DPS)
        frq_height_num=int(sky_arch[0]),
        # Maximum Main reciever array lobe zenith angle
        zenith_angle=float(sky_arch[1]),
        # Sampling Frequency in MHz
        sampl_freq=float(sky_arch[2]),
        # Group Range in KM.
        group_range=float(sky_arch[3]),
        # Gain Amplitude number
        gain_ampl=float(sky_arch[4]),
        # Height Spectrum Amplitude number
        height_spctrum_ampl=float(sky_arch[5]),
        # Maximum Height Spectrum Amplitude number
        max_height_spctrum_ampl=float(sky_arch[6]),
        # Number of sources, Maximum set for each line is for 26 sources
        # more sources are wrapped around below the first 26 as shown for
        # height spectrum No.2 below.
        n_sources=int(sky_arch[7]),
        # Height Spectrum cleaning threshold
        height_spctrum_cl_th=float(sky_arch[8]),
        # Spectral line cleaning threshold
        spect_line_cl_th=float(sky_arch[9]),
        # Polarization Identifier 0/1=O/X-mode
        polarization=int(sky_arch[10]),
        # Skymap dataset
        sky_data=None,
    )
    return parsed_freq_header

extract()

Parse the SKY file into :attr:sky_struct.

The parser walks the file line-by-line, identifies data headers (indentation 1 or 2), then parses frequency headers and nested sky-data blocks for each detected source. The final structure is wrapped into :attr:sky (a namespace) and returned.

Returns:

Type Description
SimpleNamespace

Namespace wrapper around the parsed sky_struct dictionary.
Source code in pynasonde/digisonde/parsers/sky.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def extract(self) -> SimpleNamespace:
    """Parse the SKY file into :attr:`sky_struct`.

    The parser walks the file line-by-line, identifies data headers
    (indentation 1 or 2), then parses frequency headers and nested
    sky-data blocks for each detected source. The final structure is
    wrapped into :attr:`sky` (a namespace) and returned.

    Returns:
        Namespace wrapper around the parsed ``sky_struct`` dictionary.
    """
    self.sky_struct = dict(dataset=[])
    # Read file lines and set the datastructure by indent
    sky_arch_list = self.read_file()
    _i = 0
    while _i < len(sky_arch_list):
        line_indent, sky_arch = self.parse_line(sky_arch_list, _i)
        # print(">", _i, line_indent, sky_arch)
        if line_indent in [1, 2]:  # Consider it first 'Data Header'
            ds = dict(
                data_header=self.parse_data_header(sky_arch),
                freq_headers=[],
            )
            _i_nest, n_spectrums = 0, ds["data_header"]["n_spectrums"]
            # print("n_spectrums", ds["data_header"]["n_spectrums"])
            while _i_nest < n_spectrums:
                # add next line to above to this nest (previous while loop)
                _i += 1  # you need to add this previous to work on sky_arch
                line_indent, sky_arch = self.parse_line(sky_arch_list, _i)
                # print(">>", _i, line_indent, sky_arch)
                if line_indent == 4:  # Frequency header
                    fh = self.parse_freq_header(sky_arch)
                    # Nesting 2nd layer for datasets from sources 'n_sources'
                    n_sources = fh["n_sources"]
                    # print("n_sources", n_sources)
                    if n_sources > 0:  # Check if data esists
                        fh["sky_data"], _i = self.parse_sky_data(
                            sky_arch_list, _i, n_sources
                        )
                    ds["freq_headers"].append(fh)
                _i_nest += 1
            self.sky_struct["dataset"].append(ds)
        # Loop add next line
        _i += 1
    self.sky = to_namespace(self.sky_struct)
    return self.sky

parse_sky_data(sky_arch_list, _i, n_sources)

Parse a nested sky-data block belonging to a frequency header.

The sky-data block encodes coordinates and spectral arrays across multiple following lines. Depending on n_sources the block may wrap to multiple groups of five lines.

Parameters:

Name Type Description Default
sky_arch_list

list[str] Full list of file lines.

 required
_i

int Current index in the file (frequency header line index).

 required
n_sources

int Number of sky sources to extract; determines how many lines to consume.

 required

Returns:

Type Description

(data_dict, new_index) where data_dict contains keys y_coords, x_coords, spect_amp, spect_dop and rms_error and new_index points to the last consumed input index.
Source code in pynasonde/digisonde/parsers/sky.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def parse_sky_data(self, sky_arch_list, _i, n_sources):
    """Parse a nested sky-data block belonging to a frequency header.

    The sky-data block encodes coordinates and spectral arrays across
    multiple following lines. Depending on ``n_sources`` the block may
    wrap to multiple groups of five lines.

    Parameters:
        sky_arch_list: list[str]
            Full list of file lines.
        _i: int
            Current index in the file (frequency header line index).
        n_sources: int
            Number of sky sources to extract; determines how many lines
            to consume.

    Returns:
        (data_dict, new_index) where ``data_dict`` contains keys
            ``y_coords``, ``x_coords``, ``spect_amp``, ``spect_dop`` and
            ``rms_error`` and ``new_index`` points to the last consumed
            input index.
    """
    # if n_sources <= 26 then they are in 1 line otherwise check on _i+5 lines
    import re

    y_coords = re.findall(r"-?\d+\.\d+", sky_arch_list[_i + 1].strip())
    y_coords = [float(y) for y in y_coords]
    x_coords = re.findall(r"-?\d+\.\d+", sky_arch_list[_i + 2].strip())
    x_coords = [float(x) for x in x_coords]
    spect_amp = re.findall(r"-?\d+(?:\.\d+)?", sky_arch_list[_i + 3].strip())
    spect_amp = [float(a) for a in spect_amp]
    spect_dop = re.findall(r"-?\d+(?:\.\d+)?", sky_arch_list[_i + 4].strip())
    spect_dop = [float(d) for d in spect_dop]
    rms_error = re.findall(r"-?\d+(?:\.\d+)?", sky_arch_list[_i + 5].strip())
    rms_error = [float(r) for r in rms_error]
    data = dict(
        y_coords=y_coords,  # Y coordinate
        x_coords=x_coords,  # X coordinate
        spect_amp=spect_amp,  # Spectral Amplitude number
        spect_dop=spect_dop,  # Spectral Doppler line number
        rms_error=rms_error,  # Least Squares Fit RMS error
    )
    _i = _i + 5 if n_sources <= 26 else _i + (5 * (int(n_sources / 26) + 1))
    return data, _i

get_doppler_freq(L)

Convert a Doppler bin index to Doppler frequency in MHz.

Parameters:

Name Type Description Default
L float

float Doppler bin index (integer or float) returned by the skymap processing.

 required

Returns:

Type Description
float

Doppler frequency in MHz computed as L * delta_freq / n_fft.
Source code in pynasonde/digisonde/parsers/sky.py 
283
284
285
286
287
288
289
290
291
292
293
294
def get_doppler_freq(self, L: float) -> float:
    """Convert a Doppler bin index to Doppler frequency in MHz.

    Parameters:
        L: float
            Doppler bin index (integer or float) returned by the skymap
            processing.

    Returns:
        Doppler frequency in MHz computed as ``L * delta_freq / n_fft``.
    """
    return L * self.delta_freq / self.n_fft

to_pandas()

Flatten parsed sky records into a pandas DataFrame.

Returns:

Type Description
pd.DataFrame

One row per sky-source with columns for coordinates, amplitudes, Doppler values (and Doppler frequency), RMS error and timestamps where available.
Source code in pynasonde/digisonde/parsers/sky.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
def to_pandas(self) -> pd.DataFrame:
    """Flatten parsed sky records into a pandas DataFrame.

    Returns:
        One row per sky-source with columns for coordinates,
            amplitudes, Doppler values (and Doppler frequency), RMS error
            and timestamps where available.
    """
    df = pd.DataFrame()
    for ds in self.sky.dataset:
        for fh in ds.freq_headers:
            d = []
            if fh.sky_data:
                d.extend(
                    [
                        dict(
                            zenith_angle=fh.zenith_angle,
                            sampl_freq=fh.sampl_freq,
                            group_range=fh.group_range,
                            gain_ampl=fh.gain_ampl,
                            height_spctrum_ampl=fh.height_spctrum_ampl,
                            max_height_spctrum_ampl=fh.max_height_spctrum_ampl,
                            n_sources=fh.n_sources,
                            height_spctrum_cl_th=fh.height_spctrum_cl_th,
                            spect_line_cl_th=fh.spect_line_cl_th,
                            polarization=fh.polarization,
                            x_coord=fh.sky_data.x_coords[i],
                            y_coord=fh.sky_data.y_coords[i],
                            spect_amp=fh.sky_data.spect_amp[i],
                            spect_dop=fh.sky_data.spect_dop[i],
                            spect_dop_freq=self.get_doppler_freq(
                                fh.sky_data.spect_dop[i]
                            ),
                            rms_error=fh.sky_data.rms_error[i],
                            datetime=self.date if hasattr(self, "date") else None,
                            local_datetime=(
                                self.local_time
                                if hasattr(self, "local_time")
                                else None
                            ),
                        )
                        for i in range(len(fh.sky_data.y_coords))
                    ]
                )
            d = pd.DataFrame.from_records(d)
            df = pd.concat([df, d])
    return df