pipeline.src.flows.distribute_pnos

Functions

extract_species_names(→ dict)

Returns dict with species code as key and species name as value

extract_fishing_gear_names(→ dict)

Returns dict with fishing gear code as key and fishing gear name as value

extract_pnos_to_generate(→ Tuple[pandas.DataFrame, bool])

extract_facade_email_addresses(→ dict)

extract_pno_extra_subscriptions(→ pandas.DataFrame)

make_prior_notification_attachments_query(...)

execute_prior_notification_attachments_query(→ dict)

to_pnos_to_render(→ List[src.entities.pnos.PnoToRender])

pre_render_pno(→ src.entities.pnos.PreRenderedPno)

get_html_for_pdf_template(→ jinja2.Template)

get_email_body_template(→ jinja2.Template)

get_sms_template(→ jinja2.Template)

render_pno(→ src.entities.pnos.RenderedPno)

load_pno_pdf_documents(...)

Loads input pno_pdf_documents to prior_notification_pdf_documents and returns

attribute_addressees(→ src.entities.pnos.RenderedPno)

Returns a copy of the input RenderedPno's with its control_units, and

create_email(→ src.entities.pnos.PnoToSend)

create_sms(→ src.entities.pnos.PnoToSend)

send_pno_message(...)

load_prior_notification_sent_messages(...)

make_update_logbook_reports_statement(...)

Creates slqalchemy update statement to update isBeingSent and isSent fields of

make_manual_prior_notifications_statement(...)

Creates slqalchemy update statement to update isBeingSent and isSent fields of

distribute_pnos_flow(test_mode, is_integration, ...)

is_prior_notification_zero(→ bool)

Module Contents

pipeline.src.flows.distribute_pnos.extract_species_names() → dict[source]

Returns dict with species code as key and species name as value

pipeline.src.flows.distribute_pnos.extract_fishing_gear_names() → dict[source]

Returns dict with fishing gear code as key and fishing gear name as value

pipeline.src.flows.distribute_pnos.extract_pnos_to_generate(start_datetime_utc: datetime.datetime, end_datetime_utc: datetime.datetime) → Tuple[pandas.DataFrame, bool][source]
pipeline.src.flows.distribute_pnos.extract_facade_email_addresses() → dict[source]
pipeline.src.flows.distribute_pnos.extract_pno_extra_subscriptions() → pandas.DataFrame[source]
pipeline.src.flows.distribute_pnos.make_prior_notification_attachments_query(pnos: List[src.entities.pnos.RenderedPno], prior_notification_uploads_table: sqlalchemy.Table) → sqlalchemy.Select | None[source]
pipeline.src.flows.distribute_pnos.execute_prior_notification_attachments_query(query: sqlalchemy.Select | None) → dict[source]
pipeline.src.flows.distribute_pnos.to_pnos_to_render(pnos: pandas.DataFrame) → List[src.entities.pnos.PnoToRender][source]
pipeline.src.flows.distribute_pnos.pre_render_pno(pno: src.entities.pnos.PnoToRender, species_names: dict, fishing_gear_names: dict) → src.entities.pnos.PreRenderedPno[source]
pipeline.src.flows.distribute_pnos.get_html_for_pdf_template() → jinja2.Template[source]
pipeline.src.flows.distribute_pnos.get_email_body_template() → jinja2.Template[source]
pipeline.src.flows.distribute_pnos.get_sms_template() → jinja2.Template[source]
pipeline.src.flows.distribute_pnos.render_pno(pno: src.entities.pnos.PreRenderedPno, html_for_pdf_template: jinja2.Template, email_body_template: jinja2.Template, sms_template: jinja2.Template) → src.entities.pnos.RenderedPno[source]
pipeline.src.flows.distribute_pnos.load_pno_pdf_documents(pno_pdf_documents: List[src.entities.pnos.RenderedPno]) → Tuple[List[src.entities.pnos.RenderedPno], bool][source]

Loads input pno_pdf_documents to prior_notification_pdf_documents and returns the subset of the input that must be distributed, i.e. the list of RenderedPno on which is_being_sent is True.

Parameters:

pno_pdf_documents (List[RenderedPno]) – RenderedPnos to load

Returns:

subset of input having is_being_sent equal to True.

Return type:

List[RenderedPno]

