swarmpal.io

class swarmpal.io.ExternalData(source: str = 'vires', collection: str | None = None, model: str | None = None, start_time: str | datetime | None = None, end_time: str | datetime | None = None, pad_times: list[datetime.timedelta] | None = None, parameters: dict | None = None, viresclient_kwargs: dict | None = None, initialise: bool = True)

Bases: object

Fetches and loads data from external sources, e.g. VirES

Parameters

• collection (str) – One of ExternalData.COLLECTIONS

• model (str) – VirES-compatible model specification. Defaults to “CHAOS” (i.e. full CHAOS model)

• start_time (str | datetime) –

• end_time (str | datetime) –

• pad_times (list[datetime.timedelta]) – Extend the requested time window by these two amounts

• source (str) – Defaults to “vires” (only one possible currently)

• parameters (dict) – Override the parameters passed to ViresDataFetcher

• viresclient_kwargs (dict) – Pass extra kwargs to viresclient

Notes

The model variable in the returned data will be renamed to “Model” rather than, e.g., “CHAOS”

Examples

>>> from swarmpal.io import ExternalData
>>> # Customise the class (if not using a subclass)
>>> ExternalData.COLLECTIONS = [f"SW_OPER_MAG{x}_LR_1B" for x in "ABC"]
>>> ExternalData.DEFAULTS["measurements"] = ["F", "B_NEC", "Flags_B"]
>>> ExternalData.DEFAULTS["model"] = "CHAOS"
>>> ExternalData.DEFAULTS["auxiliaries"] = ["QDLat", "QDLon", "MLT"]
>>> ExternalData.DEFAULTS["sampling_step"] = None
>>> # Request data
>>> d = ExternalData(
>>>     collection="SW_OPER_MAGA_LR_1B", model="None",
>>>     start_time="2022-01-01", end_time="2022-01-02",
>>>     viresclient_kwargs=dict(asynchronous=True, show_progress=True)
>>> )
>>> # Access data stored in memory as xarray.Dataset
>>> d.xarray
# The returned dataset will contain "B_NEC" and "B_NEC_Model"

COLLECTIONS: list[str] = []

DEFAULTS: dict = {'auxiliaries': [], 'measurements': [], 'model': '', 'pad_times': None, 'sampling_step': None}

property analysis_window: list[datetime]

append_array(varname, data, dims=('Timestamp',), units=None, description=None)

Append a new variable to the dataset

Parameters

• varname (str) – Name to give to the data variable

• data (ndarray) – Array of data, of same dimensions as dims

• dims (tuple, default=("Timestamp",)) – Dimension names

• units (str) – Units to attach to the data

• description (str) – Description to attach to the data

get_array(variable: str) → numpy.ndarray

Extract numpy array from dataset

initialise(xarray_or_file: Dataset | str | None = None)

Load the data

Parameters

xarray_or_file (Dataset | str | None, optional) – Optionally supply an xarray.Dataset or a file name, by default None

property magnetic_model_name

property source: str

to_file(filepath)

Save the current data to a file

property xarray: xarray.core.dataset.Dataset

class swarmpal.io.MagExternalData(source: str = 'vires', collection: str | None = None, model: str | None = None, start_time: str | datetime | None = None, end_time: str | datetime | None = None, pad_times: list[datetime.timedelta] | None = None, parameters: dict | None = None, viresclient_kwargs: dict | None = None, initialise: bool = True)

Bases: swarmpal.io._data_container.ExternalData

Demo class for accessing magnetic data

Examples

>>> d = MagExternalData(
>>>     collection="SW_OPER_MAGA_LR_1B", model="IGRF",
>>>     start_time="2022-01-01", end_time="2022-01-02",
>>>     viresclient_kwargs=dict(asynchronous=True, show_progress=True)
>>> )
>>> d.xarray  # Returns xarray of data
>>> d.get_array("B_NEC")  # Returns numpy array

COLLECTIONS: list[str] = ['SW_OPER_MAGA_LR_1B', 'SW_OPER_MAGB_LR_1B', 'SW_OPER_MAGC_LR_1B', 'SW_OPER_MAGA_HR_1B', 'SW_OPER_MAGB_HR_1B', 'SW_OPER_MAGC_HR_1B']

