Skip to content
Snippets Groups Projects
Commit 6b80145d authored by Brady James Garvin's avatar Brady James Garvin
Browse files

Switched to the agromonitoring.com web service.

parent 382c7e01
Branches main
No related tags found
No related merge requests found
API_KEY = '[replace this string with your API key]'
from kivy.app import App from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.label import Label from kivy.uix.label import Label
from kivy.properties import ListProperty
from kivy.logger import Logger from kivy.logger import Logger
from datetime import datetime
from json import dumps from json import dumps
from openmrs import RESTConnection from api_key import API_KEY
from rest import RESTConnection
UNL_LATITUDE = 40.81750
UNL_LONGITUDE = -96.701389
def to_human_readable_time(timestamp):
return datetime.utcfromtimestamp(timestamp).strftime('%A %I:%M %p')
def to_celsius(kelvin):
return kelvin - 273.15
def format_forecast_record(record):
try:
time = to_human_readable_time(record['dt'])
description = ', '.join(condition['description'] for condition in record['weather'])
temperature = to_celsius(record['main']['temp'])
humidity = record['main']['humidity']
feels_like = to_celsius(record['main']['feels_like'])
return \
f'{time}: {description}, {temperature:.1f} degrees, {humidity:.0f}% humidity, ' + \
f'feels like {feels_like:.1f} degrees'
except KeyError:
return '[invalid forecast record]'
class Record(Label):
pass
class Records(BoxLayout):
records = ListProperty([])
def rebuild(self):
self.clear_widgets()
for record in self.records:
self.add_widget(Record(text=record))
def on_records(self, _, __):
self.rebuild()
class RestApp(App): class RestApp(App):
def load_locations(self): records = ListProperty([])
self.root.ids.results.clear_widgets()
openmrs_connection = RESTConnection('localhost', 8080, 'admin', 'Admin123') def load_records(self):
openmrs_connection.send_request('location', None, None, self.on_locations_loaded, self.records = []
self.on_locations_not_loaded, self.on_locations_not_loaded) connection = RESTConnection('api.agromonitoring.com', 443, '/agro/1.0')
connection.send_request(
'weather/forecast',
{
'appid': API_KEY,
'lat': UNL_LATITUDE,
'lon': UNL_LONGITUDE,
},
None,
self.on_records_loaded,
self.on_records_not_loaded,
self.on_records_not_loaded
)
def on_locations_loaded(self, _, response): def on_records_loaded(self, _, response):
print(dumps(response, indent=4, sort_keys=True)) print(dumps(response, indent=4, sort_keys=True))
results_layout = self.root.ids.results self.records = [format_forecast_record(record) for record in response]
for result in response['results']:
results_layout.add_widget(Label(text=result['display']))
def on_locations_not_loaded(self, _, error): def on_records_not_loaded(self, _, error):
self.root.ids.results.add_widget(Label(text='[Failed to load locations]')) self.records = ['[Failed to load records]']
Logger.error(f'{self.__class__.__name__}: {error}') Logger.error(f'{self.__class__.__name__}: {error}')
......
BoxLayout: BoxLayout:
orientation: 'vertical' orientation: 'vertical'
BoxLayout: ScrollView:
id: results Records:
orientation: 'vertical' records: app.records
Button: Button:
text: 'Load Locations' text: 'Load Records'
on_press: app.load_locations() size_hint: (1.0, 0.25)
on_press: app.load_records()
<Records>:
orientation: 'vertical'
size_hint: (1.0, None)
height: sum(child.height for child in self.children)
<Record>:
text_size: self.size
halign: 'center'
from kivy.network.urlrequest import UrlRequest
import base64 import base64
import json import json
try:
from urllib import quote from kivy.network.urlrequest import UrlRequest
except ImportError:
from urllib.parse import quote from urllib.parse import quote
class RESTConnection: class RESTConnection:
def __init__(self, authority, port, username, password): def __init__(self, authority, port, root_path, username=None, password=None):
self.authority = authority self.authority = authority
self.port = port self.port = port
credentials = f'{username}:{password}' self.root_path = root_path
credentials = base64.standard_b64encode(credentials.encode('UTF8')).decode('UTF8')
self.headers = { self.headers = {
'Authorization': f'Basic {credentials}',
'Content-type': 'application/json', 'Content-type': 'application/json',
} }
if username is not None and password is not None:
credentials = base64.standard_b64encode(f'{username}:{password}'.encode('UTF8')).decode('UTF8')
self.headers['Authorization'] = f'Basic {credentials}'
def construct_url(self, resource, get_parameters=None): def construct_url(self, resource, get_parameters=None):
get_parameters = '&'.join(f'{quote(str(key))}={quote(str(value))}' for key, value in get_parameters.items()) \ parameter_string = '&'.join(f'{quote(str(key))}={quote(str(value))}' for key, value in get_parameters.items()) \
if get_parameters is not None else '' if get_parameters is not None else ''
return f'http://{self.authority}:{self.port}/openmrs/ws/rest/v1/{resource}?{get_parameters}' return f'https://{self.authority}:{self.port}{self.root_path}/{resource}?{parameter_string}'
def send_request_by_url(self, url, post_parameters, on_success, on_failure, on_error): def send_request_by_url(self, url, post_parameters, on_success, on_failure, on_error):
UrlRequest(url, req_headers=self.headers, UrlRequest(url, req_headers=self.headers,
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment