lip_pps_run_manager package

Module contents

class lip_pps_run_manager.RunManager(path_to_run_directory: Path, telegram_bot_name: Optional[str] = None, telegram_chat_name: Optional[str] = None, telegram_bot_token: Optional[str] = None, telegram_chat_id: Optional[str] = None, rate_limit: bool = True)[source]

Bases: object

Class to manage PPS Runs

This Class initializes the on disk structures if necessary.

When using a telegram bot, it is possible to use a configuration file to load the secret bot token or/and the unique chat id. Named strings are used to identify these quantities, see the example config file for the syntax. If the telegram_bot_name parameter is passed, the telegram_bot_token is ignored since it takes precedence. The telegram_chat_name also takes precedence over the telegram_chat_id. This helps keep these values secret when using the option to automatically back up scripts for each task.

A default config file should be placed in the home directory and have the name ‘.run_manager_telegram_config.json’. A specific config file should be placed in the current running directory and have the name ‘run_manager_telegram_config.json’

Example Telegram Bot Configuration File
 1{
 2    "bots": {
 3        "bot_name": "bot_token",
 4        "other_bot_name": "other_bot_token"
 5    },
 6    "chats": {
 7        "chat_name": "chat_id",
 8        "other_chat_name": "other_chat_id"
 9    }
10}
Parameters:

  • path_to_run_directory – The path to the directory where all the run related information is stored. Typically, there will be multiple processing steps/tasks applied to a single run and each will have its data stored in a single subdirectory.

  • telegram_bot_name – The bot name of the bot to be used to publish the report message to telegram. This property takes precedence over telegram_bot_token, which will be ignored if this parameter is passed. A config file should exist with the bot configuration, see the example.

  • telegram_chat_name – The chat name of the chat to publish the report messages to telegram. This property takes precedence over telegram_chat_id, which will be ignored if this parameter is passed. A config file should exist with the chat configuration, see the example.

  • telegram_bot_token – The telegram bot token to use (this value should be a secret, so do not share it) for the TelegramReporter, if any.

  • telegram_chat_id – The telegram chat ID the reporter should send messages to

Raises:

TypeError – If a parameter has the incorrect type

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
property backup_directory: Path

The backup directory property getter method

This method fetches the backup directory path attribute, which points to the path containing the backup data of this run.

Returns:

Path – The path to the directory where the backup data is stored.

backup_file(source: Path)[source]

Creates a backup of the file inside the run directory under the backup subdirectory.

Parameters:

source – The path to the source file. The file must exist

Raises:

  • TypeError – If the type of one of the parameters is not correct

  • RuntimeError – If the source does not exist.

Examples

>>> import lip_pps_run_manager as RM
>>> from pathlib import Path
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)
...   John.backup_file(Path("src.file"))

The above code should create a backup copy of the src.file in the backup directory of the run, if the src.file exists. If not, a RuntimeError will be raised.

copy_file_to(source: Path, destination: Path, overwrite: bool = False)[source]

Creates a copy of the source file to the destination.

Parameters:

  • source – The path to the source file. The file must exist

  • destination – The path to the destination. If the destination does not exist, the parent must exist and the file will be created. If the destination does exist, it must be a directory and a file with the same name as the source will be created. If the destination exists and is a file, action will only be taken if the overwrite flag is set.

  • overwrite – Whether to overwrite the destination file in case it already exists

Raises:

  • TypeError – If the type of one of the parameters is not correct

  • RuntimeError – If the source does not exist, if the destination does not exist and the parent directory does not exist or if the destination does exist and overwrite is not set.

  • SameFileError – If source and destination are the same file, a SameFileError will be raised.

Examples

>>> import lip_pps_run_manager as RM
>>> from pathlib import Path
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)
...   John.copy_file_to(Path("src.file"), Path("copy_of.file"))

The above code should create a copy of the src.file in the copy_of.file, if the src.file exists. If not, a RuntimeError will be raised.

create_run(raise_error: bool = False)[source]

Creates a run where this RunManager is pointing to.

Parameters:

raise_error – If True a RuntimeError is raised if the run already exists. If False no error is raised whether the run exists or not.

Raises:

  • RuntimeError – If the raise_error parameter is True and the run already exists or if the run directory already exists.

  • Warning – If any irregularity, during communication with telegram, it is reinterpreted as a warning

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)

The above code should create the Run0001 directory if it doesn’t exist or exit with a RuntimeError if it does.

property data_directory: Path

The data directory property getter method

