stormgeo.advisor-core 1.6.0

SDK to access the advisor core API.

Python SDK

Advisor Software Development Kit for python.

Contents

• Python SDK
  • How to get your token
  • Contents
  • Importing
  • Routes
    • Examples
      • Chart
      • Climatology
      • Current Weather
      • Forecast
      • Monitoring
      • Observed
      • Stations
      • Plan Information
      • Schema/Parameter
      • Static Map
      • Storage
      • Tms (Tiles Map Server)
      • Pmtiles
  • Headers Configuration
  • Response Format
  • Payload Types
    • WeatherPayload
    • StationPayload
    • StationsLastDataPayload
    • ClimatologyPayload
    • CurrentWeatherPayload
    • RadiusPayload
    • GeometryPayload
    • LightningLitePayload
    • TmsPayload
    • PmtilesPayload
    • PlanInfoPayload
    • RequestDetailsPayload
    • PlanLocalePayload
    • StorageListPayload
    • StorageDownloadPayload
    • StaticMapPayload

---

Importing

To install this package, use the following command:`

pip install stormgeo.advisor-core

Make sure you're using python 3.8 or higher.

Routes

First you need to import the SDK on your application and instancy the AdvisorCore class setting up your access token and needed configurations:

from advisor_core import *

advisor = AdvisorCore("<your_token>", retries=5, delay=5)

Examples

Get data from different routes with theses examples

Chart

payload = WeatherPayload(
  locale_id=1234,
  variables=["temperature", "precipitation"]
)

# requesting daily forecast chart image
response = advisor.chart.get_forecast_daily(payload)

# requesting hourly forecast chart image
response = advisor.chart.get_forecast_hourly(payload)

# requesting daily observed chart image
response = advisor.chart.get_observed_daily(payload)

# requesting hourly observed chart image
response = advisor.chart.get_observed_hourly(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  with open("response.png", "wb") as f:
    f.write(response["data"])

Climatology

payload = ClimatologyPayload(
  locale_id=1234,
  variables=["temperature", "precipitation"]
)

# requesting daily climatology data
response = advisor.climatology.get_daily(payload)

# requesting monthly climatology data
response = advisor.climatology.get_monthly(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Current Weather

payload = CurrentWeatherPayload(
  locale_id=1234
)

response = advisor.current_weather.get(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Forecast

payload = WeatherPayload(
  locale_id=1234,
  variables=["temperature", "precipitation"]
)

# requesting daily forecast data
response = advisor.forecast.get_daily(payload)

# requesting hourly forecast data
response = advisor.forecast.get_hourly(payload)

# requesting period forecast data
response = advisor.forecast.get_period(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Monitoring

response = advisor.monitoring.get_alerts()

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Observed

payload = WeatherPayload(
  locale_id=1234,
)

payload_for_station = StationPayload(
  station_id="ABC123abc321CBA"
)

payload_for_radius = RadiusPayload(
  locale_id=1234,
  radius=1000
)

payload_for_geometry = GeometryPayload(
  geometry="{\"type\": \"MultiPoint\", \"coordinates\": [[-41.88, -22.74]]}",
  radius=10000
)

payload_for_lightning_lite = LightningLitePayload(
  geometry="{\"type\": \"MultiPoint\", \"coordinates\": [[-41.88, -22.74]]}",
  radius=10000,
  page=1,
  page_size=10
)

# requesting daily observed data
response = advisor.observed.get_daily(payload)

# requesting hourly observed data
response = advisor.observed.get_hourly(payload)

# requesting period observed data
response = advisor.observed.get_period(payload)

# requesting station observed data
response = advisor.observed.get_station_data(payload_for_station)

# requesting fire-focus observed data
response = advisor.observed.get_fire_focus(payload_for_radius)

# requesting lightning observed data
response = advisor.observed.get_lightning(payload_for_radius)

# requesting lightning observed details data
response = advisor.observed.get_lightning_details(payload_for_radius)

# requesting fire-focus observed data by geometry
response = advisor.observed.get_fire_focus_by_geometry(payload_for_geometry)

# requesting lightning observed data by geometry
response = advisor.observed.get_lightning_by_geometry(payload_for_geometry)

# requesting lightning lite observed data by geometry
response = advisor.observed.get_lightning_lite(payload_for_lightning_lite)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Stations

payload = StationsLastDataPayload(
  station_ids=["ABC123abc321CBA", "XYZ789xyz987ZYX"], # optional
  variables=["temperature", "humidity"] # optional
)

# requesting last observed data for multiple stations
response = advisor.stations.get_last_data(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Storage

payload = StorageListPayload(
  page=1,
  page_size=10,
  file_types=["pdf", "csv"]
)

payload_for_download = StorageDownloadPayload(
  file_name="Example.pdf",
  access_key="a1b2c3d4-0010"
)

# requesting the files list
response = advisor.storage.list_files(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

# downloading a file from the list
response = advisor.storage.download_file(payload_for_download)

if response['error']:
    print('Error trying to get data')
    print(response['error'])
else:
    with open(payload_for_download.file_name, "wb") as f:
        f.write(response["data"])

# downloading a file by stream
response = advisor.storage.download_file_by_stream(payload_for_download)

if response['error']:
    print('Error trying to get data')
    print(response['error'])
else:
    with open(payload_for_download.file_name, "wb") as f:
        for chunk in response['data']:
            if chunk:
                f.write(chunk)

Plan Information

# requesting plan information
plan_info_payload = PlanInfoPayload(
    timezone=-3
)

plan_info_response = advisor.plan.get_info(plan_info_payload)

if plan_info_response['error']:
    print('Error trying to get plan information!')
    print(plan_info_response['error'])
else:
    print(plan_info_response['data'])

# requesting locale details
plan_locale_payload = PlanLocalePayload(
    locale_id=1234,
    # You can also set Latitude/Longitude or StationId instead of LocaleId
)

plan_locale_response = advisor.plan.get_locale(plan_locale_payload)

if plan_locale_response['error']:
    print('Error trying to get plan locale!')
    print(plan_locale_response['error'])
else:
    print(plan_locale_response['data'])

# requesting access history
request_details_payload = RequestDetailsPayload(
    page=1,
    page_size=10
)

request_details_response = advisor.plan.get_request_details(request_details_payload)

if request_details_response['error']:
    print('Error trying to get request details!')
    print(request_details_response['error'])
else:
    print(request_details_response['data'])

Static Map

payload = StaticMapPayload(
    type="periods",
    category="observed",
    variable="temperature",
    aggregation="max",
    start_date="2025-07-01 00:00:00",
    end_date="2025-07-05 23:59:59",
    dpi=50,
    title=True,
    titlevariable="Static Map",
)

response = advisor.static_map.get_static_map(payload)

if response['error']:
    print('Error trying to get data!')
    print(response['error'])
else:
    with open("response.png", "wb") as f:
        f.write(response["data"])

Schema/Parameter

# Arbitrary example on how to define a schema
payload_schema_definition = {
  "identifier": "arbitraryIdentifier",
  "arbitraryField1": {
      "type": "boolean",
      "required": True,
      "length": 125,
  },
  "arbitraryField2": {
      "type": "number",
      "required": True,
  },
  "arbitraryField3": {
      "type": "string",
      "required": False,
  }
}

# Arbitrary example on how to upload data to parameters from schema 
payload_schema_parameters = {
  "identifier": "arbitraryIdentifier",
  "arbitraryField1": True,
  "arbitraryField2": 15
}

# requesting all schemas from token
response = advisor.schema.get_definition()

# requesting to upload a new schema
response = advisor.schema.post_definition(payload_schema_definition)

# requesting to upload data to parameters from schema
response = advisor.schema.post_parameters(payload_schema_parameters)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  print(response['data'])

Tms (Tiles Map Server)

payload = TmsPayload(
  istep="2024-12-25 10:00:00",
  fstep="2024-12-25 12:00:00",
  server="a",
  mode="forecast",
  variable="precipitation",
  aggregation="sum",
  x=2,
  y=3,
  z=4
)

response = advisor.tms.get(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  with open("response.png", "wb") as f:
    f.write(response["data"])

Pmtiles

payload = PmtilesPayload(
  mode="forecast",
  model="ct2w15_as",
  variable="precipitation",
  aggregation="sum",
  istep="2026-03-02 00:00:00",
  fstep="2026-03-02 01:00:00",
  max_zoom=4,
)

response = advisor.pmtiles.get(payload)

if response['error']:
  print('Error trying to get data!')
  print(response['error'])
else:
  with open("response.pmtiles", "wb") as f:
    f.write(response["data"])

---

Headers Configuration

You can also set headers to translate the error descriptions or to receive the response in a different format type. This functionality is only available for some routes, consult the API documentation to find out which routes have this functionality.

Available languages:

• en-US (default)
• pt-BR
• es-ES

Available response types:

• application/json (default)
• application/xml
• text/csv

Example:

advisor = AdvisorCore("invalid-token")

advisor.setHeaderAccept("application/xml")
advisor.setHeaderAcceptLanguage("es-ES")

response = advisor.plan.get_info()

print(response["error"])

// <response>
//   <error>
//     <type>UNAUTHORIZED_ACCESS</type>
//     <message>UNAUTHORIZED_REQUEST</message>
//     <description>La solicitud no está autorizada.</description>
//   </error>
// </response>

Response Format

All the methods will return the same pattern:

{
  "data": Any | None,
  "error": Any | None,
}

Payload Types

WeatherPayload

• locale_id: int
• station_id: str
• latitude: float
• longitude: float
• timezone: int
• variables: List[str]
• start_date: str
• end_date: str

StationPayload

• station_id: str
• layer: str
• timezone: int
• variables: List[str]
• start_date: str
• end_date: str

StationsLastDataPayload

• station_ids: List[str]
• variables: List[str]

ClimatologyPayload

• locale_id: int
• station_id: str
• latitude: float
• longitude: float
• variables: List[str]

CurrentWeatherPayload

• locale_id: int
• station_id: str
• latitude: float
• longitude: float
• timezone: int
• variables: List[str]

RadiusPayload

• locale_id: int
• station_id: str
• latitude: float
• longitude: float
• start_date: str
• end_date: str
• radius: int

GeometryPayload

• start_date: str
• end_date: str
• radius: int
• geometry: str

LightningLitePayload

• start_date: str
• end_date: str
• radius: int
• geometry: str
• page: number
• page_size: number
• sources: List[str]

TmsPayload

• server: str
• mode: str
• variable: str
• aggregation: str
• x: int
• y: int
• z: int
• istep: str
• fstep: str
• timezone: int

PmtilesPayload

• mode: str
• model: str
• variable: str
• aggregation: str
• istep: str
• fstep: str
• timezone: int
• max_zoom: int
• cmap: str
• dynamic_elevation: str
• dynamic_type: str
• dynamic_variable: str

PlanInfoPayload

• timezone: int

PlanLocalePayload

• locale_id: int
• station_id: str
• latitude: str
• longitude: str

RequestDetailsPayload

• page: int
• page_size: int
• path: str
• status: int
• start_date: str
• end_date: str

StorageListPayload

• page: int
• page_size: int
• start_date: str
• end_date: str
• file_name: str
• file_extension: str
• file_types: List[str]

StorageDownloadPayload

• file_name: str
• access_key: str

StaticMapPayload

• start_date: str
• end_date: str
• aggregation: str
• model: str
• lonmin: float
• latmin: float
• lonmax: float
• latmax: float
• dpi: int
• title: bool
• titlevariable: str
• hours: int
• type: str
• category: str
• variable: str

---