Source code for

# --------------------------------------------------------------------------
# Source file provided under Apache License, Version 2.0, January 2004,
# (c) Copyright IBM Corp. 2015, 2016
# --------------------------------------------------------------------------

from import DOcplexException, DOcplexLimitsExceeded, resolve_pattern, is_int, is_string

from enum import Enum
import os

# Error handling

class IErrorHandler(object):
    def __init__(self):
        pass  # pragma: no cover

    def info(self, msg, args=None):
        pass  # pragma: no cover

    def warning(self, msg, args=None):
        pass  # pragma: no cover

    def error(self, msg, args=None):
        pass  # pragma: no cover

    def fatal(self, msg, args=None):
        pass  # pragma: no cover

    def ok(self):
        return False  # pragma: no cover

    def get_output_level(self):
        return 0  # pragma: no cover

    def set_output_level(self, new_level):
        pass  # pragma: no cover

    def ensure(self, condition, msg, *args):
        if not condition:
            self.fatal(msg, args)

[docs]class InfoLevel(Enum): """ Enumerated type for the possible output levels. Info levels are sorted in increasing order of severity: `INFO`, `WARNING`, `ERROR`, `FATAL`. Setting a level enables the printing of all messages from that severity and above. Example: Setting the level to `WARNING` enables the printing of `WARNING`, `ERROR`, and `FATAL` messages, but not `INFO` level messages. Setting the level to `FATAL` suppresses all messages, except for fatal errors. """ INFO, WARNING, ERROR, FATAL = 1, 10, 100, 9999999 @classmethod def parse(cls, arg, default_level=INFO): # INTERNAL if not arg: return cls.INFO elif isinstance(arg, cls): return arg elif is_string(arg): return cls._name2level_map().get(arg.lower(), default_level) elif is_int(arg): if arg < 10: # anything below 10 is INFO return cls.INFO elif arg < 100: return cls.WARNING elif arg < 1000: return cls.ERROR else: # level fatal prints nothing except fatal errors return cls.FATAL else: raise DOcplexException("Cannot convert this to InfoLevel: {0!r}".format(arg)) def __str__(self): return @staticmethod def _headers(): return {InfoLevel.FATAL: "FATAL", InfoLevel.INFO: "*", InfoLevel.WARNING: "Warning:", InfoLevel.ERROR: "Error:" } @staticmethod def _name2level_map(): return {"fatal": InfoLevel.FATAL, "error": InfoLevel.ERROR, "warning": InfoLevel.WARNING, "info": InfoLevel.INFO} def header(self): # cannot put the dict in the class # as it willbe interpreted as another enum value. return self._headers().get(self, "???")
class AbstractErrorHandler(IErrorHandler): TRACE_HEADER = "--" def __init__(self, output_level=InfoLevel.INFO): IErrorHandler.__init__(self) self._trace_enabled = False self._number_of_errors = 0 self._number_of_warnings = 0 self._number_of_fatals = 0 self._output_level = InfoLevel.INFO self._is_print_suspended = False self._postponed = [] self.set_output_level(output_level) @property def number_of_warnings(self): """ Returns the number of warnings. """ return self._number_of_warnings @property def number_of_errors(self): """ Returns the number of errors. """ return self._number_of_errors @property def number_of_fatals(self): return self._number_of_fatals def get_output_level(self): return self._output_level def set_output_level(self, output_level_arg): output_level = InfoLevel.parse(output_level_arg) if output_level != self._output_level: self._output_level = output_level def set_trace_mode(self, trace_mode): self._trace_enabled = trace_mode def enable_trace(self): self.set_trace_mode(True) def disable_trace(self): self.set_trace_mode(False) def is_trace_enabled(self): return self._trace_enabled def set_quiet(self): """ Changes the output level to enable only error messages. """ self.set_output_level(InfoLevel.ERROR) def reset(self): self._number_of_errors = 0 self._number_of_warnings = 0 self._number_of_fatals = 0 def _internal_is_printed(self, level): return self._output_level.value <= level.value def _internal_print_if(self, level, msg, args): if self._internal_is_printed(level): self._internal_print(level, msg, args) def _internal_print(self, level, msg, args): # resolve message w/ args header = level.header() self._internal_print_header(header, msg, args) def _internal_print_header(self, header, msg, args): resolved_message = resolve_pattern(msg, args) mline = '%s %s' % (header, resolved_message) if self._is_print_suspended: self._postponed.append(mline) else: print(mline) def trace_header(self): return self.TRACE_HEADER def trace(self, msg, args=None): if self.is_trace_enabled(): self._internal_print_header(self.trace_header(), msg, args) def info(self, msg, args=None): self._internal_print_if(InfoLevel.INFO, msg, args) def warning(self, msg, args=None): self._number_of_warnings += 1 self._internal_print_if(InfoLevel.WARNING, msg, args) def error(self, msg, args=None): docplex_error_stop_here() self._number_of_errors += 1 self._internal_print_if(InfoLevel.ERROR, msg, args) def fatal(self, msg, args=None): self._number_of_fatals += 1 resolved_message = resolve_pattern(msg, args) docplex_error_stop_here() raise DOcplexException(resolved_message) def fatal_limits_exceeded(self, nb_vars, nb_constraints): docplex_error_stop_here() raise DOcplexLimitsExceeded(nb_vars, nb_constraints) def ok(self): """ Checks whether the handler has not recorded any error. """ return self._number_of_errors == 0 and self._number_of_fatals == 0 def prints_trace(self): return self.is_trace_enabled() def prints_info(self): return self._internal_is_printed(InfoLevel.INFO) def prints_warning(self): return self._internal_is_printed(InfoLevel.WARNING) def prints_error(self): return self._internal_is_printed(InfoLevel.ERROR) def suspend(self): self._is_print_suspended = True def flush(self): self._is_print_suspended = False for m in self._postponed: print(m) self._postponed = [] def docplex_error_stop_here(): # INTERNAL, use to set breakpoints pass def docplex_add_trivial_infeasible_ct_here(): # INTERNAL: set breakpoint here to inspect ct pass def docplex_fatal(msg, *args): resolved_message = resolve_pattern(msg, args) docplex_error_stop_here() raise DOcplexException(resolved_message) def is_docplex_debug(): return not not os.environ.get('DOCPLEX_DEBUG') def docplex_debug_msg(*args): if is_docplex_debug(): msg = ' '.join(str(x) for x in args) #print(f"-- {msg}") print("-- {0}".format(msg))
[docs]class DefaultErrorHandler(AbstractErrorHandler): """ The default error handler class. """ def __init__(self, output_level=InfoLevel): AbstractErrorHandler.__init__(self, output_level)
class SilentErrorHandler(AbstractErrorHandler): def __init__(self, output_level=InfoLevel): AbstractErrorHandler.__init__(self, output_level) def _internal_print(self, level, msg, args): # nothing out, this is the point! pass def suspend(self): pass def flush(self): pass def handle_error(logger, error, msg): if error == "raise": raise logger.fatal(msg) elif error == "warn": logger.warn(msg) elif error =="ignore": pass else: print("unknown error directive: {0}, expecting ignore|warn|raise, msg is: {1}".format(error, msg))