This method fetches the data directory path attribute, which points to the path containing the run data of this run.

Returns:

Path – The path to the directory where the data is stored.

edit_message(message: str, message_id: str)[source]

Edit a message previously sent to telegram

Parameters:

  • message – The message to be sent

  • message_id – The message ID of the message to edit

Raises:

  • TypeError – If any parameter has the incorrect type

  • RuntimeError – If the telegram reporter is not configured

  • Warning – If any irregularity, during communication with telegram, it is reinterpreted as a warning

Returns:

message_id (str) – The telegram message id of the message which was just written

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001", telegram_bot_token="bot_token", telegram_chat_id="chat_id") as John:
...   John.create_run()
...   message_id = John.send_message("This is an example message")
...   John.edit_message("This is the edited message", message_id)
get_task_path(task_name: str) → Path[source]

Retrieve the Path of a given task

Parameters:

task_name – The name of the task of which to get the path

Raises:

TypeError – If any parameter has the incorrect type

Returns:

Path – The Path to the task directory.

Warning

The returned directory may or may not exist.

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   print("Processing task")
>>> John.get_task_directory("myTask")

The example above is using the with syntax, which will ensure the task directory is created.

handle_task(task_name: str, drop_old_data: bool = True, backup_python_file: bool = True, telegram_loop_iterations: Optional[int] = None, minimum_update_time_seconds: int = 60, minimum_warn_time_seconds: int = 60)[source]

Method that creates a handle to a manager for a specific task

The TaskManager that is created is under the current RunManager.

Parameters:

  • task_name – The name of the task which is going to be processed

  • drop_old_data – If a previous directory with the same name as this tasks exists, this flag controls whether that data is removed or not. Useful when testing and rerunning multiple times, in order to ensure that old data from previous runs is cleaned.

  • backup_python_file – If True a copy of the current python file will be backed up in the task directory. Useful for keeping a log of exactly what was done.

  • telegram_loop_iterations – If telegram reporting is enabled, this is the number of expected iterations in the processing loop. Use loop_tick() to keep track of progress during the loop.

  • minimum_update_time_seconds – The minimum time allowed between updates to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

  • minimum_warn_time_seconds – The minimum time allowed between warnings to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

Raises:

TypeError – If any parameter has the incorrect type

Returns:

TaskManager – The TaskManager to handle the task.

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001") as John:
...   John.create_run()
...   with John.handle_task("myTask") as taskHandler:
...     print("Processing task")

The above code should create the Run0001 directory and then create a subdirectory for the task “myTask”.

property path_directory: Path

The path directory property getter method

This method fetches the path_directory internal attribute, which contains the path to the directory containing the run information for this run.

Returns:

Path – The path to the directory where the run information is stored.

property run_name: str

The name of the run property getter method

send_message(message: str, reply_to_message_id: Optional[str] = None)[source]

Send a message to telegram

Parameters:

  • message – The message to be sent

  • reply_to_message_id – If the message is a reply to another message, place the message ID here

Raises:

  • TypeError – If any parameter has the incorrect type

  • RuntimeError – If the telegram reporter is not configured

  • Warning – If any irregularity, during communication with telegram, it is reinterpreted as a warning

Returns:

message_id (str) – The telegram message id of the message which was just written

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001", telegram_bot_token="bot_token", telegram_chat_id="chat_id") as John:
...   John.create_run()
...   John.send_message("This is an example message")
task_completed(task_name: str) → bool[source]

Check if a task has completed with success

Parameters:

task_name – The name of the task to check

Raises:

TypeError – If any parameter has the incorrect type

Returns:

bool – True if completed, False otherwise

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> John.task_completed("myTask")
task_ran_successfully(task_name: str) → bool[source]

Check if a task has ran with success

Parameters:

task_name – The name of the task to check

Raises:

TypeError – If any parameter has the incorrect type

Returns:

bool – True if successfull, False otherwise

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> John.task_ran_successfully("myTask")
class lip_pps_run_manager.SetupManager[source]

Bases: object

This class holds details about the experimental setup (particularly useful for device configuration)

class lip_pps_run_manager.TaskManager(path_to_run: Path, task_name: str, drop_old_data: bool = True, script_to_backup: Optional[Path] = None, telegram_bot_name: Optional[str] = None, telegram_chat_name: Optional[str] = None, telegram_bot_token: Optional[str] = None, telegram_chat_id: Optional[str] = None, loop_iterations: Optional[int] = None, minimum_update_time_seconds: int = 60, minimum_warn_time_seconds: int = 60, rate_limit: bool = True)[source]

