-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Mobile] - Add fetch wetland data api
- Loading branch information
1 parent
0cfebf9
commit 19c439a
Showing
5 changed files
with
163 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
from django.test import TestCase | ||
from unittest.mock import patch | ||
|
||
from requests import HTTPError | ||
|
||
from bims.utils.feature_info import get_feature_info_from_wms | ||
|
||
|
||
class TestGetFeatureInfoFromWMS(TestCase): | ||
|
||
@patch('requests.get') | ||
def test_successful_request(self, mock_get): | ||
mock_response = mock_get.return_value | ||
mock_response.status_code = 200 | ||
mock_response.json.return_value = {'feature': 'some_info'} | ||
|
||
result = get_feature_info_from_wms( | ||
"http://your.wms.server", | ||
"your_layer", "EPSG:4326", 40.7128, -74.0060, 800, 600) | ||
|
||
self.assertEqual(result, {'feature': 'some_info'}) | ||
|
||
@patch('requests.get') | ||
def test_unsuccessful_request(self, mock_get): | ||
mock_response = mock_get.return_value | ||
mock_response.status_code = 404 | ||
|
||
result = get_feature_info_from_wms( | ||
"http://your.wms.server", | ||
"your_layer", "EPSG:4326", 40.7128, -74.0060, 800, 600) | ||
|
||
self.assertEqual(result, None) | ||
|
||
@patch('requests.get') | ||
def test_request_exception(self, mock_get): | ||
mock_get.side_effect = HTTPError() | ||
|
||
result = get_feature_info_from_wms( | ||
"http://your.wms.server", | ||
"your_layer", "EPSG:4326", 40.7128, -74.0060, 800, 600) | ||
|
||
self.assertEqual(result, None) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,55 @@ | ||
from typing import Union, Dict | ||
|
||
import requests | ||
from requests import HTTPError | ||
|
||
|
||
def get_feature_info_from_wms( | ||
base_wms_url: str, | ||
layer: str, srs: str, | ||
latitude: float, | ||
longitude: float, width: int, height: int) -> Union[Dict, None]: | ||
""" | ||
Fetch feature information from a WMS layer based on given latitude and longitude. | ||
Parameters: | ||
base_wms_url (str): The base URL of the WMS service. | ||
layer (str): The name of the WMS layer to query. | ||
srs (str): The Spatial Reference System (e.g., "EPSG:4326"). | ||
latitude (float): The latitude of the location. | ||
longitude (float): The longitude of the location. | ||
width (int): The width of the map in pixels. | ||
height (int): The height of the map in pixels. | ||
Returns: | ||
Union[Dict, None]: Returns a dictionary containing the feature info if successful, or None if unsuccessful. | ||
Examples: | ||
>>> get_feature_info_from_wms("http://your.wms.server", "your_layer", "EPSG:4326", 40.7128, -74.0060, 800, 600) | ||
{...} # Feature information in dictionary form | ||
""" | ||
|
||
params = { | ||
'request': 'GetFeatureInfo', | ||
'service': 'WMS', | ||
'srs': srs, | ||
'version': '1.1.1', | ||
'format': 'image/png', | ||
'layers': layer, | ||
'query_layers': layer, | ||
'info_format': 'application/json', | ||
'width': width, | ||
'height': height, | ||
'x': int(width / 2), | ||
'y': int(height / 2), | ||
'bbox': f"{longitude-0.1},{latitude-0.1},{longitude+0.1},{latitude+0.1}" | ||
} | ||
|
||
try: | ||
response = requests.get(base_wms_url, params=params) | ||
if response.status_code == 200: | ||
return response.json() | ||
except HTTPError: | ||
return None | ||
|
||
return None |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,60 @@ | ||
from typing import Union, Dict | ||
|
||
from guardian.mixins import LoginRequiredMixin | ||
from rest_framework.views import APIView | ||
from rest_framework.response import Response | ||
|
||
from bims.utils.feature_info import get_feature_info_from_wms | ||
|
||
|
||
def fetch_wetland_data(latitude: float, longitude: float) -> Union[Dict, None]: | ||
""" | ||
Fetch wetland data from GeoContext based on given latitude and longitude. | ||
Parameters: | ||
latitude (float): The latitude of the LocationSite. | ||
longitude (float): The longitude of the LocationSite. | ||
Returns: | ||
Union[Dict, None]: Returns a dictionary containing the wetland data if the fetch is successful. | ||
Returns None if the fetch fails. | ||
Examples: | ||
>>> fetch_wetland_data(40.7128, -74.0060) | ||
{...} # some wetland data in dictionary form | ||
>>> fetch_wetland_data(0, 0) | ||
None | ||
""" | ||
wetland = None | ||
wetland_layer_name = 'kartoza:nwm6_beta_v3_20230714' | ||
base_wms_layer = 'https://maps.kartoza.com/geoserver/wms' | ||
|
||
wms_data = get_feature_info_from_wms( | ||
base_wms_layer, | ||
wetland_layer_name, | ||
'EPSG:4326', | ||
latitude, | ||
longitude, | ||
800, 600 | ||
) | ||
|
||
if wms_data and 'features' in wms_data and wms_data['features']: | ||
feature_data = wms_data['features'][0] | ||
wetland = feature_data['properties'] | ||
|
||
return wetland | ||
|
||
|
||
class FetchWetland(LoginRequiredMixin, APIView): | ||
|
||
def get(self, request, *args): | ||
lat = float(request.GET.get('lat', '0')) | ||
lon = float(request.GET.get('lon', '0')) | ||
|
||
wetland = fetch_wetland_data( | ||
lat, | ||
lon | ||
) | ||
|
||
return Response(wetland) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters