dschick/BreCal
dschick/BreCal
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

preparing notifications and refactored the legacy object

Browse Source
This commit is contained in:
Max Metz 2024-05-29 11:25:58 +02:00
parent b4c105946a
commit 8732a4a13a
2 changed files with 64 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
src/server/BreCal/notifications/notification_content_email.py Normal file
View File

@ -0,0 +1,53 @@
import pandas as pd
from server.BreCal.database.enums import ParticipantType, ShipcallType, StatusFlags
#### Verbosity Functions ####
def get_default_header()->str:
# HEADER (greeting and default message)
header = "Dear Sir or Madam\n\nThank you for participating in the project 'Bremen Calling'. During analysis, our software has identified an event, which may be worth a second look. Here is the summary. \n\n"
return header
def get_default_footer()->str:
# FOOTER (signature)
footer = "\n\nWe would kindly ask you to have a look at the shipcall and verify, if any action is required from your side. \n\nKind regards\nThe 'Bremen Calling' Team"
return footer
def get_agency_name(sql_handler, times_df):
times_agency = times_df.loc[times_df["participant_type"]==ParticipantType.AGENCY.value,"participant_id"]
if len(times_agency)==0:
agency_name = ""
else:
agency_participant_id = times_agency.iloc[0]
agency_name = sql_handler.df_dict.get("participant").loc[agency_participant_id,"name"]
return agency_name
def get_ship_name(sql_handler, shipcall):
ship = sql_handler.df_dict.get("ship").loc[shipcall.ship_id]
ship_name = ship.loc["name"] # when calling ship.name, the ID is returned (pandas syntax)
return ship_name
def create_notification_body(sql_handler, times_df, shipcall, result)->str:
# #TODO: add 'Link zum Anlauf'
# URL: https://trello.com/c/qenZyJxR/75-als-bsmd-m%C3%B6chte-ich-%C3%BCber-gelbe-und-rote-ampeln-informiert-werden-um-die-systembeteiligung-zu-st%C3%A4rken
header = get_default_header()
footer = get_default_footer()
agency_name = get_agency_name(sql_handler, times_df)
ship_name = get_ship_name(sql_handler, shipcall)
verbosity_introduction = f"Respective Shipcall:\n"
traffic_state_verbosity = f"\tTraffic Light State: {StatusFlags(result[0]).name}\n"
ship_name_verbosity = f"\tShip: {ship_name} (the ship is {ShipcallType(shipcall.type).name.lower()})\n"
agency_name_verbosity = f"\tResponsible Agency: {agency_name}\n"
eta_verbosity = f"\tEstimated Arrival Time: {shipcall.eta.isoformat()}\n" if not pd.isna(shipcall.eta) else ""
etd_verbosity = f"\tEstimated Departure Time: {shipcall.etd.isoformat()}\n" if not pd.isna(shipcall.etd) else ""
error_verbosity = f"\nError Description:\n\t" + "\n\t".join(result[1])
message_body = "".join([header, verbosity_introduction, traffic_state_verbosity, ship_name_verbosity, agency_name_verbosity, eta_verbosity, etd_verbosity, error_verbosity, footer])
return message_body

67
src/server/BreCal/notifications/notification_functions.py
View File

@ -1,7 +1,6 @@
import datetime
import pandas as pd
from BreCal.schemas.model import Notification
from BreCal.database.enums import NotificationType, ParticipantType, ShipcallType, StatusFlags
from BreCal.database.enums import NotificationType, StatusFlags
def create_notification(id, times_id, message, level, notification_type:NotificationType, created=None, modified=None):
created = (datetime.datetime.now()).isoformat() or created
@ -12,60 +11,12 @@ def create_notification(id, times_id, message, level, notification_type:Notifica
)
return notification
class NotifierFunctions():
"""
# #TODO_refactor: this legacy object needs strong refactoring and shall be simplified in its execution.
#### Verbosity Functions ####
def get_default_header()->str:
# HEADER (greeting and default message)
header = "Dear Sir or Madam\n\nThank you for participating in the project 'Bremen Calling'. During analysis, our software has identified an event, which may be worth a second look. Here is the summary. \n\n"
return header
def get_default_footer()->str:
# FOOTER (signature)
footer = "\n\nWe would kindly ask you to have a look at the shipcall and verify, if any action is required from your side. \n\nKind regards\nThe 'Bremen Calling' Team"
return footer
def get_agency_name(sql_handler, times_df):
times_agency = times_df.loc[times_df["participant_type"]==ParticipantType.AGENCY.value,"participant_id"]
if len(times_agency)==0:
agency_name = ""
else:
agency_participant_id = times_agency.iloc[0]
agency_name = sql_handler.df_dict.get("participant").loc[agency_participant_id,"name"]
return agency_name
def get_ship_name(sql_handler, shipcall):
ship = sql_handler.df_dict.get("ship").loc[shipcall.ship_id]
ship_name = ship.loc["name"] # when calling ship.name, the ID is returned (pandas syntax)
return ship_name
def create_notification_body(sql_handler, times_df, shipcall, result)->str:
# #TODO: add 'Link zum Anlauf'
# URL: https://trello.com/c/qenZyJxR/75-als-bsmd-m%C3%B6chte-ich-%C3%BCber-gelbe-und-rote-ampeln-informiert-werden-um-die-systembeteiligung-zu-st%C3%A4rken
header = get_default_header()
footer = get_default_footer()
agency_name = get_agency_name(sql_handler, times_df)
ship_name = get_ship_name(sql_handler, shipcall)
verbosity_introduction = f"Respective Shipcall:\n"
traffic_state_verbosity = f"\tTraffic Light State: {StatusFlags(result[0]).name}\n"
ship_name_verbosity = f"\tShip: {ship_name} (the ship is {ShipcallType(shipcall.type).name.lower()})\n"
agency_name_verbosity = f"\tResponsible Agency: {agency_name}\n"
eta_verbosity = f"\tEstimated Arrival Time: {shipcall.eta.isoformat()}\n" if not pd.isna(shipcall.eta) else ""
etd_verbosity = f"\tEstimated Departure Time: {shipcall.etd.isoformat()}\n" if not pd.isna(shipcall.etd) else ""
error_verbosity = f"\nError Description:\n\t" + "\n\t".join(result[1])
message_body = "".join([header, verbosity_introduction, traffic_state_verbosity, ship_name_verbosity, agency_name_verbosity, eta_verbosity, etd_verbosity, error_verbosity, footer])
return message_body
class Notifier():
"""An object that helps with the logic of selecting eligible shipcalls to create the correct notifications for the respective users."""
An object that helps with the logic of selecting eligible shipcalls to create the correct notifications for the respective users.
"""
def __init__(self)->None:
pass
@ -103,7 +54,7 @@ class Notifier():
state_old = max(int(state_old), StatusFlags.GREEN.value)
return int(state_new) > int(state_old)
def get_notification_times(self, evaluation_states_new)->list[datetime.datetime]:
def get_notification_times(self, evaluation_states_new)->list[str]:
"""# build the list of evaluation times ('now', as isoformat)"""
evaluation_times = [datetime.datetime.now().isoformat() for _i in range(len(evaluation_states_new))]
return evaluation_times
@ -113,3 +64,7 @@ class Notifier():
evaluation_notifications_sent = [self.determine_notification_state(state_old=int(state_old), state_new=int(state_new)) for state_old, state_new in zip(evaluation_states_old, evaluation_states_new)]
return evaluation_notifications_sent
class Notifier(NotifierFunctions):
def __init__(self):
super().__init__()