Bases: RunManager

Class to manage PPS Tasks

This Class initializes the on disk structures if necessary.

Parameters:

  • path_to_run – The path to the directory where all the run related information is stored

  • task_name – The name of this task

  • drop_old_data – If a previous directory with the same name as this tasks exists, this flag controls whether that data is removed or not. Useful when testing and rerunning multiple times, in order to ensure that old data from previous runs is cleaned.

  • script_to_backup – Path to the script to be backed up to the task directory

  • telegram_bot_name – The bot name of the bot to be used to publish the report message to telegram. This property takes precedence over telegram_bot_token, which will be ignored if this parameter is passed. A config file should exist with the bot configuration, see the RunManager example.

  • telegram_chat_name – The chat name of the chat to publish the report messages to telegram. This property takes precedence over telegram_chat_id, which will be ignored if this parameter is passed. A config file should exist with the chat configuration, see the RunManager example.

  • telegram_bot_token – The telegram bot token to use (this value should be a secret, so do not share it) for the TelegramReporter, if any.

  • telegram_chat_id – The telegram chat ID the reporter should send messages to

  • loop_iterations – The number of iterations in the loop of the task being processed. This value is used mostly for the telegram reporting in order to report the current status of the task progress. Use loop_tick() to keep track of progress during the loop.

  • minimum_update_time_seconds – The minimum time allowed between updates to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

  • minimum_warn_time_seconds – The minimum time allowed between warnings to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

Raises:

  • TypeError – If the parameter has the incorrect type

  • RuntimeError – If the paths point to the wrong types (i.e. not a file for a file) If a directory which is not the directory of a run is passed

Examples

It is recommended to always use this class through the RunManager since it is necessary to have the underlying run directories and files created. This is what is done in the example below, where the handle_task method implicitly calls this class.

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001") as John:
...   John.create_run()
...   with John.handle_task("myTask") as taskHandler:
...     print("Process task here...")
backup_file(source: Path)[source]

Creates a backup of the source file inside the task directory.

Parameters:

source – The path to the source file. The file must exist

Raises:

  • TypeError – If the type of one of the parameters is not correct

  • RuntimeError – If the source does not exist.

Examples

>>> import lip_pps_run_manager as RM
>>> from pathlib import Path
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)
...   John.backup_file(Path("src.file"))

TODO: Fix this examples

The above code should create a backup copy of the src.file in the backup directory of the run, if the src.file exists. If not, a RuntimeError will be raised.

clean_task_directory()[source]

Clean directory of task of all previous data

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   (taskHandler.task_path/"testFile.tmp").touch()
...   taskHandler.clean_task_directory()
...   print((taskHandler.task_path/"testFile.tmp").is_file())
property expected_finish_time

The time at which the task is expected to be finished

loop_tick(count: int = 1)[source]

Increase the internal loop count, it is assumed this method is called at the end of the loop

When skipping multiple iterations, you can use the count parameter to increase the counter for multiple iterations. This method is also used internally to do some bookeeping operations such as keeping the status message on telegram up to date and sending warnings if there have been any. For this reason, it is good to call this method with some frequency. If the task loop is so long that the frequency of updates is deemed too slow, consider calling this method with count=0 or by calling explicitly the _update_status method, but keep in mind that _update_status keeps an eye on when the status was last updated, so if it is too soon, no update will be made.

Parameters:

count – The amount by which to increase the internal loop counter, by default, this parameter is set to 1

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   # Do work
...   taskHandler.loop_tick()
property processed_iterations: int

The processed iterations property getter method

set_completed()[source]

Set the task as if it had completed

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   # Do work
...   taskHandler.set_completed()
property task_name: str

The task name property getter method

property task_path: Path

The task path property getter method

warn(message: str)[source]

Send a warning to telegram

Send the warning as a reply to the original status message. If the warnings are sent with less than minimum_warn_time_seconds parameter, the messages are stored and sent later.

Parameters:

message – The warning message to be sent

Raises:

TypeError – If the parameter has the incorrect type

class lip_pps_run_manager.TelegramReporter(bot_token: str, chat_id: str, rate_limit: bool = True)[source]

Bases: object

Class to report to telegram

This class is used to send messages to telegram via the telegram bot API. For regular usage, the RunManager and TaskManager classes use this class automatically, as long as configured correctly. For fine-grained control, this class can be used on its own.