DEFAULTS: dict = {'auxiliaries': ['QDLat', 'QDLon', 'MLT'], 'measurements': ['F', 'B_NEC', 'Flags_B'], 'model': 'IGRF', 'sampling_step': None}

property analysis_window: list[datetime]

append_array(varname, data, dims=('Timestamp',), units=None, description=None)

Append a new variable to the dataset

Parameters

• varname (str) – Name to give to the data variable

• data (ndarray) – Array of data, of same dimensions as dims

• dims (tuple, default=("Timestamp",)) – Dimension names

• units (str) – Units to attach to the data

• description (str) – Description to attach to the data

get_array(variable: str) → numpy.ndarray

Extract numpy array from dataset

initialise(xarray_or_file: Dataset | str | None = None)

Load the data

Parameters

xarray_or_file (Dataset | str | None, optional) – Optionally supply an xarray.Dataset or a file name, by default None

property magnetic_model_name

property source: str

to_file(filepath)

Save the current data to a file

property xarray: xarray.core.dataset.Dataset

class swarmpal.io.PalData(name: str | None = None, datatree: DataTree | None = None)

Bases: object

Holds an Xarray DataTree of PAL inputs and outputs

• PalData can be created from PalDataItem objects passed to .set_inputs.

• PalData can be created from existing datatree (e.g. stored on disk)

  See https://xarray-datatree.readthedocs.io/en/latest/io.html

• The data is stored within the .datatree property.

Parameters

• name (str) – Provide a name to label the datatree

• datatree (DataTree) – Provide an existing DataTree directly instead

Examples

>>> from swarmpal.io import PalData, PalDataItem
>>>
>>> params = dict(
>>>     collection="SW_OPER_MAGA_LR_1B",
>>>     measurements=["F", "B_NEC"],
>>>     start_time="2016-01-01T00:00:00",
>>>     end_time="2016-01-02T00:00:00",
>>>     server_url="https://vires.services/ows",
>>> )
>>> my_pal = PalData(name="trial")
>>> my_pal.set_inputs(
>>>     SwarmAlpha=PalDataItem.from_vires(**params)
>>> )
>>> # Access data directly through the datatree
>>> my_pal.datatree["inputs/SwarmAlpha"]

set_inputs(**paldataitems: swarmpal.io._paldata.PalDataItem)

Supply named input datasets through PalDataItem

Parameters

**paldataitems_kw (PalDataItem) – Provide instances of PalDataItem as keyword arguments

class swarmpal.io.PalDataItem(fetcher: swarmpal.io._datafetchers.DataFetcherBase)

Bases: object

An Item (i.e. dataset) that will be stored within a PalData object

Intended use is through the .from_… creator tools

Parameters

fetcher (DataFetcherBase) –

Examples

>>> from swarmpal.io import PalDataItem
>>>
>>> # Create the item with a configuration
>>> params = dict(
>>>     collection="SW_OPER_MAGA_LR_1B",
>>>     measurements=["F", "B_NEC"],
>>>     start_time="2016-01-01T00:00:00",
>>>     end_time="2016-01-02T00:00:00",
>>>     server_url="https://vires.services/ows",
>>> )
>>> item = PalDataItem.from_vires(**params)
>>> # "Initialise" - triggers the expensive part, fetching the data
>>> item.initialise()
>>> # Data is available as an xarray.Dataset
>>> item.xarray
>>> # or as a DataTree
>>> item.datatree

property analysis_window: tuple[datetime]

