Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/api.rst
Original file line number Diff line number Diff line change
Expand Up @@ -15,3 +15,5 @@ API Reference

.. automodapi:: hvpy.datasources
:no-inheritance-diagram:

.. automodapi:: hvpy.utils
1 change: 1 addition & 0 deletions hvpy/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,3 +2,4 @@
from .config import set_api_url
from .version import __version__
from .datasources import *
from .utils import create_layers
34 changes: 17 additions & 17 deletions hvpy/facade.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@

from hvpy.core import execute_api_call
from hvpy.parameters import *
from hvpy.utils import add_shared_docstring
from hvpy.utils import _add_shared_docstring

__all__ = [
"getJP2Image",
Expand All @@ -25,7 +25,7 @@
]


@add_shared_docstring(getJP2ImageInputParameters)
@_add_shared_docstring(getJP2ImageInputParameters)
def getJP2Image(
date: datetime,
sourceId: int,
Expand All @@ -49,7 +49,7 @@ def getJP2Image(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getJP2HeaderInputParameters)
@_add_shared_docstring(getJP2HeaderInputParameters)
def getJP2Header(
id: int,
callback: Optional[str] = None,
Expand All @@ -71,7 +71,7 @@ def getJP2Header(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getJPXClosestToMidPointInputParameters)
@_add_shared_docstring(getJPXClosestToMidPointInputParameters)
def getJPXClosestToMidPoint(
startTimes: List[datetime],
endTimes: List[datetime],
Expand Down Expand Up @@ -111,7 +111,7 @@ def getJPXClosestToMidPoint(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getJPXInputParameters)
@_add_shared_docstring(getJPXInputParameters)
def getJPX(
startTime: List[datetime],
endTime: List[datetime],
Expand Down Expand Up @@ -153,7 +153,7 @@ def getJPX(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getStatusInputParameters)
@_add_shared_docstring(getStatusInputParameters)
def getStatus() -> Union[bytes, str, Dict[str, Any]]:
"""
Returns information about how far behind the latest available JPEG2000
Expand All @@ -172,7 +172,7 @@ def getStatus() -> Union[bytes, str, Dict[str, Any]]:
return execute_api_call(input_parameters=params)


@add_shared_docstring(getClosestImageInputParameters)
@_add_shared_docstring(getClosestImageInputParameters)
def getClosestImage(
date: datetime,
sourceId: int,
Expand Down Expand Up @@ -200,7 +200,7 @@ def getClosestImage(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getDataSourcesInputParameters)
@_add_shared_docstring(getDataSourcesInputParameters)
def getDataSources(
verbose: Optional[bool] = False,
enable: Optional[str] = None,
Expand All @@ -226,7 +226,7 @@ def getDataSources(
return execute_api_call(input_parameters=params)


@add_shared_docstring(takeScreenshotInputParameters)
@_add_shared_docstring(takeScreenshotInputParameters)
def takeScreenshot(
date: datetime,
imageScale: float,
Expand Down Expand Up @@ -295,7 +295,7 @@ def takeScreenshot(
return execute_api_call(input_parameters=params)


@add_shared_docstring(downloadScreenshotInputParameters)
@_add_shared_docstring(downloadScreenshotInputParameters)
def downloadScreenshot(id: int) -> Union[bytes, str, Dict[str, Any]]:
"""
Download a custom screenshot that was generated using the
Expand All @@ -314,7 +314,7 @@ def downloadScreenshot(id: int) -> Union[bytes, str, Dict[str, Any]]:
return execute_api_call(input_parameters=params)


@add_shared_docstring(queueMovieInputParameters)
@_add_shared_docstring(queueMovieInputParameters)
def queueMovie(
startTime: datetime,
endTime: datetime,
Expand Down Expand Up @@ -398,7 +398,7 @@ def queueMovie(
return execute_api_call(input_parameters=params)


@add_shared_docstring(reQueueMovieInputParameters)
@_add_shared_docstring(reQueueMovieInputParameters)
def reQueueMovie(
id: str,
force: Optional[bool] = False,
Expand All @@ -422,7 +422,7 @@ def reQueueMovie(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getMovieStatusInputParameters)
@_add_shared_docstring(getMovieStatusInputParameters)
def getMovieStatus(
id: str,
format: str,
Expand Down Expand Up @@ -452,7 +452,7 @@ def getMovieStatus(
return execute_api_call(input_parameters=params)


@add_shared_docstring(downloadMovieInputParameters)
@_add_shared_docstring(downloadMovieInputParameters)
def downloadMovie(
id: str,
format: str,
Expand All @@ -479,7 +479,7 @@ def downloadMovie(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getNewsFeedInputParameters)
@_add_shared_docstring(getNewsFeedInputParameters)
def getNewsFeed(
callback: Optional[str] = None,
) -> Union[bytes, str, Dict[str, Any]]:
Expand All @@ -501,7 +501,7 @@ def getNewsFeed(
return execute_api_call(input_parameters=params)


@add_shared_docstring(shortenURLInputParameters)
@_add_shared_docstring(shortenURLInputParameters)
def shortenURL(
queryString: str,
callback: Optional[str] = None,
Expand All @@ -525,7 +525,7 @@ def shortenURL(
return execute_api_call(input_parameters=params)


@add_shared_docstring(getTileInputParameters)
@_add_shared_docstring(getTileInputParameters)
def getTile(
id: int,
x: int,
Expand Down
32 changes: 32 additions & 0 deletions hvpy/tests/test_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
import pytest

from hvpy import DataSources
from hvpy.utils import _create_layer_string, _to_datasource, create_layers


def test_create_layers():
assert create_layers([(9, 100), (11, 50)]) == "[9,1,100],[11,1,50]"

with pytest.raises(ValueError, match="999 is not a valid DataSources"):
create_layers([(9, 100), (999, 100)])

with pytest.raises(ValueError, match="must be between 0 and 100"):
create_layers([(9, 101), (14, 100)])

with pytest.raises(ValueError, match="must be between 0 and 100"):
create_layers([(9, 100), (14, -1)])


def test_to_datasource():
assert _to_datasource(9) == DataSources.AIA_131
assert _to_datasource(DataSources.AIA_94) == DataSources.AIA_94
with pytest.raises(ValueError, match="999 is not a valid DataSources"):
_to_datasource(999)


def test_create_layers_string():
assert _create_layer_string(DataSources.AIA_131, 100) == "[9,1,100]"
with pytest.raises(ValueError, match="must be between 0 and 100"):
_create_layer_string(DataSources.AIA_131, 101)
with pytest.raises(ValueError, match="must be between 0 and 100"):
_create_layer_string(DataSources.AIA_131, -1)
52 changes: 49 additions & 3 deletions hvpy/utils.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,16 @@
from typing import Any, Callable
from typing import Any, Union, Callable
from datetime import datetime

__all__ = ["add_shared_docstring", "convert_date_to_isoformat", "convert_date_to_unix"]
from hvpy.datasources import DataSources

__all__ = [
"convert_date_to_isoformat",
"convert_date_to_unix",
"create_layers",
]

def add_shared_docstring(input_class) -> Callable[[Any], Any]:

def _add_shared_docstring(input_class) -> Callable[[Any], Any]:
def decorator(func):
if "{Shared}" in input_class.__doc__:
split_doc = input_class.__doc__.split("{Shared}")
Expand All @@ -29,3 +35,43 @@ def convert_date_to_unix(v: list) -> str:
commas.
"""
return ",".join([str(int(datetime.timestamp(d))) for d in v])


def _to_datasource(val: Union[int, DataSources]) -> DataSources:
"""
Validates the input and converts it to a DataSources enum.
"""
if isinstance(val, int) and val in [x.value for x in DataSources]:
return DataSources(val)
elif isinstance(val, DataSources):
return val
else:
raise ValueError(f"{val} is not a valid DataSources")


def _create_layer_string(source_id: DataSources, opacity: int) -> str:
"""
Generates a string of the form "[source_id,1,opacity]" for a layer.
"""
if not 0 <= opacity <= 100:
raise ValueError(f"opacity ({opacity}) must be between 0 and 100")
else:
return f"[{source_id.value},1,{opacity}]"


def create_layers(layer: list) -> str:
"""
Creates a string of layers separated by commas.

Parameters
----------
layer
A list of tuples of the form (``source_id``, ``opacity``).

Examples
--------
>>> from hvpy import create_layers
>>> create_layers([(3, 50), (10, 50)])
'[3,1,50],[10,1,50]'
"""
return ",".join([_create_layer_string(_to_datasource(s), o) for s, o in layer])