Parameters:

  • bot_token – The telegram bot token to use (this value should be a secret, so do not share it)

  • chat_id – The telegram chat ID the reporter should send messages to

Variables:

  • bot_token –

  • chat_id –

Raises:

TypeError – If a parameter has the incorrect type

Examples

>>> import lip_pps_run_manager as RM
>>> bot = RM.TelegramReporter("SecretBotToken", "PostToThisChat_ID")
property bot_token: str

The token of the telegram bot property getter method

property chat_id: str

The chat ID property getter method

edit_message(message_text: str, message_id: str)[source]

Edit a message that was previously sent to the chat using the bot.

Parameters:

  • message_text – The message the bot should change to message to

  • message_id – The ID of the message to edit

Raises:

  • TypeError – If a parameter has the wrong type

  • Warning – If any irregularity, leading to an exception occurs, it is reinterpreted as a warning

Examples

>>> import lip_pps_run_manager as RM
>>> bot = RM.TelegramReporter("SecretBotToken", "PostToThisChat_ID")
>>> bot.edit_message("New Message", "OldMessage_ID")
send_message(message_text: str, reply_to_message_id: Optional[str] = None)[source]

Send a message to the chat using the bot.

Parameters:

  • message_text – The message the bot should send to the chat

  • reply_to_message_id – If the message is in reply to another message, place the ID of the message being replied to here

Raises:

  • TypeError – If a parameter has the wrong type

  • Warning – If any irregularity, leading to an exception occurs, it is reinterpreted as a warning

Examples

>>> import lip_pps_run_manager as RM
>>> bot = RM.TelegramReporter("SecretBotToken", "PostToThisChat_ID")
>>> bot.send_message("Hello World!")

Submodules

lip_pps_run_manager.run_manager module

The Run Manager module

Contains classes and functions used to manage the runs and their tasks.

class lip_pps_run_manager.run_manager.RunManager(path_to_run_directory: Path, telegram_bot_name: Optional[str] = None, telegram_chat_name: Optional[str] = None, telegram_bot_token: Optional[str] = None, telegram_chat_id: Optional[str] = None, rate_limit: bool = True)[source]

Bases: object

Class to manage PPS Runs

This Class initializes the on disk structures if necessary.

When using a telegram bot, it is possible to use a configuration file to load the secret bot token or/and the unique chat id. Named strings are used to identify these quantities, see the example config file for the syntax. If the telegram_bot_name parameter is passed, the telegram_bot_token is ignored since it takes precedence. The telegram_chat_name also takes precedence over the telegram_chat_id. This helps keep these values secret when using the option to automatically back up scripts for each task.

A default config file should be placed in the home directory and have the name ‘.run_manager_telegram_config.json’. A specific config file should be placed in the current running directory and have the name ‘run_manager_telegram_config.json’

Example Telegram Bot Configuration File
 1{
 2    "bots": {
 3        "bot_name": "bot_token",
 4        "other_bot_name": "other_bot_token"
 5    },
 6    "chats": {
 7        "chat_name": "chat_id",
 8        "other_chat_name": "other_chat_id"
 9    }
10}
Parameters:

  • path_to_run_directory – The path to the directory where all the run related information is stored. Typically, there will be multiple processing steps/tasks applied to a single run and each will have its data stored in a single subdirectory.

  • telegram_bot_name – The bot name of the bot to be used to publish the report message to telegram. This property takes precedence over telegram_bot_token, which will be ignored if this parameter is passed. A config file should exist with the bot configuration, see the example.

  • telegram_chat_name – The chat name of the chat to publish the report messages to telegram. This property takes precedence over telegram_chat_id, which will be ignored if this parameter is passed. A config file should exist with the chat configuration, see the example.

  • telegram_bot_token – The telegram bot token to use (this value should be a secret, so do not share it) for the TelegramReporter, if any.

  • telegram_chat_id – The telegram chat ID the reporter should send messages to

Raises:

TypeError – If a parameter has the incorrect type

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
property backup_directory: Path

The backup directory property getter method

This method fetches the backup directory path attribute, which points to the path containing the backup data of this run.

Returns:

Path – The path to the directory where the backup data is stored.

backup_file(source: Path)[source]

Creates a backup of the file inside the run directory under the backup subdirectory.

Parameters:

source – The path to the source file. The file must exist

Raises:

  • TypeError – If the type of one of the parameters is not correct

  • RuntimeError – If the source does not exist.

Examples

>>> import lip_pps_run_manager as RM
>>> from pathlib import Path
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)
...   John.backup_file(Path("src.file"))