pipeline.src.flows.distribute_pnos.attribute_addressees(pno_to_distribute: src.entities.pnos.RenderedPno, units_targeting_vessels: pandas.DataFrame, units_ports_and_segments_subscriptions: pandas.DataFrame, pno_extra_subscriptions: pandas.DataFrame, control_units: List[src.entities.control_units.ControlUnit]) → src.entities.pnos.RenderedPno[source]

Returns a copy of the input RenderedPno’s with its control_units, and addtionnal_addressees attributes updated, representing control units and other addressees that should receive the PNO. The control units attributed to the PNO are :

  • control units who target the vessel

  • control units who subscribed to the port with the “receive all pnos” option

  • Plus : - control units who subscribed to the port, if the PNO is in

    verification scope

    • control units who subscribed to the port AND to a segment of the PNO if the PNO is not in verification scope

The addtionnal addressees attributed to the PNO are those in pno_extra_subscriptions who subscribed to the port and type of the PNO.

Parameters:

  • pno_to_distribute (RenderedPno) – RenderedPno

  • units_targeting_vessels (pd.DataFrame) – DataFrame with columns control_unit_ids_targeting_vessel and vessel_id

  • units_ports_and_segments_subscriptions (pd.DataFrame) – DataFrame with columns control_unit_id, port_locode, receive_all_pnos_from_port, and unit_subscribed_segments facade email addresses as values. The email address of the corresponding facade will be added to addressees in PNO emails.

  • pno_extra_subscriptions (pd.DataFrame) – DataFrame with columns pno_type_name, port_locode, recipient_name, recipient_organization, communication_means and recipient_email_address_or_number.

  • control_units (List[ControlUnit]) – List of all active ControlUnits

Returns:

copy of the input pno_to_distribute with addressees added

Return type:

RenderedPno

pipeline.src.flows.distribute_pnos.create_email(pno: src.entities.pnos.RenderedPno, uploaded_attachments: dict, facade_email_addresses: dict, test_mode: bool) → src.entities.pnos.PnoToSend[source]
pipeline.src.flows.distribute_pnos.create_sms(pno: src.entities.pnos.RenderedPno, test_mode: bool) → src.entities.pnos.PnoToSend[source]
pipeline.src.flows.distribute_pnos.send_pno_message(pno_to_send: src.entities.pnos.PnoToSend, is_integration: bool) → List[src.entities.pnos.PriorNotificationSentMessage][source]
pipeline.src.flows.distribute_pnos.load_prior_notification_sent_messages(prior_notification_sent_messages: List[src.entities.pnos.PriorNotificationSentMessage])[source]
pipeline.src.flows.distribute_pnos.make_update_logbook_reports_statement(pnos_to_update: List[src.entities.pnos.RenderedPno], sent_messages: List[src.entities.pnos.PriorNotificationSentMessage], start_datetime_utc: datetime.datetime, end_datetime_utc: datetime.datetime) → sqlalchemy.Executable | None[source]

Creates slqalchemy update statement to update isBeingSent and isSent fields of PNO of source LOGBOOK in table logbook_reports.

Parameters:

  • pnos_to_update (List[RenderedPno]) – PNOs to update

  • sent_messages (List[PriorNotificationSentMessage]) – PNOs that were sent

  • start_datetime_utc (datetime) – start date

  • end_datetime_utc (datetime) – end date

Raises:

prefect.engine.signals.SKIP – If no PNOs with source = ‘LOGBOOK’ is in the input list

Returns:

SQLAlchemy Update Statement

Return type:

Executable

pipeline.src.flows.distribute_pnos.make_manual_prior_notifications_statement(pnos_to_update: List[src.entities.pnos.RenderedPno], sent_messages: List[src.entities.pnos.PriorNotificationSentMessage]) → sqlalchemy.Executable | None[source]

Creates slqalchemy update statement to update isBeingSent and isSent fields of PNO of source MANUAL in table manual_prior_notifications.

Parameters:

pnos_to_update (List[RenderedPno]) – PNOs to update

Raises:

prefect.engine.signals.SKIP – If no PNOs with source = ‘MANUAL’ is in the input list

Returns:

SQLAlchemy Update Statement

Return type:

Executable

pipeline.src.flows.distribute_pnos.distribute_pnos_flow(test_mode: bool, is_integration: bool, start_hours_ago: int | float, end_hours_ago: int | float)[source]
pipeline.src.flows.distribute_pnos.is_prior_notification_zero(pno: src.entities.pnos.PnoToRender) → bool[source]