### ===================================================================================================================
### Create and send email following the termination of analysis on the server
### ===================================================================================================================
# Copyright ©VIIA 2025
### ===================================================================================================================
### 1. Import modules
### ===================================================================================================================
# General imports
from __future__ import annotations
import base64
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Union, Dict, Optional
from jinja2 import Template
from pathlib import Path
from os import getenv
# References for functions and classes in the rhdhv_fem package
from rhdhv_fem.tools import fem_send_server_email
from rhdhv_fem.fem_tools import fem_move_file
from rhdhv_fem.analyses.analysis_log import NonlinearAnalysisLog
# References for functions and classes in the viiaPackage
if TYPE_CHECKING:
from viiapackage.viiaStatus import ViiaProject
### ===================================================================================================================
### 2. Functions to collect the content for the DIANA server email
### ===================================================================================================================
def _get_email_template(project: ViiaProject) -> str:
""" Helper function to get the template for the server email."""
# Read the template file for the server email from the VIIA project
with open(
project.viia_settings.project_specific_package_location /
'tools' / 'server_email' / 'server_email_template.htm') as file:
template_content = file.read()
return template_content
[docs]def viia_collect_contents_server_email(
project: ViiaProject, out_file: Path, viia_object_number: str, analysis_name: Optional[str] = None) \
-> Dict[str, str]:
"""
Function to collect the data from the convergence graph function for adding to the server email.
Input:
- project (obj): VIIA project object containing collections of fem objects and project variables.
- out_file (Path): Path of the DIANA out-file with the information for the convergence behaviour.
- viia_object_number (str): Number of the VIIA object, for example '1137A'.
- analysis_name (str): Optional input for the name of the analysis. If provided it will be added to the
title of the graph.
Output:
- Returns dictionary with information for the server email, which can be used to insert in the template.
"""
# Time stamp of the termination of the analysis (format: hour:minute:second, day-month-year)
time_reference = datetime.now().strftime('%H%M%S%d%m%Y')
# Get signal name
signal = out_file.parent.name
if not (signal[0].upper() == 'S' and signal[1:].isnumeric()):
signal = None
# Collect the analysis if provided as string
analysis = None
if analysis_name is not None:
analysis = project.viia_get(collection='analyses', name=analysis_name)
# Read the DIANA dat-file and collect the data for the convergence behaviour
logs = project.read_diana_outfile(file=out_file, analysis=analysis)
# Plot the convergence graph and get information about the total steps
image_file = project.viia_convergence_graph(
file=out_file, signal=signal, show=False, analysis=analysis_name)[0]
nl_logs = [log for log in logs if isinstance(log, NonlinearAnalysisLog)]
approximate_elapsed_time = [nl_log.elapsed_time for nl_log in nl_logs]
if None in approximate_elapsed_time:
approximate_elapsed_time = 'Unknown'
else:
approximate_elapsed_time = timedelta(seconds=sum(approximate_elapsed_time))
approximate_elapsed_time = \
(f"{'{:02d}'.format(approximate_elapsed_time.days)}D:"
f"{'{:02d}'.format(approximate_elapsed_time.seconds//3600)}H:"
f"{'{:02d}'.format((approximate_elapsed_time.seconds//60)%60)}M:"
f"{'{:02d}'.format(approximate_elapsed_time.seconds%60)}S")
# Collect the contents
context = {
'server_name': getenv('COMPUTERNAME'),
'analysis_name': analysis_name,
'viia_object_number': viia_object_number,
'folder_path': out_file.parent.as_posix(),
'signal': out_file.parent.name,
'total_calculated_steps': sum([len(log.convergence.steps) for log in nl_logs]),
'time_of_termination':
f"{time_reference[0:2]}:{time_reference[2:4]}:{time_reference[4:6]} on "
f"{time_reference[6:8]}-{time_reference[8:10]}-{time_reference[10:14]}",
'convergence_graph': None,
'approximate_elapsed_time': approximate_elapsed_time}
if logs[-1].elapsed_time is not None:
# If the elapsed time is available, analysis is considered to be completed
context.update(termination_status='Completed')
elif getattr(logs[-1], 'convergence') is None:
context.update(termination_status='Unknown')
elif logs[-1].convergence.diverged:
context.update(termination_status='Diverged')
else:
context.update(termination_status='Aborted')
# Convert convergence image
if image_file.exists:
with open(image_file, 'rb') as file:
binary_data = file.read()
# Return base 64 encoded image
context.update(convergence_graph=base64.b64encode(binary_data).decode('utf-8'))
return context
### ===================================================================================================================
### 3. Function to generate and send an email from the DIANA servers
### ===================================================================================================================
[docs]def viia_send_server_email(
viia_object_number: str, folder_path: Union[str, Path], recipient_email: str,
analysis_name: Optional[str] = None, total_steps: Optional[Union[int, str]] = 'Not specified',
object_part: str = 'Not specified', version_nr: int = 0, analysis_nr: str = 'A?', signal: str = 'S?') -> bool:
"""
This function creates the contents and sends the email from server, when the analysis has finished. For the contents
of the email the output file in the provided folder is read and extracts the analysis information.
Input:
- viia_object_number (str): The object number used in the VIIA project, for example '1485V'.
- folder_path (str or Path): The path object of the folder containing the output file as string or as instance
of Path.
- recipient_email (str): Email-address to which the email should be sent.
- analysis_name (str): Optional input for the name of the analysis. If provided it will be added to the
title of the graph.
- total_steps (int): Optional input for the total number of steps in the analysis. Default is None.
- object_part (str): In case of multiple object parts, select the correct part for this analysis. On MYVIIA you
can see which object parts are available. Default value is 'Gehele object', which is the name to be used if
the object is not split in separate parts on MYVIIA. There can be multiple object-parts for reruns, multiple
inspections and objects that are split for analyses (front house vs barn). Default is 'Not specified'.
- version_nr (int): The version number of the analysis. Default is 0.
- analysis_nr (str): The analysis number of the analysis. Default is 'A?'.
- signal (str): The signal of the analysis. Default is 'S?'.
Output:
- If possible an email will be sent to the recipient_email about the termination of the analysis.
- If the email is sent True will be returned, otherwise False.
"""
from viiapackage.viiaStatus import viia_create_project
if isinstance(folder_path, str):
folder_path = Path(folder_path)
# Create project
project = viia_create_project(project_name=f'test-{viia_object_number}_email')
# Set the work folder and analysis folder to save the convergence graph
project.current_analysis_folder = folder_path
# Get the output file from the workfolder
out_file = project.get_file(path=folder_path, suffix='.out')
# Collect the contents for the server email
context = viia_collect_contents_server_email(
project=project, out_file=out_file, viia_object_number=viia_object_number, analysis_name=analysis_name)
context['total_steps'] = total_steps
context['signal'] = signal
context['version_nr'] = version_nr
context['analysis_nr'] = analysis_nr
context['object_part'] = object_part
# Load the html template file and create Jinja template object
template = Template(_get_email_template(project=project))
# Move log file to folder_path if that is not yet the case
if project.logfile.parent != folder_path and project.logfile.exists():
project.logfile = fem_move_file(file_to_move=project.logfile, folder=folder_path)
# Render the template with the context
output_html = template.render(context)
# Create the subject
subject = \
(f"VIIA analysis on server {context['server_name']} is terminated for {viia_object_number} in folder "
f"{folder_path.name} ({analysis_nr}-{signal}-v{version_nr})")
# Create and send the email using functionality in the FEM-client
return fem_send_server_email(
folder_path=folder_path, recipient_email=recipient_email, subject=subject,
message=[{'text': output_html, 'subtype': 'html'}])
### ===================================================================================================================
### 4. End of script
### ===================================================================================================================