The above code should create a backup copy of the src.file in the backup directory of the run, if the src.file exists. If not, a RuntimeError will be raised.

copy_file_to(source: Path, destination: Path, overwrite: bool = False)[source]

Creates a copy of the source file to the destination.

Parameters:

  • source – The path to the source file. The file must exist

  • destination – The path to the destination. If the destination does not exist, the parent must exist and the file will be created. If the destination does exist, it must be a directory and a file with the same name as the source will be created. If the destination exists and is a file, action will only be taken if the overwrite flag is set.

  • overwrite – Whether to overwrite the destination file in case it already exists

Raises:

  • TypeError – If the type of one of the parameters is not correct

  • RuntimeError – If the source does not exist, if the destination does not exist and the parent directory does not exist or if the destination does exist and overwrite is not set.

  • SameFileError – If source and destination are the same file, a SameFileError will be raised.

Examples

>>> import lip_pps_run_manager as RM
>>> from pathlib import Path
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)
...   John.copy_file_to(Path("src.file"), Path("copy_of.file"))

The above code should create a copy of the src.file in the copy_of.file, if the src.file exists. If not, a RuntimeError will be raised.

create_run(raise_error: bool = False)[source]

Creates a run where this RunManager is pointing to.

Parameters:

raise_error – If True a RuntimeError is raised if the run already exists. If False no error is raised whether the run exists or not.

Raises:

  • RuntimeError – If the raise_error parameter is True and the run already exists or if the run directory already exists.

  • Warning – If any irregularity, during communication with telegram, it is reinterpreted as a warning

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)

The above code should create the Run0001 directory if it doesn’t exist or exit with a RuntimeError if it does.

property data_directory: Path

The data directory property getter method

This method fetches the data directory path attribute, which points to the path containing the run data of this run.

Returns:

Path – The path to the directory where the data is stored.

edit_message(message: str, message_id: str)[source]

Edit a message previously sent to telegram

Parameters:

  • message – The message to be sent

  • message_id – The message ID of the message to edit

Raises:

  • TypeError – If any parameter has the incorrect type

  • RuntimeError – If the telegram reporter is not configured

  • Warning – If any irregularity, during communication with telegram, it is reinterpreted as a warning

Returns:

message_id (str) – The telegram message id of the message which was just written

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001", telegram_bot_token="bot_token", telegram_chat_id="chat_id") as John:
...   John.create_run()
...   message_id = John.send_message("This is an example message")
...   John.edit_message("This is the edited message", message_id)
get_task_path(task_name: str) → Path[source]

Retrieve the Path of a given task

Parameters:

task_name – The name of the task of which to get the path

Raises:

TypeError – If any parameter has the incorrect type

Returns:

Path – The Path to the task directory.

Warning

The returned directory may or may not exist.

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   print("Processing task")
>>> John.get_task_directory("myTask")

The example above is using the with syntax, which will ensure the task directory is created.

handle_task(task_name: str, drop_old_data: bool = True, backup_python_file: bool = True, telegram_loop_iterations: Optional[int] = None, minimum_update_time_seconds: int = 60, minimum_warn_time_seconds: int = 60)[source]

Method that creates a handle to a manager for a specific task

The TaskManager that is created is under the current RunManager.

Parameters:

  • task_name – The name of the task which is going to be processed

  • drop_old_data – If a previous directory with the same name as this tasks exists, this flag controls whether that data is removed or not. Useful when testing and rerunning multiple times, in order to ensure that old data from previous runs is cleaned.

  • backup_python_file – If True a copy of the current python file will be backed up in the task directory. Useful for keeping a log of exactly what was done.

  • telegram_loop_iterations – If telegram reporting is enabled, this is the number of expected iterations in the processing loop. Use loop_tick() to keep track of progress during the loop.

  • minimum_update_time_seconds – The minimum time allowed between updates to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

  • minimum_warn_time_seconds – The minimum time allowed between warnings to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

Raises:

TypeError – If any parameter has the incorrect type

Returns:

TaskManager – The TaskManager to handle the task.

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001") as John:
...   John.create_run()
...   with John.handle_task("myTask") as taskHandler:
...     print("Processing task")

The above code should create the Run0001 directory and then create a subdirectory for the task “myTask”.

property path_directory: Path

The path directory property getter method

This method fetches the path_directory internal attribute, which contains the path to the directory containing the run information for this run.

Returns:

Path – The path to the directory where the run information is stored.

