src.fairreckitlib.core.io.event_io

  1"""This module contains all event ids, event args and a print switch that are IO related.
  2
  3Constants:
  4
  5    ON_MAKE_DIR: id of the event that is used when a directory is created.
  6    ON_REMOVE_DIR: id of the event that is used when a directory is removed.
  7    ON_CREATE_FILE: id of the event that is used when a file is created.
  8    ON_REMOVE_FILE: id of the event that is used when a file is removed.
  9    ON_RENAME_FILE: id of the event that is used when a file is renamed.
 10
 11Classes:
 12
 13    DirEventArgs: event args related to a directory.
 14    FileEventArgs: event args related to a file.
 15    DataframeEventArgs: event args related to a dataframe.
 16    RenameFileEventArgs: event args related to renaming a file.
 17
 18Functions:
 19
 20    get_io_events: list of IO event IDs.
 21    get_io_event_print_switch: switch to print IO event arguments by ID.
 22    print_load_dataframe_event_args: print dataframe event arguments for loading.
 23
 24This program has been developed by students from the bachelor Computer Science at
 25Utrecht University within the Software Project course.
 26© Copyright Utrecht University (Department of Information and Computing Sciences)
 27"""
 28
 29from dataclasses import dataclass
 30from typing import Callable, Dict, List
 31
 32from ..events.event_dispatcher import EventArgs
 33
 34ON_MAKE_DIR = 'IO.on_make_dir'
 35ON_REMOVE_DIR = 'IO.on_remove_dir'
 36
 37ON_CREATE_FILE = 'IO.on_create_file'
 38ON_REMOVE_FILE = 'IO.on_remove_file'
 39
 40
 41@dataclass
 42class DirEventArgs(EventArgs):
 43    """Directory Event Arguments.
 44
 45    event_id: the unique ID that classifies the directory event.
 46    directory: the path to the directory.
 47    """
 48
 49    directory: str
 50
 51
 52@dataclass
 53class FileEventArgs(EventArgs):
 54    """File Event Arguments.
 55
 56    event_id: the unique ID that classifies the file event.
 57    file_path: the path to the file.
 58    """
 59
 60    file_path: str
 61
 62
 63@dataclass
 64class DataframeEventArgs(FileEventArgs):
 65    """Dataframe Event Arguments.
 66
 67    event_id: the unique ID that classifies the dataframe event.
 68    file_path: the path to the dataframe file.
 69    dataframe_name: the name of the dataframe.
 70    """
 71
 72    dataframe_name: str
 73
 74
 75def get_io_events() -> List[str]:
 76    """Get a list of IO event IDs.
 77
 78    Returns:
 79        a list of unique IO event IDs.
 80    """
 81    return [
 82        # DirEventArgs
 83        ON_MAKE_DIR,
 84        ON_REMOVE_DIR,
 85        # FileEventArgs
 86        ON_CREATE_FILE,
 87        ON_REMOVE_FILE,
 88    ]
 89
 90
 91def get_io_event_print_switch() -> Dict[str, Callable[[EventArgs], None]]:
 92    """Get a switch that prints IO event IDs.
 93
 94    Returns:
 95        the print IO event switch.
 96    """
 97    return {
 98        ON_MAKE_DIR:
 99            lambda args: print('Creating directory:', args.directory),
100        ON_REMOVE_DIR:
101            lambda args: print('Removing directory:', args.directory),
102        ON_CREATE_FILE:
103            lambda args: print('Creating file:', args.file_path),
104        ON_REMOVE_FILE:
105            lambda args: print('Removing file:', args.file_path),
106    }
107
108
109def print_load_df_event_args(event_args: DataframeEventArgs, elapsed_time: float=None)-> None:
110    """Print dataframe event arguments for loading.
111
112    It is assumed that the event started when elapsed_time is None and is finished otherwise.
113
114    Args:
115        event_args: the arguments to print.
116        elapsed_time: the time that has passed since the loading started, expressed in seconds.
117    """
118    if elapsed_time is None:
119        print('Loading', event_args.dataframe_name, 'from', event_args.file_path)
120    else:
121        print('Loaded', event_args.dataframe_name, f'in {elapsed_time:1.4f}s')