Source code for helios.alerts_api

"""
Helios Alerts API.

Methods are meant to represent the core functionality in the developer
documentation.  Some may have additional functionality for convenience.

"""
import logging

from helios.core.mixins import SDKCore, IndexMixin, ShowMixin

logger = logging.getLogger(__name__)


[docs]class Alerts(ShowMixin, IndexMixin, SDKCore): """ Helios alerts provide real-time severe weather alerts. **National Weather Service:** - Severe weather alerts for the United States are provided by the National Weather Service. These alerts cover events such as Flood Warnings, Severe Thunderstorm Warnings, and Special Weather Statements. **Helios:** - Alerts generated by Helios are based on the sensor measurements from the Observations API. These alerts represent regional areas with a high detection confidence and currently include: Road Wetness Watch, Poor Visibility Watch, and Heavy Precip Watch. """ _core_api = 'alerts' def __init__(self, session=None): """ Initialize Alerts instance. Args: session (helios.HeliosSession): A HeliosSession instance. """ super().__init__(session=session)
[docs] def index(self, **kwargs): """ Get alerts matching the provided spatial, text, or metadata filters. The maximum skip value is 4000. If this is reached, truncated results will be returned. You will need to refine your query to avoid this. .. _alerts_index_documentation: https://helios.earth/developers/api/alerts/#index Args: **kwargs: Any keyword arguments found in the alerts_index_documentation_. Returns: tuple: A tuple containing: feature_collection (:class:`AlertsFeatureCollection <helios.alerts_api.AlertsFeatureCollection>`): Alerts feature collection. failed (list of :class:`Record <helios.core.structure.Record>`): Failed API call records. """ succeeded, failed = super().index(**kwargs) content = [] for record in succeeded: for features in record.content['features']: content.append(AlertsFeature(features)) return AlertsFeatureCollection(content), failed
[docs] def show(self, alert_ids): """ Get attributes for alerts. Args: alert_ids (str or list of strs): Helios alert ID(s). Returns: tuple: A tuple containing: feature_collection (:class:`AlertsFeatureCollection <helios.alerts_api.AlertsFeatureCollection>`): Alerts feature collection. failed (list of :class:`Record <helios.core.structure.Record>`): Failed API call records. """ succeeded, failed = super().show(alert_ids) content = [] for record in succeeded: content.append(AlertsFeature(record.content)) return AlertsFeatureCollection(content), failed
[docs]class AlertsFeature: """ Individual Alert GeoJSON feature. Attributes: area_description (str): 'areaDesc' value for the feature. bbox (list of floats): 'bbox' value for the feature. category (str): 'category' value for the feature. certainty (str): 'certainty' value for the feature. country (str): 'country' value for the feature. description (str): 'description' value for the feature. effective (str): 'effective' value for the feature. event (str): 'event' value for the feature. expires (str): 'expires' value for the feature. headline (str): 'headline' value for the feature. id (str): 'id' value for the feature. json (dict): Raw 'json' for the feature. origin (str): 'origin' value for the feature. severity (str): 'severity' value for the feature. states (list of strs): 'states' value for the feature. status (str): 'status' value for the feature. urgency (str): 'urgency' value for the feature. """ def __init__(self, feature): self.json = feature @property def area_description(self): return self.json['properties'].get('areaDesc') @property def bbox(self): return self.json.get('bbox') @property def category(self): return self.json['properties'].get('category') @property def certainty(self): return self.json['properties'].get('certainty') @property def country(self): return self.json['properties'].get('country') @property def description(self): return self.json['properties'].get('description') @property def effective(self): return self.json['properties'].get('effective') @property def event(self): return self.json['properties'].get('event') @property def expires(self): return self.json['properties'].get('expires') @property def headline(self): return self.json['properties'].get('headline') @property def id(self): return self.json.get('id') @property def origin(self): return self.json['properties'].get('origin') @property def severity(self): return self.json['properties'].get('severity') @property def states(self): return self.json['properties'].get('states') @property def status(self): return self.json['properties'].get('status') @property def urgency(self): return self.json['properties'].get('urgency')
[docs]class AlertsFeatureCollection: """ Collection of GeoJSON features obtained via the Alerts API. Convenience properties are available to extract values from every feature. Attributes: features (list of :class:`AlertsFeature <helios.alerts_api.AlertsFeature>`): All features returned from a query. """ def __init__(self, features): self.features = features @property def area_description(self): """'areaDesc' values for every feature.""" return [x.area_description for x in self.features] @property def bbox(self): """'bbox' values for every feature.""" return [x.bbox for x in self.features] @property def category(self): """'category' values for every feature.""" return [x.category for x in self.features] @property def certainty(self): """'certainty' values for every feature.""" return [x.certainty for x in self.features] @property def country(self): """'country' values for every feature.""" return [x.country for x in self.features] @property def description(self): """'description' values for every feature.""" return [x.description for x in self.features] @property def effective(self): """'effective' values for every feature.""" return [x.effective for x in self.features] @property def event(self): """'event' values for every feature.""" return [x.event for x in self.features] @property def expires(self): """'expires' values for every feature.""" return [x.expires for x in self.features] @property def headline(self): """'headline' values for every feature.""" return [x.headline for x in self.features] @property def id(self): """'id' values for every feature.""" return [x.id for x in self.features] @property def json(self): """Raw 'json' for every feature.""" return [x.json for x in self.features] @property def origin(self): """'origin' values for every feature.""" return [x.origin for x in self.features] @property def severity(self): """'severity' values for every feature.""" return [x.severity for x in self.features] @property def states(self): """'states' values for every feature.""" return [x.states for x in self.features] @property def status(self): """'status' values for every feature.""" return [x.status for x in self.features] @property def urgency(self): """'urgency' values for every feature.""" return [x.urgency for x in self.features]