Commit 269d0d7b authored by Kate Patterson's avatar Kate Patterson
Browse files

Merge branch 'update_app_common' into 'master'

update app_common: 1. added some comment 2.added few time format function 3....

See merge request ceotr/app_common!9
parents 80f0d5ef 60ff4395
from .slack_message import SlackMessenger
# PLease set up the Slack Messenger credential before use any function
sm = SlackMessenger()
del SlackMessenger
import json
from urllib import request
SLACK_MESSAGE_COLOR_GREEN = "good"
SLACK_MESSAGE_COLOR_RED = "danger"
SLACK_WEB_HOOK_BASE_URL = "https://hooks.slack.com/services/"
SUCCEEDED = "Succeeded"
FAILED = "Failed"
class SlackMessenger:
SUCCEEDED = SUCCEEDED
FAILED = FAILED
def __init__(self, slack_token=None, project_name=None, channel=None):
self.slack_token = slack_token
self.project_name = project_name
self.channel = channel
def _set_up_checking(self):
if any([not self.channel, not self.slack_token, not self.project_name]):
raise ValueError("Has to set up the slack message before sending message")
def set_up(self, slack_token, project_name, channel):
self.slack_token = slack_token
self.project_name = project_name
self.channel = channel
def send_success_message(self, job_name, detail):
self._set_up_checking()
self.send_message(job_name, detail)
def send_fail_message(self):
self._set_up_checking()
raise NotImplementedError
def send_message(self, job_name, detail, status=SUCCEEDED, title=None, text=None):
sending_dict = self.text_dict_maker(job_name, status, detail, title=title, text=text)
self._sending_message(sending_dict)
def _sending_message(self, text_dict):
post = text_dict
try:
json_data = json.dumps(post)
req = request.Request(SLACK_WEB_HOOK_BASE_URL + self.slack_token,
data=json_data.encode('ascii'),
headers={'Content-Type': 'application/json'})
resp = request.urlopen(req)
except Exception as em:
print("EXCEPTION: " + str(em))
def text_dict_maker(self, job_name, status, detail, title=None, text=None):
if status == SUCCEEDED:
color = SLACK_MESSAGE_COLOR_GREEN
elif status == FAILED:
color = SLACK_MESSAGE_COLOR_RED
else:
msg = "status code should be either {} or {}".format(SUCCEEDED, FAILED)
raise ValueError(msg)
sending_dict = {
"channel": self.channel,
"attachments": [
{
"fallback": "This message was sent from {} project".format(self.project_name),
"color": color,
"title": title,
"text": "",
"fields": [
{
"title": "Job Name",
"value": job_name,
"short": True
},
{
"title": "Project",
"value": self.project_name,
"short": True,
},
{
"title": "Status",
"value": status,
"short": True
},
{
"title": "Detail",
"value": detail,
"short": True
}
]
}
]
}
if title:
sending_dict["attachments"][0]["title"] = title
if text:
sending_dict["attachments"][0]["text"] = text
return sending_dict
from .gps_points_tools import Haversine, gps_point_range_check, simplified_float
from .time_format_functions import convert_to_erddap_time_format, time_zone_replace, time_to_str, str_to_timeobj, \
str_to_timestamp, timestamp_format_check, time_comparison
from .list_compare import list_common_element
from .dataset_id_generate_function import IDGenerator
......@@ -45,3 +45,18 @@ def gps_point_range_check(point):
if -90.0 < lat < 90.0 and -180.0 < lng < 180.0:
return True
return False
def simplified_float(number, keep_digit):
"""
A function which can round the float (it is faster than round())
"""
keep_digit = 10 ** keep_digit
keep_digit = keep_digit / 100
number = float(number)
if number:
number = number * keep_digit
number = int(number)
number = float(number) / keep_digit
return number
......@@ -41,3 +41,20 @@ def compare_list(list1, list2):
return True
else:
return False
def list_common_element(list1, list2, case_sensitive=False):
"""Given two lists and return a new list with common element of two list, the element comes from the first list
"""
new_list = []
for item1 in list1:
for item2 in list2:
if case_sensitive:
if item1 == item2:
new_list.append(item1)
else:
if str(item1).lower() == str(item2).lower():
new_list.append(item1)
return new_list
......@@ -26,52 +26,57 @@ def timeit(method):
return timed
def time_zone_replace(time):
if time:
t = type(time)
def time_zone_replace(time_str_or_time_object, time_zone='UTC'):
"""
This function can convert time str to datetime object and replace its time zone to make Django happen
enter None return None. default time zoom is UTC
"""
if time_str_or_time_object:
t = type(time_str_or_time_object)
if t is str:
et = datetime.datetime.strptime(time, "%Y-%m-%d %H:%M:%S")
et = datetime.datetime.strptime(time_str_or_time_object, "%Y-%m-%d %H:%M:%S")
else:
et = time
time = et.replace(tzinfo=timezone('UTC'))
return time
et = time_str_or_time_object
time_str_or_time_object = et.replace(tzinfo=timezone(time_zone))
return time_str_or_time_object
def time_to_str(time):
def time_to_str(time_str, time_str_format="%Y-%m-%d %H:%M:%S"):
"""Given time str return datetime object
Given None return None
Given datetime object return itself
other raise"""
res = None
if time:
t = type(time)
if time_str:
t = type(time_str)
if t is not str:
str_time = time.strftime("%Y-%m-%d %H:%M:%S")
str_time = time_str.strftime(time_str_format)
res = str_time
elif t == datetime:
res = time_str
else:
res = time
msg = "Input must be string or datetime object"
raise ValueError(msg)
return res
def str_to_timeobj(time_str, time_format=False):
if time_format:
time_f = time_format
else:
time_f = '%Y-%m-%d %H:%M:%S'
def str_to_timeobj(time_str, time_format='%Y-%m-%d %H:%M:%S'):
"""Convert time str to datetime object"""
try:
time = datetime.datetime.strptime(time_str, time_f)
time_obj = datetime.datetime.strptime(time_str, time_format)
except Exception as e:
if type(time_str) is datetime:
return time_str
raise ValueError
msg = "Input must be string or datetime object"
raise ValueError(msg)
return time
return time_obj
def str_to_timestamp(time_str, time_format=False):
try:
time_obj = str_to_timeobj(time_str, time_format=time_format)
time_stamp = time_obj.timestamp()
except ValueError as e:
print("There is the error {}".format(e))
time_stamp = time_str
def str_to_timestamp(time_str, time_format='%Y-%m-%d %H:%M:%S'):
"""Convert time string to Unix timestamp"""
time_obj = str_to_timeobj(time_str, time_format=time_format)
time_stamp = time_obj.timestamp()
return time_stamp
......@@ -97,3 +102,21 @@ def time_comparison(first_time, second_time):
if first_time_obj < second_time_obj:
# first time is earlier than second time
return -1
def convert_to_erddap_time_format(time_str):
"""
accept time format like:
2019-03-12 14:55:02
convert it to
2019-03-12T14:55:02Z
if input was None, return None
"""
ret_time_str = None
if time_str and type(time_str) is str:
ret_time_str = time_str.lstrip(' ')
ret_time_str = ret_time_str.rstrip(' ')
ret_time_str = ret_time_str.replace(" ", 'T')
ret_time_str = ret_time_str + 'Z'
return ret_time_str
from unittest import TestCase
from app_common.utilities import simplified_float
class TestGPSPointTools(TestCase):
def test_gps_point_tool(self):
test_point = 42.113
res = simplified_float(test_point, 3)
print(res)
import unittest
from app_common.utilities import convert_to_erddap_time_format, time_zone_replace, str_to_timeobj, time_to_str
class TimeFormatFunctionTest(unittest.TestCase):
def setUp(self):
self.time1 = "2019-03-12 14:55:02"
self.errdap_time1 = "2019-03-12T14:55:02Z"
def test_convert_to_erddap_time_format(self):
self.assertEqual(convert_to_erddap_time_format(self.time1), self.errdap_time1)
def test_time_zone_replace(self):
time_1 = time_zone_replace(self.time1)
self.assertEqual(time_1.tzinfo.zone, 'UTC')
def test_time_str_convert_function(self):
time_obj = str_to_timeobj(self.time1)
time_str = time_to_str(time_obj)
self.assertEqual(time_str, self.time1)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment