nexuslims_logger namespace

Submodules

nexuslims_logger.dbsessionlogger module

class nexuslims_logger.dbsessionlogger.DBSessionLogger(dbapi_url, dbapi_username=None, dbapi_password=None, user=None, logger=None)

Bases: object

communicate with database.

check_exit_queue(thread_queue, exit_queue)

Check to see if a queue (exit_queue) has anything in it. If so, immediately exit.

Parameters

• thread_queue (queue.Queue) –

• exit_queue (queue.Queue) –

db_logger_setup(thread_queue=None, exit_queue=None)

get instrument info (pid, schema name).

db_logger_teardown(thread_queue=None, exit_queue=None)

teardown routine

classmethod from_config(config, user=None, logger=None)

last_session_ended(thread_queue=None, exit_queue=None)

Check the database for this instrument to make sure that the last entry in the db was an “END” (properly ended). If it’s not, return False so the GUI can query the user for additional input on how to proceed.

Parameters

• thread_queue (queue.Queue) – Main queue for communication with the GUI

• exit_queue (queue.Queue) – Queue containing any errors so the GUI knows to exit as needed

Returns

state_is_consistent – If the database is consistent (i.e. the last log for this instrument is an “END” log), return True. If not (it’s a “START” log), return False

Return type

bool

process_end(thread_queue=None, exit_queue=None)

Insert a session ‘END’ log for this computer’s instrument, and change the status of the corresponding ‘START’ entry from ‘WAITING_FOR_END’ to ‘TO_BE_BUILT’

process_start(thread_queue=None, exit_queue=None)

Insert a session ‘START’ log for this computer’s instrument

Returns True if successful, False if not

save_note(thread_queue=None, exit_queue=None)

nexuslims_logger.filewatcher module

FileWatcher will watch a directory,and sync with Cloud periodically.

It will upload any files (require file types match if specified) (modified after certain time if specified) that checksum changed wrt. cache (if any) to a GCP cloud bucket in a specified interval.

class nexuslims_logger.filewatcher.FileWatcher(watch_dir, bucket_name, bucket_dir, credentials, cache_fn, project=None, interval=600, file_types=None, mtime_since=None, instr_info=None, logger=None)

Bases: object

property bucket_dir

classmethod from_config(config, credentials, cache_fn, logger=None)

get_files_to_upload()

find files to upload recursively and return list of abs file names and content checksum.

file satisfying the following condition will be considered for uploading: - file type is allowed (set in app config) - file modification timestamp is after the set threshold (session start time) - file content checksum does not exist in cache or updated.

Returns

List of tuple consisting file names and MD5 checksum.

Return type

List[Tuple[str, str]]

property instr_info

property interval

property mtime_since

upload()

upload to the cloud object storage, set metadata and update the cache.

nexuslims_logger.gui module

GUI impl

class nexuslims_logger.gui.App(db_logger, instrument, filewatcher, screen_res=None, logger=None, log_text=None)

Bases: tkinter.Tk

close_warning(num_to_show)

set loading text to remind the closing action

create_widgets()

draw widgets on main frame.

disable_buttons()

bring state of all buttons on main window to DISABLED

done_loading()

actions by the end of loading.

put off setup_frame, put up running_frame and labels, enable buttons.

draw_buttons()

buttons at the bottom

draw_info()

information section

incl. loading text, progress bar, instrument label, datetime

draw_logo()

Top NexusLIMS logo with tooltip.

enable_buttons()

bring state of all buttons on main window to NORMAL

on_closing()

actions when user is closing the main window.

session_end()

routines for session ending.

signal startup_thread to exit, start end_thread, change the frame, update loading text and progress bar.

session_end_worker()

session ending routine.

db_logger perform ending actions, stop file syncing thread, final file syncing, tear down db_logger

session_startup()

start startup_thread, update loading progress bar.

session_startup_worker()

session startup routine

setup db_logger, check if last session successfully ended; If so, starts a new session, otherwise a dialogue window should prompt user to decide whether to continue the unfinished session or start a new session.

After the db_logger has been successfully setup, the information of the instrument attached with this computer (fetched by db_logger) will be passed to filewatcher. Then timeloop thread will start to syncing the instrument outputs to the cloud storage.

show_error_if_needed(res)

show error box if res is an Exception

switch_gui_to_end()

actions by the start of ending.

put off running_frame; putup setup_frame, diable buttons

watch_for_end_result()

Check if there is something in the queue.

update loading text and progress bar.

watch_for_startup_result()

Check if there is something in the queue.

update loading text and progress bar.

class nexuslims_logger.gui.ScreenRes(logger=None)

Bases: object

get_center_geometry_string(width, height)

This method will return a Tkinter geometry string that will place a Toplevel window into the middle of the screen given the widget’s width and height (using a Windows command or xrandr as needed). If it fails for some reason, a basic resolution of 800x600 is assumed.

Parameters

• width (int) – The width of the widget desired

• height (int) – The height of the widget desired

Returns

geometry_string – The Tkinter geometry string that will put a window of width and height at the center of the screen given the current resolution (of the format “WIDTHxHEIGHT+XPOSITION+YPOSITION”)

Return type

str

run_cmd(cmd)

Run a command using the subprocess module and return the output. Note that because we want to run the eventual logger without a console visible, we do not have access to the standard stdin, stdout, and stderr, and these need to be redirected subprocess pipes, accordingly.

Parameters

cmd (str) – The command to run (will be run in a new Windows cmd shell). stderr will be redirected for stdout and included in the returned output

Returns

output – The output of cmd

Return type

str

nexuslims_logger.instrument module

class nexuslims_logger.instrument.GCPInstrument(output_dir, bucket_name, bucket_dir, credentials, project=None, logger=None)

Bases: nexuslims_logger.instrument.Instrument

Mock an instrument with data stored in GCP.

classmethod from_config(config, credentials, logger=None)

generate_data()

Download random file from GCP bucket.

save to self.output_dir, and rename by timestamp.

get_file_pool()

List all files in GCP self.bucket_dir

nexuslims_logger.run module

nexuslims_logger.run.help()

nexuslims_logger.run.main()

nexuslims_logger.run.validate_config(config)

simple validation of config settings

nexuslims_logger.utils module

utility functions

nexuslims_logger.utils.check_singleton()

make sure only ONE instance of the program running.

nexuslims_logger.utils.get_logger(name, verbose=20, stream=None)

get a logger from logging module, direct output to stdout, verbose level set by verbose.

If additional stream is provided, direct output (DEBUG) to that stream too.

nexuslims_logger.utils.show_error_msg_box(msg)

show a tkinter error box.