property run_name: str

The name of the run property getter method

send_message(message: str, reply_to_message_id: Optional[str] = None)[source]

Send a message to telegram

Parameters:

  • message – The message to be sent

  • reply_to_message_id – If the message is a reply to another message, place the message ID here

Raises:

  • TypeError – If any parameter has the incorrect type

  • RuntimeError – If the telegram reporter is not configured

  • Warning – If any irregularity, during communication with telegram, it is reinterpreted as a warning

Returns:

message_id (str) – The telegram message id of the message which was just written

Examples

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001", telegram_bot_token="bot_token", telegram_chat_id="chat_id") as John:
...   John.create_run()
...   John.send_message("This is an example message")
task_completed(task_name: str) → bool[source]

Check if a task has completed with success

Parameters:

task_name – The name of the task to check

Raises:

TypeError – If any parameter has the incorrect type

Returns:

bool – True if completed, False otherwise

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> John.task_completed("myTask")
task_ran_successfully(task_name: str) → bool[source]

Check if a task has ran with success

Parameters:

task_name – The name of the task to check

Raises:

TypeError – If any parameter has the incorrect type

Returns:

bool – True if successfull, False otherwise

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> John.task_ran_successfully("myTask")
class lip_pps_run_manager.run_manager.TaskManager(path_to_run: Path, task_name: str, drop_old_data: bool = True, script_to_backup: Optional[Path] = None, telegram_bot_name: Optional[str] = None, telegram_chat_name: Optional[str] = None, telegram_bot_token: Optional[str] = None, telegram_chat_id: Optional[str] = None, loop_iterations: Optional[int] = None, minimum_update_time_seconds: int = 60, minimum_warn_time_seconds: int = 60, rate_limit: bool = True)[source]

Bases: RunManager

Class to manage PPS Tasks

This Class initializes the on disk structures if necessary.

Parameters:

  • path_to_run – The path to the directory where all the run related information is stored

  • task_name – The name of this task

  • drop_old_data – If a previous directory with the same name as this tasks exists, this flag controls whether that data is removed or not. Useful when testing and rerunning multiple times, in order to ensure that old data from previous runs is cleaned.

  • script_to_backup – Path to the script to be backed up to the task directory

  • telegram_bot_name – The bot name of the bot to be used to publish the report message to telegram. This property takes precedence over telegram_bot_token, which will be ignored if this parameter is passed. A config file should exist with the bot configuration, see the RunManager example.

  • telegram_chat_name – The chat name of the chat to publish the report messages to telegram. This property takes precedence over telegram_chat_id, which will be ignored if this parameter is passed. A config file should exist with the chat configuration, see the RunManager example.

  • telegram_bot_token – The telegram bot token to use (this value should be a secret, so do not share it) for the TelegramReporter, if any.

  • telegram_chat_id – The telegram chat ID the reporter should send messages to

  • loop_iterations – The number of iterations in the loop of the task being processed. This value is used mostly for the telegram reporting in order to report the current status of the task progress. Use loop_tick() to keep track of progress during the loop.

  • minimum_update_time_seconds – The minimum time allowed between updates to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

  • minimum_warn_time_seconds – The minimum time allowed between warnings to telegram. This parameter is important in order to guarantee that the limits imposed by telegram are respected.

Raises:

  • TypeError – If the parameter has the incorrect type

  • RuntimeError – If the paths point to the wrong types (i.e. not a file for a file) If a directory which is not the directory of a run is passed

Examples

It is recommended to always use this class through the RunManager since it is necessary to have the underlying run directories and files created. This is what is done in the example below, where the handle_task method implicitly calls this class.

>>> import lip_pps_run_manager as RM
>>> with RM.RunManager("Run0001") as John:
...   John.create_run()
...   with John.handle_task("myTask") as taskHandler:
...     print("Process task here...")
backup_file(source: Path)[source]

Creates a backup of the source file inside the task directory.

Parameters:

source – The path to the source file. The file must exist

Raises:

  • TypeError – If the type of one of the parameters is not correct

  • RuntimeError – If the source does not exist.

Examples

>>> import lip_pps_run_manager as RM
>>> from pathlib import Path
>>> with RM.RunManager("Run0001") as John
...   John.create_run(True)
...   John.backup_file(Path("src.file"))

TODO: Fix this examples

The above code should create a backup copy of the src.file in the backup directory of the run, if the src.file exists. If not, a RuntimeError will be raised.

clean_task_directory()[source]