The start and end times of the analysis window (considering optional padding

property dataset_name: str

Name of the dataset, used as the datatree label

property datatree: datatree.datatree.DataTree

A datatree containing the dataset labelled with the dataset name

static from_file(filename: PathLike | None = None, group: str | None = None, **params) → PalDataItem

Create a PalDataItem from a file

Parameters

• filename (PathLike) – Path to the (netCDF) file to load

• group (str) – Group name within the (netCDF) file

Return type

PalDataItem

static from_hapi(**params) → swarmpal.io._paldata.PalDataItem

Create PalDataItem from HAPI source

TODO: Detail params given by swarmpal.io.datafetchers.HapiParameters

static from_manual(xarray_dataset: Dataset | None = None, **params) → PalDataItem

Create a PalDataItem manually from an existing xarray Dataset

Parameters

xarray_dataset (Dataset) – An existing xarray.Dataset

Return type

PalDataItem

static from_vires(**params) → swarmpal.io._paldata.PalDataItem

Create PalDataItem from VirES source

TODO: Detail params given by swarmpal.io.datafetchers.ViresParameters

initialise()

Trigger the fetching of the data and attach PAL metadata

property magnetic_models: dict

Dictionary of model names and details

property xarray: Dataset | None

xarray.Dataset containing the data, generated if not already present

class swarmpal.io.PalDataTreeAccessor(datatree)

Bases: object

Provide custom attributes and methods on XArray DataTrees for SwarmPAL functionality.

See e.g. https://github.com/Unidata/MetPy/blob/main/src/metpy/xarray.py

apply(palprocess: swarmpal.io._paldata.PalProcess) → datatree.datatree.DataTree

property magnetic_model_name: str

Model name if one and only one has been set

property magnetic_model_names: list[str]

List of the model names used in the dataset

magnetic_residual(model: str = '') → xarray.core.dataarray.DataArray

Magnetic data-model residual in NEC frame

property pal_meta: dict

class swarmpal.io.PalProcess(config: dict | None = None, active_tree: str = '/', inplace: bool = True)

Bases: abc.ABC

Abstract class to define processes to act on datatrees

property active_tree: str

Defines which branch of the datatree will be used

property config: dict

Dictionary that configures the process behaviour

abstract property process_name: str

abstract set_config(**kwargs) → None

class swarmpal.io.ViresDataFetcher(url: str | None = None, parameters: dict | None = None)

Bases: object

Connects to and retrieves data from VirES through viresclient

Parameters

• url (str) – Server URL, defaults to “https://vires.services/ows”

• parameters (dict) – Parameters to pass to viresclient

Examples

>>> from swarmpal.io import ViresDataFetcher
>>> # Initialise request
>>> v = ViresDataFetcher(
>>>     parameters={
>>>         'collection': 'SW_OPER_MAGA_LR_1B',
>>>         'measurements': ['F', 'B_NEC', 'Flags_B'],
>>>         'models': ['CHAOS'],
>>>         'auxiliaries': ['QDLat', 'QDLon'],
>>>         'sampling_step': None
>>>     }
>>> )
>>> # Fetch data and extract as xarray.Dataset
>>> ds = v.fetch_data("2022-01-01", "2022-01-02")

VIRES_URL = 'https://vires.services/ows'

fetch_data() → xarray.core.dataset.Dataset

property parameters: dict

swarmpal.io.create_paldata(*paldataitems: swarmpal.io._paldata.PalDataItem, **paldataitems_kw: swarmpal.io._paldata.PalDataItem) → datatree.datatree.DataTree

Generates a Datatree from a number of PalDataItems

Returns

A Datatree containing Datasets defined from each PalDataItem

Return type

Datatree

Examples

>>> from swarmpal.io import create_paldata, PalDataItem
>>>
>>> # Parameters to control a particular data request
>>> data_params = dict(
>>>     collection="SW_OPER_MAGA_LR_1B",
>>>     measurements=["B_NEC"],
>>>     models=["IGRF"],
>>>     start_time="2016-01-01T00:00:00",
>>>     end_time="2016-01-01T03:00:00",
>>>     server_url="https://vires.services/ows",
>>>     options=dict(asynchronous=False, show_progress=False),
>>> )
>>> # Create the datatree from a list of items
>>> data = create_paldata(
>>>     PalDataItem.from_vires(**data_params)
>>> )
>>> # Create the datatree from labelled items
>>> data = create_paldata(
>>>     one=PalDataItem.from_vires(**data_params),
>>>     two=PalDataItem.from_vires(**data_params),
>>> )