Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Copyright (c) 2013-2015 Siphon Contributors.
# Distributed under the terms of the BSD 3-Clause License.
# SPDX-License-Identifier: BSD-3-Clause
"""Support talking to the CDMRemote endpoint on a THREDDS Data Server (TDS)."""
from io import BytesIO
from .ncstream import read_ncstream_messages
from ..http_util import HTTPEndPoint
class CDMRemote(HTTPEndPoint):
"""Provide access to the various methods on the CDMRemote endpoint on a TDS."""
def __init__(self, url):
"""Initialize access to a particular url."""
super(CDMRemote, self).__init__(url)
self.deflate = 0
def _fetch(self, query):
return read_ncstream_messages(BytesIO(self.get_query(query).content))
def fetch_capabilities(self):
"""Query the CDMRemote end point for its capabilities."""
return self.get_query(self.query().add_query_parameter(req='capabilities'))
def fetch_cdl(self):
"""Retrieve the CDL response from CDMRemote."""
import numpy as np
from .http_util import DataQuery, HTTPEndPoint, parse_iso_date
from .ncss_dataset import NCSSDataset
def default_unit_handler(data, units=None): # pylint:disable=unused-argument
"""Handle units in the default manner.
Ignores units and just returns :func:`numpy.array`.
"""
return np.array(data)
class NCSS(HTTPEndPoint):
"""Wrap access to the NetCDF Subset Service (NCSS) on a THREDDS server.
Simplifies access via HTTP to the NCSS endpoint. Parses the metadata, provides
data download and parsing based on the appropriate query.
Attributes
----------
metadata : NCSSDataset
Contains the result of parsing the NCSS endpoint's dataset.xml. This has
information about the time and space coverage, as well as full information
about all of the variables.
variables : set(str)
Names of all variables available in this dataset
unit_handler : callable
Function to handle units that come with CSV/XML data. Should be a callable that
takes a list of string values and unit str (can be :data:`None`), and returns the
from datetime import datetime
import json
import warnings
import numpy as np
import numpy.ma as ma
import pandas as pd
from .._tools import get_wind_components
from ..http_util import HTTPEndPoint
warnings.filterwarnings('ignore', "Pandas doesn't allow columns to be created", UserWarning)
class IAStateUpperAir(HTTPEndPoint):
"""Download and parse data from the Iowa State's upper air archive."""
def __init__(self):
"""Set up endpoint."""
super(IAStateUpperAir, self).__init__('http://mesonet.agron.iastate.edu/json')
@classmethod
def request_data(cls, time, site_id, interp_nans=False, **kwargs):
"""Retrieve upper air observations from Iowa State's archive for a single station.
Parameters
----------
time : datetime
The date and time of the desired observation.
site_id : str
from io import StringIO
import itertools
import sys
import warnings
from zipfile import ZipFile
import numpy as np
import pandas as pd
from .._tools import get_wind_components
from ..http_util import HTTPEndPoint, HTTPError
warnings.filterwarnings('ignore', "Pandas doesn't allow columns to be created", UserWarning)
class IGRAUpperAir(HTTPEndPoint):
"""Download and parse data from NCEI's Integrated Radiosonde Archive version 2."""
def __init__(self):
"""Set http site address and file suffix based on desired dataset."""
self.suffix = ''
self.begin_date = ''
self.end_date = ''
self.site_id = ''
self.folder = ''
super(IGRAUpperAir, self).__init__('https://www1.ncdc.noaa.gov/pub/data/igra/')
@classmethod
def request_data(cls, time, site_id, derived=False):
"""Retreive IGRA version 2 data for one station.
Parameters
# SPDX-License-Identifier: BSD-3-Clause
"""Read data from the National Data Buoy Center."""
from io import StringIO
import warnings
import numpy as np
import pandas as pd
import requests
from ..http_util import HTTPEndPoint
warnings.filterwarnings('ignore', "Pandas doesn\'t allow columns to be created", UserWarning)
class NDBC(HTTPEndPoint):
"""Download and parse data from the National Data Buoy Center."""
def __init__(self):
"""Set up endpoint."""
super(NDBC, self).__init__('https://www.ndbc.noaa.gov/')
@classmethod
def realtime_observations(cls, buoy, data_type='txt'):
"""Retrieve the realtime buoy data from NDBC.
Parameters
----------
buoy : str
Name of buoy
data_type : str
Type of data requested, must be one of
Parameters
----------
stns : one or more strings
One or more names of variables to request
Returns
-------
self : RadarQuery
Returns self for chaining calls
"""
self._set_query(self.spatial_query, stn=stns)
return self
class RadarServer(HTTPEndPoint):
"""Wrap access to the THREDDS radar query service (radar server).
Simplifies access via HTTP to the radar server endpoint. Parses the metadata, provides
query catalog results download and parsing based on the appropriate query.
Attributes
----------
metadata : :class:`~siphon.metadata.TDSCatalogMetadata`
Contains the result of parsing the radar server endpoint's dataset.xml. This has
information about the time and space coverage, as well as full information
about all of the variables.
variables : set(str)
Names of all variables available in this dataset
stations : dict[str, Station]
Mapping of station ID to a :class:`Station`, which is a namedtuple containing the
station's id, name, latitude, longitude, and elevation.
from datetime import datetime
from io import StringIO
import warnings
from bs4 import BeautifulSoup
import numpy as np
import pandas as pd
from .._tools import get_wind_components
from ..http_util import HTTPEndPoint
warnings.filterwarnings('ignore', 'Pandas doesn\'t allow columns to be created', UserWarning)
class WyomingUpperAir(HTTPEndPoint):
"""Download and parse data from the University of Wyoming's upper air archive."""
def __init__(self):
"""Set up endpoint."""
super(WyomingUpperAir, self).__init__('http://weather.uwyo.edu/cgi-bin/sounding')
@classmethod
def request_data(cls, time, site_id, **kwargs):
r"""Retrieve upper air observations from the Wyoming archive.
Parameters
----------
time : datetime
The date and time of the desired observation.
site_id : str