Clean directory of task of all previous data

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   (taskHandler.task_path/"testFile.tmp").touch()
...   taskHandler.clean_task_directory()
...   print((taskHandler.task_path/"testFile.tmp").is_file())
property expected_finish_time

The time at which the task is expected to be finished

loop_tick(count: int = 1)[source]

Increase the internal loop count, it is assumed this method is called at the end of the loop

When skipping multiple iterations, you can use the count parameter to increase the counter for multiple iterations. This method is also used internally to do some bookeeping operations such as keeping the status message on telegram up to date and sending warnings if there have been any. For this reason, it is good to call this method with some frequency. If the task loop is so long that the frequency of updates is deemed too slow, consider calling this method with count=0 or by calling explicitly the _update_status method, but keep in mind that _update_status keeps an eye on when the status was last updated, so if it is too soon, no update will be made.

Parameters:

count – The amount by which to increase the internal loop counter, by default, this parameter is set to 1

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   # Do work
...   taskHandler.loop_tick()
property processed_iterations: int

The processed iterations property getter method

set_completed()[source]

Set the task as if it had completed

Examples

>>> import lip_pps_run_manager as RM
>>> John = RM.RunManager("Run0001")
>>> John.create_run()
>>> with John.handle_task("myTask") as taskHandler:
...   # Do work
...   taskHandler.set_completed()
property task_name: str

The task name property getter method

property task_path: Path

The task path property getter method

warn(message: str)[source]

Send a warning to telegram

Send the warning as a reply to the original status message. If the warnings are sent with less than minimum_warn_time_seconds parameter, the messages are stored and sent later.

Parameters:

message – The warning message to be sent

Raises:

TypeError – If the parameter has the incorrect type

lip_pps_run_manager.run_manager.clean_path(path_to_clean: Path) → Path[source]

Clean a path from dangerous characters

Some characters are not recommended/supported by a given filesystem. To make matters worse, the set of supported characters varies from operating system to operating system. In order to make sure this code is portable and that things remain compatible, we choose a subset of characters on which to limit the paths used. The subset is essentially all letters (lower and upper case), all numbers augmented with the dot, underscore and dash.

Parameters:

path_to_clean – The path to the directory to clean

Raises:

TypeError – If the parameter has the wrong type

Returns:

Path – The path_to_clean path cleaned of all characters not part of the reduced set

Examples

>>> import lip_pps_run_manager.run_manager as RM
>>> print(RM.clean_path(Path("/tmp/2@#_run")))
lip_pps_run_manager.run_manager.create_run(path_to_directory: Path, run_name: str) → Path[source]

Create a new run in a given directory

Parameters:

  • path_to_directory – The path to the directory where to create the run

  • run_name – The name of the run to create

Raises:

  • TypeError – If either of the parameters has the wrong type

  • RuntimeError – If the run directory already exists

Returns:

Path – The path to the creaated run.

Examples

>>> import lip_pps_run_manager.run_manager as RM
>>> print(RM.create_run(Path("."), "Run0001"))
lip_pps_run_manager.run_manager.load_telegram_config() → <module 'json' from '/home/docs/.pyenv/versions/3.7.9/lib/python3.7/json/__init__.py'>[source]

Load the config file with the telegram configuration information.

The function searches for a config file in the working directory first with the name ‘run_manager_telegram_config.json’ and if it does not exist, it tries searching for a config file in the home directory with the name ‘.run_manager_telegram_config.json’

Returns:

json – The json representation of the config file

Examples

>>> import lip_pps_run_manager.run_manager as RM
>>> print(RM.create_run(Path("."), "Run0001"))
lip_pps_run_manager.run_manager.run_exists(path_to_directory: Path, run_name: str) → bool[source]

Check if a given run already exists in a given directory

Parameters:

  • path_to_directory – The path to the directory where to check if the run exists

  • run_name – The name of the run to check for

Raises:

TypeError – If either of the parameters has the wrong type

Returns:

bool – True if the run already exists, False if it does not exist.

Examples

>>> import lip_pps_run_manager.run_manager as RM
>>> print(RM.run_exists(Path("."), "Run0001"))

lip_pps_run_manager.telegram_reporter module

The Telegram Reporter module

Contains classes and functions used to manage bots to report things to telegram.

class lip_pps_run_manager.telegram_reporter.TelegramReporter(bot_token: str, chat_id: str, rate_limit: bool = True)[source]

Bases: object

Class to report to telegram

