Added comments

parent ad67fc7b
......@@ -12,26 +12,34 @@ from .parameters import WINDOW_WIDTH, START_TIME, END_TIME, DELTA_SAMPLE, DATA_W
class Presence():
def __init__(self):
# On start the cache is not valid
self._cache_expiration = datetime(year=1970, month=1, day=1)
self._cache = None
def _get_raw_data_from_db(self) -> List[List[List[Tuple[time, bool]]]]:
"""Return a list of day, each day is a list of week, each week is a list of tuples of timestamp and status"""
# Calculate the window for the data
window_date = (
datetime.combine(, time()) - WINDOW_WIDTH,
datetime.combine(, time()))
# Prepare the data struct
data = [[], [], [], [], []]
# Open a db session
with session_scope() as session:
# For each day of the week
for i in range(7):
# Backwards from today parse the days
current_day = window_date[1] - timedelta(days=(1 + i))
# Skip saturday and sunday
# Skip saturday and sunday (the data are useless)
if (current_day.weekday() == 5 or current_day.weekday() == 6): continue
# For each day of the week in the window
while current_day > window_date[0]:
# Calculate the limit of the day
current_limit = (
datetime.combine(current_day, START_TIME),
datetime.combine(current_day, END_TIME),
......@@ -39,18 +47,23 @@ class Presence():
day = []
# Extract the data from the db
for row in session.query(Status) \
day.append((row.timestamp.time(), True if row.value == row.OPEN else False))
# Add the data for the current week and the current day of the week
# Go to the previous week
current_day = current_day - timedelta(weeks=1)
return data
def get_data(self) -> List[List[List[int]]]:
# Prepare the final structure for the data
"""Provides the data on presence distribute over weeks"""
# Prepare the structure for the data
data = [[], [], [], [], []]
# Get the raw data from the db
......@@ -82,19 +95,28 @@ class Presence():
return data
def get_prevision(self, with_average=False) -> Union[List[List[float]], Tuple[List[List[float]], List[float]]]:
"""Provides prevision data on the HQ's status distribute over hours"""
# Get the data for the weeks
raw_data = self.get_data()
# Prepare the data struct
data = [[], [], [], [], []]
# For each day of the week
for i in range(5):
# Get the data of the current day
day_data = raw_data[i]
# Calculate the weighted average over the data weeks
for j, _ in enumerate(day_data):
data[i].append(numpy.average(day_data[j], weights=DATA_WEIGHTS[:len(day_data[j])]))
# If it's required a average for the days it's provided
if with_average:
average = []
# Calculate the average for the days
for i in range(5):
......@@ -103,8 +125,11 @@ class Presence():
return data
def _status(self, data: List[Tuple[time, bool]], hour: time) -> bool:
"""Provided data raw data for a single day and a moment returns True if the HQ was opening in that moment"""
status = False
# Check the status until the data are ended or a status following the moment required
for i in [(time(hour=0), False), *data]:
if i[0] > hour:
return status
......@@ -114,10 +139,14 @@ class Presence():
return status
def get_graph(self) -> PresenceGraph:
"""Returns the graph of the presence"""
# Check if the cache is expired
if self._cache_expiration <
# Generate a new graph and save it in the cache
self._cache = PresenceGraph(*self.get_prevision(True))
# Update the cache expiration time
self._cache_expiration = + GRAPH_CACHE_DURATION
return self._cache
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment