Skip to content

Commit

Permalink
Connection & Time to reach healthy distribution (#48)
Browse files Browse the repository at this point in the history
* Create class decorator for check if path exists.

* Create plot function to use for analysis of reach Nº of connections
  • Loading branch information
AlbertoSoutullo authored Jan 21, 2025
1 parent 8f47c8f commit 176a610
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 1 deletion.
25 changes: 24 additions & 1 deletion src/mesh_analysis/waku_message_log_analyzer.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,10 @@
from src.mesh_analysis.tracers.waku_tracer import WakuTracer
from src.plotting.utils import add_boxplot_stat_labels
from src.utils import file_utils, log_utils, path_utils, list_utils
from src.utils.path_utils import check_path_exists

logger = logging.getLogger(__name__)

sns.set_theme()

class WakuMessageLogAnalyzer:
def __init__(self, stateful_sets: List[str], timestamp_to_analyze: str = None,
Expand Down Expand Up @@ -381,3 +382,25 @@ def plot_message_distribution(self, received_summary_path: Path, plot_title: str
plt.show()

return Ok(None)


@check_path_exists
def check_time_to_reach_value_plot(self, file_data_path: Path, threshold_value: int, value_name: str,
dump_path: Path) -> Result[None, str]:
df = pd.read_csv(file_data_path)
df['Time'] = pd.to_datetime(df['Time'])
df.set_index('Time', inplace=True)

mask = df >= threshold_value
first_reach = mask.idxmax()
first_reach[~mask.any()] = pd.NaT # Set to NaT if the target is never reached
time_to_target = (first_reach - df.index[0]).dt.total_seconds() / 60

plt.figure(figsize=(10, 6))
plt.boxplot(time_to_target.dropna(), vert=False)
plt.title(f'Time to Reach {value_name} - {time_to_target.dropna().shape[0]}/{df.shape[1]}')
plt.xlabel('Time to Reach Target (minutes)')
plt.savefig(dump_path)
plt.show()

return Ok(None)
12 changes: 12 additions & 0 deletions src/utils/path_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,3 +36,15 @@ def prepare_path_for_folder(folder_location: Union[str, Path]) -> Result[Path, s
return Err(f'Error creating {folder_location.parent}. {e}')

return Ok(folder_location)


def check_path_exists(func):
def wrapper(self, path: Path, *args, **kwargs):
if not path.exists():
error = f'Path {args[0]} does not exist'
logger.error(error)
return Err(error)
return func(self, path, *args, **kwargs)

return wrapper

0 comments on commit 176a610

Please sign in to comment.