136 lines
6.0 KiB
Python
136 lines
6.0 KiB
Python
import datetime
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from marshmallow import ValidationError
|
|
|
|
def validate_time_exceeds_threshold(value:datetime.datetime, seconds:int=60, minutes:int=60, hours:int=24, days:int=30, months:int=12)->bool:
|
|
"""returns a boolean when the input value is very distant in the future. The parameters provide the threshold"""
|
|
# time difference in seconds. Positive: in the future, Negative: in the past
|
|
current_time = datetime.datetime.now()
|
|
time_ = (value-current_time).total_seconds()
|
|
threshold = seconds*minutes*hours*days*months
|
|
return time_>=threshold
|
|
|
|
def validate_time_is_in_future(value:datetime.datetime):
|
|
"""returns a boolean when the input value is in the future."""
|
|
current_time = datetime.datetime.now()
|
|
return value >= current_time
|
|
|
|
def validate_time_is_in_not_too_distant_future(raise_validation_error:bool, value:datetime.datetime, seconds:int=60, minutes:int=60, hours:int=24, days:int=30, months:int=12)->bool:
|
|
"""
|
|
combines two boolean operations. Returns True when both conditions are met.
|
|
a) value is in the future
|
|
b) value is not too distant (e.g., at max. 1 year in the future)
|
|
|
|
When the value is 'None', the validation will be skipped. A ValidationError is never issued, but the method returns 'False'.
|
|
|
|
options:
|
|
raise_validation_error: boolean. If set to True, this method issues a marshmallow.ValidationError, when the conditions fail.
|
|
"""
|
|
if value is None:
|
|
return False
|
|
|
|
is_in_future = validate_time_is_in_future(value)
|
|
is_too_distant = validate_time_exceeds_threshold(value, seconds, minutes, hours, days, months)
|
|
|
|
if raise_validation_error:
|
|
if not is_in_future:
|
|
raise ValidationError({"any_date":f"The provided value must be in the future. Current Time: {datetime.datetime.now()}, Value: {value}"})
|
|
|
|
if is_too_distant:
|
|
raise ValidationError({"any_date":f"The provided value is in the too distant future and exceeds a threshold for 'reasonable' entries. Found: {value}"})
|
|
|
|
return is_in_future & (not is_too_distant)
|
|
|
|
class TimeLogic():
|
|
def __init__(self):
|
|
return
|
|
|
|
def time_delta(self, src_time, tgt_time, unit:str="m"):
|
|
"""
|
|
in brief, this function measures tgt_time - src_time
|
|
|
|
if the tgt_time is in the future, it is a positive value (tgt_time > src_time)
|
|
if the tgt_time is in the past, it is a negative value (tgt_time < src_time)
|
|
|
|
returns the delta between tgt_time and src_time as a float of minutes (or the optionally provided unit)
|
|
|
|
options:
|
|
unit: str, which defaults to 'm' (minutes). 'h' (hours) or 's' (seconds) are also common units. Determines the unit of the output time delta
|
|
"""
|
|
# convert np.datetime64
|
|
if isinstance(src_time, pd.Timestamp):
|
|
src_time = src_time.to_datetime64()
|
|
|
|
if isinstance(tgt_time, pd.Timestamp):
|
|
tgt_time = tgt_time.to_datetime64()
|
|
|
|
if isinstance(src_time, datetime.datetime):
|
|
src_time = np.datetime64(src_time)
|
|
|
|
if isinstance(tgt_time, datetime.datetime):
|
|
tgt_time = np.datetime64(tgt_time)
|
|
|
|
delta = tgt_time - src_time
|
|
minute_delta = delta / np.timedelta64(1, unit)
|
|
return minute_delta
|
|
|
|
def time_delta_from_now_to_tgt(self, tgt_time, unit="m"):
|
|
return self.time_delta(datetime.datetime.now(), tgt_time=tgt_time, unit=unit)
|
|
|
|
def time_inbetween(self, query_time:datetime.datetime, start_time:datetime.datetime, end_time:datetime.datetime) -> bool:
|
|
"""
|
|
checks, whether the query time is inbetween the start & end time. Returns a bool to indicate that.
|
|
The function internally rounds seconds and microseconds to zero, so they are ignored in the evaluation.
|
|
|
|
Example:
|
|
a = datetime.datetime(2017, 5, 16, 8, 21, 10)
|
|
b = datetime.datetime(2017, 5, 17, 8, 21, 09)
|
|
c = datetime.datetime(2017, 5, 18, 8, 21, 10)
|
|
|
|
is b between a and c? -> yes. Returns True
|
|
is c between a and b? -> no. Returns False
|
|
|
|
returns bool
|
|
"""
|
|
assert isinstance(query_time, datetime.datetime)
|
|
assert isinstance(start_time, datetime.datetime)
|
|
assert isinstance(end_time, datetime.datetime)
|
|
query_time = query_time.replace(second=0, microsecond=0)
|
|
start_time = start_time.replace(second=0, microsecond=0)
|
|
end_time = end_time.replace(second=0, microsecond=0)
|
|
return start_time <= query_time <= end_time
|
|
|
|
def time_inbetween_absolute_delta(self, query_time:datetime.datetime, start_time:datetime.datetime, end_time:datetime.datetime) -> tuple:
|
|
"""
|
|
similarly to self.time_inbetween, this function compares a query_time with the provided start and end time.
|
|
however, this function instead returns timedelta objects, which show the difference towards start and end
|
|
|
|
this function applies abs() to return only absolute deviations. Thereby, -23 becomes +23
|
|
|
|
returns: tuple(absolute_start_delta, absolute_end_delta)
|
|
"""
|
|
return (abs(query_time-start_time), abs(query_time-end_time))
|
|
|
|
def compare_query_is_inbetween_list(self, query_time, list_of_other_times) -> list:
|
|
list_of_bools = [
|
|
self.time_inbetween(query_time, time_elem_begin, time_elem_end)
|
|
for (time_elem_begin, time_elem_end) in list_of_other_times
|
|
]
|
|
return list_of_bools
|
|
|
|
def query_time_any_inbetween(self, query_time, list_of_other_times):
|
|
"""
|
|
given a query_time element, the element will be compared to every element in a list, where each
|
|
element is a tuple of (start_time, end_time)
|
|
"""
|
|
if len(list_of_other_times)==0:
|
|
# the time is not inbetween, if the provided list is empty
|
|
return False
|
|
|
|
list_of_bools = self.compare_query_is_inbetween_list(query_time, list_of_other_times)
|
|
return np.any(list_of_bools), list_of_bools
|
|
|
|
|