This class is used to send messages to telegram via the telegram bot API. For regular usage, the RunManager and TaskManager classes use this class automatically, as long as configured correctly. For fine-grained control, this class can be used on its own.

Parameters:

  • bot_token – The telegram bot token to use (this value should be a secret, so do not share it)

  • chat_id – The telegram chat ID the reporter should send messages to

Variables:

  • bot_token –

  • chat_id –

Raises:

TypeError – If a parameter has the incorrect type

Examples

>>> import lip_pps_run_manager as RM
>>> bot = RM.TelegramReporter("SecretBotToken", "PostToThisChat_ID")
property bot_token: str

The token of the telegram bot property getter method

property chat_id: str

The chat ID property getter method

edit_message(message_text: str, message_id: str)[source]

Edit a message that was previously sent to the chat using the bot.

Parameters:

  • message_text – The message the bot should change to message to

  • message_id – The ID of the message to edit

Raises:

  • TypeError – If a parameter has the wrong type

  • Warning – If any irregularity, leading to an exception occurs, it is reinterpreted as a warning

Examples

>>> import lip_pps_run_manager as RM
>>> bot = RM.TelegramReporter("SecretBotToken", "PostToThisChat_ID")
>>> bot.edit_message("New Message", "OldMessage_ID")
send_message(message_text: str, reply_to_message_id: Optional[str] = None)[source]

Send a message to the chat using the bot.

Parameters:

  • message_text – The message the bot should send to the chat

  • reply_to_message_id – If the message is in reply to another message, place the ID of the message being replied to here

Raises:

  • TypeError – If a parameter has the wrong type

  • Warning – If any irregularity, leading to an exception occurs, it is reinterpreted as a warning

Examples

>>> import lip_pps_run_manager as RM
>>> bot = RM.TelegramReporter("SecretBotToken", "PostToThisChat_ID")
>>> bot.send_message("Hello World!")

lip_pps_run_manager.setup_manager module

class lip_pps_run_manager.setup_manager.DeviceBase(device_name: str, device_type: str)[source]

Bases: object

This is the base class for implementing a device for an experimental setup

safe_shutdown()[source]
class lip_pps_run_manager.setup_manager.SetupManager[source]

Bases: object

This class holds details about the experimental setup (particularly useful for device configuration)

class lip_pps_run_manager.setup_manager.VISADevice(device_type: str, device_name: str, resource_string: str)[source]

Bases: DeviceBase

This is the base class for implementing a device for an experimental setup which communicates with the VISA interface

lip_pps_run_manager.cli module

Module that contains the command line app.

Why does this file exist, and why not put this in __main__?

You might be tempted to import things from __main__ later, but that will cause problems: the code will get executed twice:

  • When you run python -mlip_pps_run_manager python will execute __main__.py as a script. That means there won’t be any lip_pps_run_manager.__main__ in sys.modules.

  • When you import __main__ it will get executed again (as a module) because there’s no lip_pps_run_manager.__main__ in sys.modules.

Also see (1) from http://click.pocoo.org/5/setuptools/#setuptools-integration

lip_pps_run_manager.cli.main(args=None)[source]

lip_pps_run_manager.instruments submodule

class lip_pps_run_manager.instruments.Keithley6487(device_name: str, resource_string: str)[source]

Bases: VISADevice

Class for the keithley 6487 PicoAmmeter

get_current() → float[source]
get_cv() → float[source]
get_voltage() → float[source]
safe_shutdowm()[source]
set_current_range(limit: float) → bool[source]
set_source_current_limit(limit: float) → bool[source]
set_voltage(voltage: float)[source]
set_voltage_range(limit: float) → bool[source]
voltage_off() → bool[source]
voltage_on() → bool[source]
lip_pps_run_manager.instruments.get_VISA_ResourceManager()[source]

lip_pps_run_manager.instruments.functions module

lip_pps_run_manager.instruments.functions.get_VISA_ResourceManager()[source]

lip_pps_run_manager.instruments.Keithley6487 module

class lip_pps_run_manager.instruments.Keithley6487.Keithley6487(device_name: str, resource_string: str)[source]

Bases: VISADevice

Class for the keithley 6487 PicoAmmeter

get_current() → float[source]
get_cv() → float[source]
get_voltage() → float[source]
safe_shutdowm()[source]
set_current_range(limit: float) → bool[source]
set_source_current_limit(limit: float) → bool[source]
set_voltage(voltage: float)[source]
set_voltage_range(limit: float) → bool[source]
voltage_off() → bool[source]
voltage_on() → bool[source]