hrm_omero.cli
Command-line interface related functions.
1"""Command-line interface related functions.""" 2 3import argparse 4import os 5import sys 6 7import omero.gateway 8 9from loguru import logger as log 10 11from .__init__ import __version__ 12from . import formatting 13from . import hrm 14from . import omero as _omero 15from . import transfer 16from .misc import printlog 17 18 19def bool_to_exitstatus(value): 20 """Convert a boolean to a POSIX process exit code. 21 22 As boolean values in Python are a subset of int, `True` corresponds to the int value 23 '1', which is the opposite of a successful POSIX return code. Therefore, this 24 function simply inverts the boolean value to turn it into a proper exit code. In 25 case the provided value is not of type `bool` it will be returned unchanged. 26 27 Parameters 28 ---------- 29 value : bool or int 30 The value to be converted. 31 32 Returns 33 ------- 34 int 35 0 in case `value` is `True`, 1 in case `value` is `False` and `value` itself in 36 case it is not a bool. 37 """ 38 if isinstance(value, bool): 39 return not value 40 41 return value 42 43 44def arguments_parser(): 45 """Set up the commandline arguments parser. 46 47 Returns 48 ------- 49 argparse.ArgumentParser 50 The parser instance ready to be run using its `parse_args()` method. 51 """ 52 # log.debug("Parsing command line arguments...") 53 argparser = argparse.ArgumentParser( 54 description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter 55 ) 56 argparser.add_argument( 57 "-v", 58 "--verbose", 59 dest="verbosity", 60 action="count", 61 default=0, 62 help="verbose messages (repeat for more details)", 63 ) 64 65 argparser.add_argument( 66 "--version", 67 action="version", 68 version=f"%(prog)s {__version__}", 69 ) 70 71 argparser.add_argument( 72 "-c", 73 "--config", 74 default="/etc/hrm.conf", 75 help="the HRM configuration file (default: '/etc/hrm.conf')", 76 ) 77 78 argparser.add_argument( 79 "--dry-run", 80 action="store_true", 81 default=False, 82 help="print requested action and parameters without actually performing it", 83 ) 84 85 # required arguments group 86 req_args = argparser.add_argument_group( 87 "required arguments", "NOTE: MUST be given before any subcommand!" 88 ) 89 req_args.add_argument("-u", "--user", required=True, help="OMERO username") 90 91 subparsers = argparser.add_subparsers( 92 help=".", 93 dest="action", 94 description="Action to be performed, one of the following:", 95 ) 96 97 # checkCredentials parser 98 subparsers.add_parser( 99 "checkCredentials", help="check if login credentials are valid" 100 ) 101 102 # retrieveChildren parser 103 parser_subtree = subparsers.add_parser( 104 "retrieveChildren", help="get the children of a given node object (JSON)" 105 ) 106 parser_subtree.add_argument( 107 "--id", 108 type=str, 109 required=True, 110 help='ID of the parent object, e.g. "ROOT", "G:4:Experimenter:7', 111 ) 112 113 # OMEROtoHRM parser 114 parser_o2h = subparsers.add_parser( 115 "OMEROtoHRM", help="download an image from the OMERO server" 116 ) 117 parser_o2h.add_argument( 118 "-i", 119 "--imageid", 120 required=True, 121 help='the OMERO ID of the image to download, e.g. "G:4:Image:42"', 122 ) 123 parser_o2h.add_argument( 124 "-d", 125 "--dest", 126 type=str, 127 required=True, 128 help="the destination directory where to put the downloaded file", 129 ) 130 131 # HRMtoOMERO parser 132 parser_h2o = subparsers.add_parser( 133 "HRMtoOMERO", help="upload an image to the OMERO server" 134 ) 135 parser_h2o.add_argument( 136 "-d", 137 "--dset", 138 required=True, 139 dest="dset", 140 help='the ID of the target dataset in OMERO, e.g. "G:7:Dataset:23"', 141 ) 142 parser_h2o.add_argument( 143 "-f", 144 "--file", 145 type=str, 146 required=True, 147 help="the image file to upload, including the full path", 148 ) 149 parser_h2o.add_argument( 150 "-n", 151 "--name", 152 type=str, 153 required=False, 154 help="a label to use for the image in OMERO", 155 ) 156 parser_h2o.add_argument( 157 "-a", 158 "--ann", 159 type=str, 160 required=False, 161 help="annotation text to be added to the image in OMERO", 162 ) 163 164 return argparser 165 166 167def verbosity_to_loglevel(verbosity): 168 """Map the verbosity count to a named log level for `loguru`. 169 170 Parameters 171 ---------- 172 verbosity : int 173 Verbosity count as returned e.g. by the following argparse code: 174 `argparser.add_argument("-v", dest="verbosity", action="count", default=0)` 175 176 Returns 177 ------- 178 str 179 A log level name that can be used with `loguru.logger.add()`. 180 """ 181 log_level = "WARNING" # no verbosity flag has been provided -> use "WARNING" 182 if verbosity > 3: # -vvvv (4) and more will result in "TRACE" 183 log_level = "TRACE" 184 if verbosity == 3: # -vvv will be "DEBUG" 185 log_level = "DEBUG" 186 elif verbosity == 2: # -vv will be "INFO" 187 log_level = "INFO" 188 elif verbosity == 1: # -v will be "SUCCESS" 189 log_level = "SUCCESS" 190 return log_level 191 192 193def logger_add_file_sink(hrm_config, target=""): 194 """Helper to add a file sink to the logger unless disabled in the config file. 195 196 By default logging messages from the connector into a separate file is desired, so 197 this function will try to add a file sink by default. Only if the HRM configuration 198 file explicitly asks for no log file to be created it will skip this step. 199 200 Parameters 201 ---------- 202 hrm_config : dict 203 A parsed HRM configuration file as returned by `hrm_omero.hrm.parse_config()`. 204 target : str, optional 205 The path for the log file to be used. If empty (or skipped) the default 206 `$HRM_LOG/omero-connector.log` will be used, falling back to 207 `HRM_LOG="/var/log/hrm"` in case `$HRM_LOG` is not set in the hrm configuration. 208 """ 209 disable_file_logging = hrm_config.get("OMERO_CONNECTOR_LOGFILE_DISABLED", "") 210 if disable_file_logging: 211 return 212 213 if not target: 214 log_base = hrm_config.get("HRM_LOG", "/var/log/hrm") 215 target = f"{log_base}/omero-connector.log" 216 217 log_level = hrm_config.get("OMERO_CONNECTOR_LOGLEVEL", "INFO") 218 try: 219 log.add(target, level=log_level) 220 log.trace(f"Added file sink for logging: {target}.") 221 except Exception as err: # pylint: disable-msg=broad-except 222 log.error(f"Adding a file sink for logging failed: {err}") 223 224 225def run_task(args): 226 """Parse commandline arguments and initiate the requested tasks.""" 227 argparser = arguments_parser() 228 args = argparser.parse_args(args) 229 230 # one of the downsides of loguru is that the level of an existing logger can't be 231 # changed - so to adjust verbosity we actually need to remove the default logger and 232 # re-add it with the new level (see https://github.com/Delgan/loguru/issues/138) 233 log_level = verbosity_to_loglevel(args.verbosity) 234 log.remove() 235 log.add(sys.stderr, level=log_level) 236 237 log.success(f"Logging verbosity requested: {args.verbosity} ({log_level})") 238 239 hrm_config = hrm.parse_config(args.config) 240 host = hrm_config.get("OMERO_HOSTNAME", "localhost") 241 port = hrm_config.get("OMERO_PORT", 4064) 242 omero_logfile = hrm_config.get("OMERO_DEBUG_LOG", "") 243 244 log_level = hrm_config.get("OMERO_CONNECTOR_LOGLEVEL") 245 if log_level: 246 log.remove() 247 log.add(sys.stderr, level=log_level) 248 log.success(f"Log level set from config file: {log_level}") 249 250 logger_add_file_sink(hrm_config) 251 252 # NOTE: reading the OMERO password from an environment variable instead of an 253 # argument supplied on the command line improves handling of this sensitive data as 254 # the value is *NOT* immediately revealed to anyone with shell access by simply 255 # looking at the process list (which is an absolute standard procedure to do). Since 256 # it is not passed to any other functions here (except the call to `BlitzGateway`) 257 # this also prevents it from being shown in an annotated stack trace in case an 258 # uncaught exception is coming through. 259 # However, this doesn't provide super-high security as it will still be possible for 260 # an admin to inspect the environment of a running process. Nevertheless going 261 # beyond this seems a bit pointless here as an admin could also modify the code that 262 # is actually calling the connector to get hold of user credentials. 263 passwd = os.environ.get("OMERO_PASSWORD") 264 if not passwd: 265 printlog("ERROR", "ERROR: no password given to connect to OMERO!") 266 return False 267 268 if args.action == "checkCredentials": 269 log.trace("checkCredentials") 270 perform_action = _omero.check_credentials 271 kwargs = {} 272 273 elif args.action == "retrieveChildren": 274 log.trace("retrieveChildren") 275 perform_action = formatting.print_children_json 276 kwargs = {"omero_id": args.id} 277 278 elif args.action == "OMEROtoHRM": 279 log.trace("OMEROtoHRM") 280 perform_action = transfer.from_omero 281 kwargs = { 282 "omero_id": args.imageid, 283 "dest": args.dest, 284 } 285 286 elif args.action == "HRMtoOMERO": 287 log.trace("HRMtoOMERO") 288 perform_action = transfer.to_omero 289 kwargs = { 290 "omero_id": args.dset, 291 "image_file": args.file, 292 "omero_logfile": omero_logfile, 293 } 294 295 else: 296 printlog("ERROR", "No valid action specified that should be performed!") 297 return False 298 299 conn = omero.gateway.BlitzGateway( 300 username=args.user, 301 passwd=passwd, 302 host=host, 303 port=port, 304 secure=True, 305 useragent="hrm-omero.py", 306 ) 307 308 try: 309 if args.dry_run: 310 printlog("INFO", "*** dry-run, only showing action and parameters ***") 311 printlog("INFO", f"function: {perform_action.__qualname__}") 312 for key, value in kwargs.items(): 313 printlog("INFO", f"{key}: [{str(value)}]") 314 315 return True 316 317 return perform_action(conn, **kwargs) 318 319 except Exception as err: # pylint: disable-msg=broad-except # pragma: no cover 320 log.error(f"An unforeseen error occured: {err}") 321 return False 322 finally: 323 conn.close() 324 log.info(f"Closed OMERO connection [user={args.user}].") 325 326 327@log.catch 328def main(args=None): 329 """Wrapper to call the run_task() function and return its exit code.""" 330 if not args: 331 args = sys.argv[1:] 332 sys.exit(bool_to_exitstatus(run_task(args)))
20def bool_to_exitstatus(value): 21 """Convert a boolean to a POSIX process exit code. 22 23 As boolean values in Python are a subset of int, `True` corresponds to the int value 24 '1', which is the opposite of a successful POSIX return code. Therefore, this 25 function simply inverts the boolean value to turn it into a proper exit code. In 26 case the provided value is not of type `bool` it will be returned unchanged. 27 28 Parameters 29 ---------- 30 value : bool or int 31 The value to be converted. 32 33 Returns 34 ------- 35 int 36 0 in case `value` is `True`, 1 in case `value` is `False` and `value` itself in 37 case it is not a bool. 38 """ 39 if isinstance(value, bool): 40 return not value 41 42 return value
Convert a boolean to a POSIX process exit code.
As boolean values in Python are a subset of int, True
corresponds to the int value
'1', which is the opposite of a successful POSIX return code. Therefore, this
function simply inverts the boolean value to turn it into a proper exit code. In
case the provided value is not of type bool
it will be returned unchanged.
Parameters
- value (bool or int): The value to be converted.
Returns
- int: 0 in case
value
isTrue
, 1 in casevalue
isFalse
andvalue
itself in case it is not a bool.
45def arguments_parser(): 46 """Set up the commandline arguments parser. 47 48 Returns 49 ------- 50 argparse.ArgumentParser 51 The parser instance ready to be run using its `parse_args()` method. 52 """ 53 # log.debug("Parsing command line arguments...") 54 argparser = argparse.ArgumentParser( 55 description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter 56 ) 57 argparser.add_argument( 58 "-v", 59 "--verbose", 60 dest="verbosity", 61 action="count", 62 default=0, 63 help="verbose messages (repeat for more details)", 64 ) 65 66 argparser.add_argument( 67 "--version", 68 action="version", 69 version=f"%(prog)s {__version__}", 70 ) 71 72 argparser.add_argument( 73 "-c", 74 "--config", 75 default="/etc/hrm.conf", 76 help="the HRM configuration file (default: '/etc/hrm.conf')", 77 ) 78 79 argparser.add_argument( 80 "--dry-run", 81 action="store_true", 82 default=False, 83 help="print requested action and parameters without actually performing it", 84 ) 85 86 # required arguments group 87 req_args = argparser.add_argument_group( 88 "required arguments", "NOTE: MUST be given before any subcommand!" 89 ) 90 req_args.add_argument("-u", "--user", required=True, help="OMERO username") 91 92 subparsers = argparser.add_subparsers( 93 help=".", 94 dest="action", 95 description="Action to be performed, one of the following:", 96 ) 97 98 # checkCredentials parser 99 subparsers.add_parser( 100 "checkCredentials", help="check if login credentials are valid" 101 ) 102 103 # retrieveChildren parser 104 parser_subtree = subparsers.add_parser( 105 "retrieveChildren", help="get the children of a given node object (JSON)" 106 ) 107 parser_subtree.add_argument( 108 "--id", 109 type=str, 110 required=True, 111 help='ID of the parent object, e.g. "ROOT", "G:4:Experimenter:7', 112 ) 113 114 # OMEROtoHRM parser 115 parser_o2h = subparsers.add_parser( 116 "OMEROtoHRM", help="download an image from the OMERO server" 117 ) 118 parser_o2h.add_argument( 119 "-i", 120 "--imageid", 121 required=True, 122 help='the OMERO ID of the image to download, e.g. "G:4:Image:42"', 123 ) 124 parser_o2h.add_argument( 125 "-d", 126 "--dest", 127 type=str, 128 required=True, 129 help="the destination directory where to put the downloaded file", 130 ) 131 132 # HRMtoOMERO parser 133 parser_h2o = subparsers.add_parser( 134 "HRMtoOMERO", help="upload an image to the OMERO server" 135 ) 136 parser_h2o.add_argument( 137 "-d", 138 "--dset", 139 required=True, 140 dest="dset", 141 help='the ID of the target dataset in OMERO, e.g. "G:7:Dataset:23"', 142 ) 143 parser_h2o.add_argument( 144 "-f", 145 "--file", 146 type=str, 147 required=True, 148 help="the image file to upload, including the full path", 149 ) 150 parser_h2o.add_argument( 151 "-n", 152 "--name", 153 type=str, 154 required=False, 155 help="a label to use for the image in OMERO", 156 ) 157 parser_h2o.add_argument( 158 "-a", 159 "--ann", 160 type=str, 161 required=False, 162 help="annotation text to be added to the image in OMERO", 163 ) 164 165 return argparser
Set up the commandline arguments parser.
Returns
- argparse.ArgumentParser: The parser instance ready to be run using its
parse_args()
method.
168def verbosity_to_loglevel(verbosity): 169 """Map the verbosity count to a named log level for `loguru`. 170 171 Parameters 172 ---------- 173 verbosity : int 174 Verbosity count as returned e.g. by the following argparse code: 175 `argparser.add_argument("-v", dest="verbosity", action="count", default=0)` 176 177 Returns 178 ------- 179 str 180 A log level name that can be used with `loguru.logger.add()`. 181 """ 182 log_level = "WARNING" # no verbosity flag has been provided -> use "WARNING" 183 if verbosity > 3: # -vvvv (4) and more will result in "TRACE" 184 log_level = "TRACE" 185 if verbosity == 3: # -vvv will be "DEBUG" 186 log_level = "DEBUG" 187 elif verbosity == 2: # -vv will be "INFO" 188 log_level = "INFO" 189 elif verbosity == 1: # -v will be "SUCCESS" 190 log_level = "SUCCESS" 191 return log_level
Map the verbosity count to a named log level for loguru
.
Parameters
- verbosity (int):
Verbosity count as returned e.g. by the following argparse code:
argparser.add_argument("-v", dest="verbosity", action="count", default=0)
Returns
- str: A log level name that can be used with
loguru.logger.add()
.
194def logger_add_file_sink(hrm_config, target=""): 195 """Helper to add a file sink to the logger unless disabled in the config file. 196 197 By default logging messages from the connector into a separate file is desired, so 198 this function will try to add a file sink by default. Only if the HRM configuration 199 file explicitly asks for no log file to be created it will skip this step. 200 201 Parameters 202 ---------- 203 hrm_config : dict 204 A parsed HRM configuration file as returned by `hrm_omero.hrm.parse_config()`. 205 target : str, optional 206 The path for the log file to be used. If empty (or skipped) the default 207 `$HRM_LOG/omero-connector.log` will be used, falling back to 208 `HRM_LOG="/var/log/hrm"` in case `$HRM_LOG` is not set in the hrm configuration. 209 """ 210 disable_file_logging = hrm_config.get("OMERO_CONNECTOR_LOGFILE_DISABLED", "") 211 if disable_file_logging: 212 return 213 214 if not target: 215 log_base = hrm_config.get("HRM_LOG", "/var/log/hrm") 216 target = f"{log_base}/omero-connector.log" 217 218 log_level = hrm_config.get("OMERO_CONNECTOR_LOGLEVEL", "INFO") 219 try: 220 log.add(target, level=log_level) 221 log.trace(f"Added file sink for logging: {target}.") 222 except Exception as err: # pylint: disable-msg=broad-except 223 log.error(f"Adding a file sink for logging failed: {err}")
Helper to add a file sink to the logger unless disabled in the config file.
By default logging messages from the connector into a separate file is desired, so this function will try to add a file sink by default. Only if the HRM configuration file explicitly asks for no log file to be created it will skip this step.
Parameters
- hrm_config (dict):
A parsed HRM configuration file as returned by
hrm_omero.hrm.parse_config()
. - target (str, optional):
The path for the log file to be used. If empty (or skipped) the default
$HRM_LOG/omero-connector.log
will be used, falling back toHRM_LOG="/var/log/hrm"
in case$HRM_LOG
is not set in the hrm configuration.
226def run_task(args): 227 """Parse commandline arguments and initiate the requested tasks.""" 228 argparser = arguments_parser() 229 args = argparser.parse_args(args) 230 231 # one of the downsides of loguru is that the level of an existing logger can't be 232 # changed - so to adjust verbosity we actually need to remove the default logger and 233 # re-add it with the new level (see https://github.com/Delgan/loguru/issues/138) 234 log_level = verbosity_to_loglevel(args.verbosity) 235 log.remove() 236 log.add(sys.stderr, level=log_level) 237 238 log.success(f"Logging verbosity requested: {args.verbosity} ({log_level})") 239 240 hrm_config = hrm.parse_config(args.config) 241 host = hrm_config.get("OMERO_HOSTNAME", "localhost") 242 port = hrm_config.get("OMERO_PORT", 4064) 243 omero_logfile = hrm_config.get("OMERO_DEBUG_LOG", "") 244 245 log_level = hrm_config.get("OMERO_CONNECTOR_LOGLEVEL") 246 if log_level: 247 log.remove() 248 log.add(sys.stderr, level=log_level) 249 log.success(f"Log level set from config file: {log_level}") 250 251 logger_add_file_sink(hrm_config) 252 253 # NOTE: reading the OMERO password from an environment variable instead of an 254 # argument supplied on the command line improves handling of this sensitive data as 255 # the value is *NOT* immediately revealed to anyone with shell access by simply 256 # looking at the process list (which is an absolute standard procedure to do). Since 257 # it is not passed to any other functions here (except the call to `BlitzGateway`) 258 # this also prevents it from being shown in an annotated stack trace in case an 259 # uncaught exception is coming through. 260 # However, this doesn't provide super-high security as it will still be possible for 261 # an admin to inspect the environment of a running process. Nevertheless going 262 # beyond this seems a bit pointless here as an admin could also modify the code that 263 # is actually calling the connector to get hold of user credentials. 264 passwd = os.environ.get("OMERO_PASSWORD") 265 if not passwd: 266 printlog("ERROR", "ERROR: no password given to connect to OMERO!") 267 return False 268 269 if args.action == "checkCredentials": 270 log.trace("checkCredentials") 271 perform_action = _omero.check_credentials 272 kwargs = {} 273 274 elif args.action == "retrieveChildren": 275 log.trace("retrieveChildren") 276 perform_action = formatting.print_children_json 277 kwargs = {"omero_id": args.id} 278 279 elif args.action == "OMEROtoHRM": 280 log.trace("OMEROtoHRM") 281 perform_action = transfer.from_omero 282 kwargs = { 283 "omero_id": args.imageid, 284 "dest": args.dest, 285 } 286 287 elif args.action == "HRMtoOMERO": 288 log.trace("HRMtoOMERO") 289 perform_action = transfer.to_omero 290 kwargs = { 291 "omero_id": args.dset, 292 "image_file": args.file, 293 "omero_logfile": omero_logfile, 294 } 295 296 else: 297 printlog("ERROR", "No valid action specified that should be performed!") 298 return False 299 300 conn = omero.gateway.BlitzGateway( 301 username=args.user, 302 passwd=passwd, 303 host=host, 304 port=port, 305 secure=True, 306 useragent="hrm-omero.py", 307 ) 308 309 try: 310 if args.dry_run: 311 printlog("INFO", "*** dry-run, only showing action and parameters ***") 312 printlog("INFO", f"function: {perform_action.__qualname__}") 313 for key, value in kwargs.items(): 314 printlog("INFO", f"{key}: [{str(value)}]") 315 316 return True 317 318 return perform_action(conn, **kwargs) 319 320 except Exception as err: # pylint: disable-msg=broad-except # pragma: no cover 321 log.error(f"An unforeseen error occured: {err}") 322 return False 323 finally: 324 conn.close() 325 log.info(f"Closed OMERO connection [user={args.user}].")
Parse commandline arguments and initiate the requested tasks.
328@log.catch 329def main(args=None): 330 """Wrapper to call the run_task() function and return its exit code.""" 331 if not args: 332 args = sys.argv[1:] 333 sys.exit(bool_to_exitstatus(run_task(args)))
Wrapper to call the run_task() function and return its exit code.