diff --git a/ppp.py b/ppp.py index 5f30edd..927f6be 100644 --- a/ppp.py +++ b/ppp.py @@ -9,9 +9,9 @@ import numpy as np import yaml from ppp_classes import IFWILDCARDS_CHOICES, SUPPORTED_APPS, PPPInterrupt, PPPState, PPPStateOptions -from ppp_logging import DEBUG_LEVEL +from ppp_logging import DEBUG_LEVEL, log from ppp_tree import TreeProcessor -from ppp_utils import escape_single_quotes, format_output +from ppp_utils import escape_single_quotes from ppp_common import load_grammar, parse_prompt, preprocess_grammar, warn_or_stop from ppp_wildcards import PPPWildcards from ppp_enmappings import PPPExtraNetworkMappings @@ -47,27 +47,27 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in defopt = {f.name: f.default for f in dataclasses.fields(PPPStateOptions)} DEFAULT_DEBUG_LEVEL = defopt["debug_level"].value - DEFAULT_ONWARNING = defopt["gen_onwarning"].value + DEFAULT_ON_WARNING = defopt["on_warning"].value DEFAULT_STN_SEPARATOR = defopt["stn_separator"] DEFAULT_STN_IGNORE_REPEATS = defopt["stn_ignore_repeats"] - DEFAULT_WC_PROCESS = defopt["wil_process_wildcards"] - DEFAULT_IF_WILDCARDS = defopt["wil_ifwildcards"].value - DEFAULT_CHOICE_SEPARATOR = defopt["wil_choice_separator"] - DEFAULT_KEEP_CHOICES_ORDER = defopt["wil_keep_choices_order"] + DEFAULT_PROCESS_WILDCARDS = defopt["process_wildcards"] + DEFAULT_IF_WILDCARDS = defopt["if_wildcards"].value + DEFAULT_CHOICE_SEPARATOR = defopt["choice_separator"] + DEFAULT_KEEP_CHOICES_ORDER = defopt["keep_choices_order"] DEFAULT_DO_CLEANUP = defopt["cup_do_cleanup"] DEFAULT_CLEANUP_VARIABLES = defopt["cup_cleanup_variables"] - DEFAULT_CUP_EXTRA_SPACES = defopt["cup_extraspaces"] - DEFAULT_CUP_EMPTY_CONSTRUCTS = defopt["cup_emptyconstructs"] - DEFAULT_CUP_EXTRA_SEPARATORS = defopt["cup_extraseparators"] - DEFAULT_CUP_EXTRA_SEPARATORS2 = defopt["cup_extraseparators2"] - DEFAULT_CUP_EXTRA_SEPARATORS_INCLUDE_EOL = defopt["cup_extraseparators_include_eol"] + DEFAULT_CUP_EXTRA_SPACES = defopt["cup_extra_spaces"] + DEFAULT_CUP_EMPTY_CONSTRUCTS = defopt["cup_empty_constructs"] + DEFAULT_CUP_EXTRA_SEPARATORS = defopt["cup_extra_separators"] + DEFAULT_CUP_EXTRA_SEPARATORS2 = defopt["cup_extra_separators2"] + DEFAULT_CUP_EXTRA_SEPARATORS_INCLUDE_EOL = defopt["cup_extra_separators_include_eol"] DEFAULT_CUP_BREAKS = defopt["cup_breaks"] DEFAULT_CUP_BREAKS_EOL = defopt["cup_breaks_eol"] DEFAULT_CUP_ANDS = defopt["cup_ands"] DEFAULT_CUP_ANDS_EOL = defopt["cup_ands_eol"] - DEFAULT_CUP_EXTRANETWORK_TAGS = defopt["cup_extranetworktags"] - DEFAULT_CUP_MERGE_ATTENTION = defopt["cup_mergeattention"] - DEFAULT_CUP_REMOVE_EXTRANETWORK_TAGS = defopt["rem_removeextranetworktags"] + DEFAULT_CUP_EXTRANETWORK_TAGS = defopt["cup_extranetwork_tags"] + DEFAULT_CUP_MERGE_ATTENTION = defopt["cup_merge_attention"] + DEFAULT_CUP_REMOVE_EXTRANETWORK_TAGS = defopt["cup_remove_extranetwork_tags"] WILDCARD_WARNING = '(WARNING TEXT "INVALID WILDCARD" IN BRIGHT RED:1.5)\nBREAK ' WILDCARD_STOP = "INVALID WILDCARD! {0}\nBREAK " UNPROCESSED_STOP = "UNPROCESSED CONSTRUCTS!\nBREAK " @@ -95,6 +95,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in extranetwork_mappings_obj: Optional. The extranetwork mappings object to be used for processing. """ self.logger = logger + self.debug_level = options.debug_level self.interrupt_callback = interrupt self.env_info = env_info @@ -112,7 +113,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in errmsg = "Default configuration file has errors. Please restore the default configuration file and, per instructions, use a copy to adapt it." if validate_def_cfg == 2: raise PPPInterrupt(errmsg) - self.logger.warning(errmsg) + self.log(logging.WARNING, errmsg) user_config_file = self.env_info.get("ppp_config", "") user_config: dict[str, Any] = {} @@ -129,7 +130,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in if user_dir and os.path.isdir(user_dir): user_config_file = os.path.join(user_dir, "default", "ppp_config.yaml") except Exception: # pylint: disable=broad-exception-caught - self.logger.warning("Failed to get user directory for PPP config.") + self.log(logging.WARNING, "Failed to get user directory for PPP config.") if not user_config_file or not os.path.exists(user_config_file): user_config_file = os.path.join(os.path.dirname(os.path.realpath(__file__)), "ppp_config.yaml") if user_config_file and os.path.exists(user_config_file): @@ -177,15 +178,13 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in if v not in self.known_models: self.variants_definitions[v] = (m, vo["find_in_filename"]) else: - self.logger.warning( - f"Variant name '{escape_single_quotes(v)}' in model '{escape_single_quotes(m)}' conflicts with a known model name. Discarding variant." + self.log( + logging.WARNING, + f"Variant name '{escape_single_quotes(v)}' in model '{escape_single_quotes(m)}' conflicts with a known model name. Discarding variant.", ) - self.debug_level = options.debug_level - if self.debug_level != DEBUG_LEVEL.none: - self.logger.debug(format_output(f"Host configuration: {host_config}")) + self.log(logging.DEBUG, f"Host configuration: {host_config}", min_level=DEBUG_LEVEL.minimal) - # if self.debug_level != DEBUG_LEVEL.none: - # self.logger.info(f"Detected environment info: {env_info}") + # self.log(logging.INFO, f"Detected environment info: {env_info}", min_level=DEBUG_LEVEL.minimal) if grammar_content is None: grammar_content = load_grammar() @@ -198,6 +197,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": True, "ALLOW_COMMVARS": True, }, + self.logger, + self.debug_level, ) self.state = PPPState( @@ -223,6 +224,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": True, "ALLOW_COMMVARS": False, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -235,6 +238,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": False, "ALLOW_COMMVARS": True, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -247,6 +252,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": True, "ALLOW_COMMVARS": True, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -259,6 +266,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": False, "ALLOW_COMMVARS": False, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -271,6 +280,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": True, "ALLOW_COMMVARS": False, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -283,6 +294,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": False, "ALLOW_COMMVARS": True, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -295,6 +308,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in "ALLOW_CHOICES": False, "ALLOW_COMMVARS": False, }, + self.logger, + self.debug_level, ), propagate_positions=True, ), @@ -328,6 +343,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in ) self.__init_sysvars() + def log(self, kind, message: str, min_level: DEBUG_LEVEL | None = None): + log(self.logger, self.debug_level, kind, message, min_level) + def __merge_configuration(self, user_config): """ Merges the user configuration into the default configuration. @@ -376,28 +394,32 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in re.compile(find_in_filename, re.IGNORECASE) return {"regex": find_in_filename, "flags": re.IGNORECASE} except re.error: - self.logger.warning( - f"{where.title()}: Invalid regex pattern for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant." + self.log( + logging.WARNING, + f"{where.title()}: Invalid regex pattern for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant.", ) elif isinstance(find_in_filename, dict): regex = find_in_filename.get("regex", "") flags = find_in_filename.get("flags", []) if not isinstance(regex, str) or not isinstance(flags, list) or not all(isinstance(f, str) for f in flags): - self.logger.warning( - f"{where.title()}: Invalid format for 'find_in_filename' for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant." + self.log( + logging.WARNING, + f"{where.title()}: Invalid format for 'find_in_filename' for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant.", ) else: fl = self.__re_flags_from_list(flags) if fl == 0 and len(flags): - self.logger.warning( - f"{where.title()}: Invalid regex flags for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant." + self.log( + logging.WARNING, + f"{where.title()}: Invalid regex flags for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant.", ) try: re.compile(regex, fl) return {"regex": regex, "flags": fl} except re.error: - self.logger.warning( - f"{where.title()}: Invalid regex pattern for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant." + self.log( + logging.WARNING, + f"{where.title()}: Invalid regex pattern for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant.", ) return None @@ -413,14 +435,14 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in """ fatal_errors = False if not isinstance(cfg, dict): - self.logger.error(f"{where.capitalize()}: is not a dictionary.") + self.log(logging.ERROR, f"{where.capitalize()}: is not a dictionary.") fatal_errors = True else: if cfg.get("hosts") and not isinstance(cfg["hosts"], dict): - self.logger.error(f"{where.capitalize()}: 'hosts' is not a valid dictionary.") + self.log(logging.ERROR, f"{where.capitalize()}: 'hosts' is not a valid dictionary.") fatal_errors = True if cfg.get("models") and not isinstance(cfg.get("models"), dict): - self.logger.error(f"{where.capitalize()}: 'models' is not a valid dictionary.") + self.log(logging.ERROR, f"{where.capitalize()}: 'models' is not a valid dictionary.") fatal_errors = True if fatal_errors: return 2 @@ -428,8 +450,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in defcfg_hosts: dict[str, Any] = cfg.get("hosts", {}) for host_key, host_value in dict(defcfg_hosts).items(): if host_key not in SUPPORTED_APPS._value2member_map_: # pylint: disable=protected-access - self.logger.warning( - f"{where.capitalize()}: Unsupported host '{escape_single_quotes(host_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Unsupported host '{escape_single_quotes(host_key)}'. Discarding host.", ) defcfg_hosts.pop(host_key, None) result = 1 @@ -437,8 +460,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in not isinstance(host_value, dict) or not all(k in ["attention", "scheduling", "alternation", "and", "break"] for k in host_value) ): - self.logger.warning( - f"{where.capitalize()}: Invalid format for host '{escape_single_quotes(host_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for host '{escape_single_quotes(host_key)}'. Discarding host.", ) defcfg_hosts.pop(host_key, None) result = 1 @@ -449,8 +473,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in or model_value.get("detect") is None or not isinstance(model_value["detect"], dict) ): - self.logger.warning( - f"{where.capitalize()}: Invalid format for model '{escape_single_quotes(model_key)}'. Discarding model." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for model '{escape_single_quotes(model_key)}'. Discarding model.", ) defcfg_models.pop(model_key, None) result = 1 @@ -458,15 +483,17 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in defcfg_m_detect: dict[str, Any] = model_value["detect"] for host_key, host_value in dict(defcfg_m_detect).items(): if host_key not in SUPPORTED_APPS._value2member_map_: # pylint: disable=protected-access - self.logger.warning( - f"{where.capitalize()}: Unsupported host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Unsupported host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host.", ) defcfg_m_detect.pop(host_key, None) result = 1 elif host_value is not None: if not isinstance(host_value, dict): - self.logger.warning( - f"{where.capitalize()}: Invalid format for host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host.", ) defcfg_m_detect.pop(host_key, None) result = 1 @@ -474,28 +501,32 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in if not isinstance(host_value["class"], list) or not all( isinstance(c, str) for c in host_value["class"] ): - self.logger.warning( - f"{where.capitalize()}: Invalid format for 'class' in host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for 'class' in host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host.", ) defcfg_m_detect.pop(host_key, None) result = 1 elif "property" in host_value: if not isinstance(host_value["property"], str): - self.logger.warning( - f"{where.capitalize()}: Invalid format for 'property' in host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for 'property' in host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host.", ) defcfg_m_detect.pop(host_key, None) result = 1 else: - self.logger.warning( - f"{where.capitalize()}: Neither 'class' nor 'property' specified for host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host." + self.log( + logging.WARNING, + f"{where.capitalize()}: Neither 'class' nor 'property' specified for host '{escape_single_quotes(host_key)}' in 'detect' for model '{escape_single_quotes(model_key)}'. Discarding host.", ) defcfg_m_detect.pop(host_key, None) result = 1 if "variants" in model_value: if not isinstance(model_value["variants"], dict): - self.logger.warning( - f"{where.capitalize()}: Invalid format for 'variants' in model '{escape_single_quotes(model_key)}'. Discarding model." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for 'variants' in model '{escape_single_quotes(model_key)}'. Discarding model.", ) defcfg_models.pop(model_key, None) result = 1 @@ -503,16 +534,18 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in defcfg_m_variants: dict[str, Any] = model_value["variants"] for variant_key, variant_value in dict(defcfg_m_variants).items(): if not isinstance(variant_key, str) or not variant_key.isidentifier(): - self.logger.warning( - f"{where.capitalize()}: Invalid variant name '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid variant name '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant.", ) defcfg_m_variants.pop(variant_key, None) result = 1 elif not isinstance(variant_value, dict) or not isinstance( variant_value.get("find_in_filename"), (str, dict, list) ): - self.logger.warning( - f"{where.capitalize()}: Invalid format for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant." + self.log( + logging.WARNING, + f"{where.capitalize()}: Invalid format for variant '{escape_single_quotes(variant_key)}' in model '{escape_single_quotes(model_key)}'. Discarding variant.", ) defcfg_m_variants.pop(variant_key, None) result = 1 @@ -587,8 +620,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in ) is_models_true = [k for k, v in is_models.items() if v] if len(is_models_true) > 1: - self.logger.warning( - f"Multiple model variants detected at the same time in the filename!: {', '.join(is_models_true)}" + self.log( + logging.WARNING, + f"Multiple model variants detected at the same time in the filename!: {', '.join(is_models_true)}", ) sv.update({"_is_" + x: y for x, y in is_models.items()}) for x in sdchecks.keys(): @@ -735,8 +769,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in text2 = re.sub(r"\b\s*BREAK\s*\b", break_replacements[break_processing][1], text) if text2 != text: text = text2 - if self.debug_level == DEBUG_LEVEL.full: - self.logger.debug(f"BREAK construct {break_replacements[break_processing][0]}") + self.log(logging.DEBUG, f"BREAK construct {break_replacements[break_processing][0]}") elif break_processing == "error": if re.search(r"\bBREAK\b", text): warn_or_stop(self.state, where == -1, "BREAK constructs are not allowed!") @@ -756,14 +789,14 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in text = re.sub(r"(\s*\bAND)+\Z", "", text) escapedSeparator = re.escape(self.state.options.stn_separator) - optwhitespace = r"\s*" if self.state.options.cup_extraseparators_include_eol else r"[ \t\v\f]*" + optwhitespace = r"\s*" if self.state.options.cup_extra_separators_include_eol else r"[ \t\v\f]*" optwhitespace_separator = optwhitespace + escapedSeparator + optwhitespace optwhitespace_comma = optwhitespace + "," + optwhitespace sep_options = [(optwhitespace_separator, self.state.options.stn_separator)] # sendtonegative separator if optwhitespace_comma != optwhitespace_separator: sep_options.append((optwhitespace_comma, ", ")) # regular comma separator for sep, replacement in sep_options: - if self.state.options.cup_extraseparators: + if self.state.options.cup_extra_separators: # collapse separators text = re.sub(r"(?:" + sep + r"){2,}", replacement, text) # remove separator after starting parenthesis, starting bracket @@ -786,17 +819,17 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in r"\1", text, ) - if self.state.options.cup_extraseparators2: + if self.state.options.cup_extra_separators2: # remove at start of prompt or line text = re.sub(r"^(?:" + sep + r")+", "", text, flags=re.MULTILINE) # remove at end of prompt or line text = re.sub(r"(?:" + sep + r")+$", "", text, flags=re.MULTILINE) - if self.state.options.cup_extranetworktags: + if self.state.options.cup_extranetwork_tags: # remove spaces before < text = re.sub(r"\B\s+<(?!!)", "<", text) # remove spaces after > text = re.sub(r">\s+\B", ">", text) - if self.state.options.cup_extraspaces: + if self.state.options.cup_extra_spaces: # remove spaces before comma text = re.sub(r"[ ]+,", ",", text) # remove spaces at end of line @@ -891,8 +924,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in # Process prompt p_processor = TreeProcessor(self.state, rng) (prompt_parser, parser_description) = self.__get_best_parser(prompt) - if self.debug_level == DEBUG_LEVEL.full: - self.logger.debug(f"Using {parser_description} for prompt") + self.log(logging.DEBUG, f"Using {parser_description} for prompt") p_parsed = parse_prompt( self.state, "prompt", @@ -904,8 +936,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in # Process negative prompt n_processor = TreeProcessor(self.state, rng) (n_prompt_parser, n_parser_description) = self.__get_best_parser(negative_prompt) - if self.debug_level == DEBUG_LEVEL.full: - self.logger.debug(f"Using {n_parser_description} for negative prompt") + self.log(logging.DEBUG, f"Using {n_parser_description} for negative prompt") n_parsed = parse_prompt( self.state, "negative prompt", @@ -921,17 +952,14 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in if ev is None: ev = self.state.user_variables.get(k) if ev is None or not isinstance(ev, str): - if self.debug_level == DEBUG_LEVEL.full: - self.logger.debug(format_output(f"Completing variable: {k}")) + self.log(logging.DEBUG, f"Completing variable: {k}") ev = p_processor.get_final_user_variable(k) all_variables[k] = self.__cleanup(ev, 0) if self.state.options.cup_cleanup_variables else ev - if self.debug_level == DEBUG_LEVEL.full: - self.logger.debug(format_output(f"All variables: {all_variables}")) + self.log(logging.DEBUG, f"All variables: {all_variables}") # Insertions in the negative prompt - if self.debug_level == DEBUG_LEVEL.full: - self.logger.debug(format_output(f"New negative additions: {p_processor.add_at}")) - self.logger.debug(format_output(f"New negative indexes: {n_processor.insertion_at}")) + self.log(logging.DEBUG, f"New negative additions: {p_processor.add_at}") + self.log(logging.DEBUG, f"New negative indexes: {n_processor.insertion_at}") negative_prompt = self.__add_to_insertion_points( negative_prompt, p_processor.add_at["insertion_point"], n_processor.insertion_at ) @@ -948,19 +976,19 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in foundP = bool(p_processor.detectedWildcards) foundNP = bool(n_processor.detectedWildcards) if foundP or foundNP: - if self.state.options.wil_ifwildcards == IFWILDCARDS_CHOICES.stop: - self.logger.error("Found unprocessed wildcards!") + if self.state.options.if_wildcards == IFWILDCARDS_CHOICES.stop: + self.log(logging.ERROR, "Found unprocessed wildcards!") else: - self.logger.info("Found unprocessed wildcards.") + self.log(logging.INFO, "Found unprocessed wildcards.") ppwl = ", ".join(p_processor.detectedWildcards) npwl = ", ".join(n_processor.detectedWildcards) if foundP: - self.logger.error(format_output(f"In the positive prompt: {ppwl}")) + self.log(logging.ERROR, f"In the positive prompt: {ppwl}") if foundNP: - self.logger.error(format_output(f"In the negative prompt: {npwl}")) - if self.state.options.wil_ifwildcards == IFWILDCARDS_CHOICES.warn: + self.log(logging.ERROR, f"In the negative prompt: {npwl}") + if self.state.options.if_wildcards == IFWILDCARDS_CHOICES.warn: prompt = self.WILDCARD_WARNING + prompt - elif self.state.options.wil_ifwildcards == IFWILDCARDS_CHOICES.stop: + elif self.state.options.if_wildcards == IFWILDCARDS_CHOICES.stop: raise PPPInterrupt( "Found unprocessed wildcards!", self.WILDCARD_STOP.format(ppwl) if foundP else "", @@ -971,8 +999,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in compound_prompt = prompt + "\n" + negative_prompt found_sequences = re.findall(r"::|\$\$|\$\{|[{}]", compound_prompt) if found_sequences: - self.logger.warning( - f"""Found probably invalid character sequences on the result ({', '.join(map(lambda x: '"' + x + '"', set(found_sequences)))}). Something might be wrong!""" + self.log( + logging.WARNING, + f"""Found probably invalid character sequences on the result ({', '.join(map(lambda x: '"' + x + '"', set(found_sequences)))}). Something might be wrong!""", ) return prompt, negative_prompt, all_variables @@ -999,23 +1028,20 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in seed = np.random.randint(0, 2**32, dtype=np.int64) prompt = original_prompt negative_prompt = original_negative_prompt - if self.debug_level != DEBUG_LEVEL.none: - self.logger.info(f"System variables: {self.state.system_variables}") - self.logger.info(f"Input seed: {seed}") - self.logger.info(format_output(f"Input prompt: {prompt}")) - self.logger.info(format_output(f"Input negative_prompt: {negative_prompt}")) + self.log(logging.INFO, f"System variables: {self.state.system_variables}") + self.log(logging.INFO, f"Input seed: {seed}") + self.log(logging.INFO, f"Input prompt: {prompt}") + self.log(logging.INFO, f"Input negative_prompt: {negative_prompt}") t1 = time.monotonic_ns() prompt, negative_prompt, all_variables = self.__processprompts( np.random.default_rng(seed & 0xFFFFFFFF), prompt, negative_prompt ) t2 = time.monotonic_ns() - if self.debug_level != DEBUG_LEVEL.none: - self.logger.info(format_output(f"Result prompt: {prompt}")) - self.logger.info(format_output(f"Result negative_prompt: {negative_prompt}")) - self.logger.info(f"Process prompt pair time: {(t2 - t1) / 1_000_000_000:.3f} seconds") + self.log(logging.INFO, f"Result prompt: {prompt}") + self.log(logging.INFO, f"Result negative_prompt: {negative_prompt}") + self.log(logging.INFO, f"Process prompt pair time: {(t2 - t1) / 1_000_000_000:.3f} seconds") - # if self.debug_level != DEBUG_LEVEL.none: - # self.logger.debug(f"Wildcards memory usage: {self.state.wildcards_obj.__sizeof__()}") + # self.log(logging.DEBUG,f"Wildcards memory usage: {self.state.wildcards_obj.__sizeof__()}") # Check for constructs not processed due to parsing problems fullcontent: str = prompt + negative_prompt if fullcontent.find("= 0: @@ -1026,14 +1052,14 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in ) return prompt, negative_prompt, all_variables except PPPInterrupt as e: - self.logger.error(e.message) + self.log(logging.ERROR, e.message) if e.pos_prefix: prompt = e.pos_prefix + prompt if e.neg_prefix: negative_prompt = e.neg_prefix + negative_prompt - self.logger.error("Interrupting!") + self.log(logging.ERROR, "Interrupting!") self.interrupt() return prompt, negative_prompt, all_variables except Exception as e: # pylint: disable=broad-exception-caught - self.logger.exception(e) + self.log(logging.ERROR, f"Unexpected error: {e}") return original_prompt, original_negative_prompt, all_variables diff --git a/ppp_classes.py b/ppp_classes.py index a3064ac..55629a4 100644 --- a/ppp_classes.py +++ b/ppp_classes.py @@ -49,27 +49,27 @@ class PPPStateOptions: """Options that can be set for prompt processing.""" debug_level: DEBUG_LEVEL = DEBUG_LEVEL.minimal - gen_onwarning: ONWARNING_CHOICES = ONWARNING_CHOICES.warn - wil_process_wildcards: bool = True - wil_keep_choices_order: bool = True - wil_choice_separator: str = ", " - wil_ifwildcards: IFWILDCARDS_CHOICES = IFWILDCARDS_CHOICES.stop + on_warning: ONWARNING_CHOICES = ONWARNING_CHOICES.warn + process_wildcards: bool = True + keep_choices_order: bool = True + choice_separator: str = ", " + if_wildcards: IFWILDCARDS_CHOICES = IFWILDCARDS_CHOICES.stop stn_ignore_repeats: bool = True stn_separator: str = ", " cup_do_cleanup: bool = True # whether to do cleanup at all (if False, all other cleanup options are ignored) cup_cleanup_variables: bool = True - cup_extraspaces: bool = True - cup_emptyconstructs: bool = True - cup_extraseparators: bool = True - cup_extraseparators2: bool = True - cup_extraseparators_include_eol: bool = False + cup_extra_spaces: bool = True + cup_empty_constructs: bool = True + cup_extra_separators: bool = True + cup_extra_separators2: bool = True + cup_extra_separators_include_eol: bool = False cup_breaks: bool = False cup_breaks_eol: bool = False cup_ands: bool = False cup_ands_eol: bool = False - cup_extranetworktags: bool = False - cup_mergeattention: bool = True - rem_removeextranetworktags: bool = False + cup_extranetwork_tags: bool = False + cup_merge_attention: bool = True + cup_remove_extranetwork_tags: bool = False @dataclass(frozen=True) diff --git a/ppp_comfyui.py b/ppp_comfyui.py index 364f588..e4c82ab 100644 --- a/ppp_comfyui.py +++ b/ppp_comfyui.py @@ -1,3 +1,4 @@ +import logging import os import folder_paths # pylint: disable=import-error # type: ignore @@ -5,7 +6,7 @@ import nodes # pylint: disable=import-error # type: ignore from ppp import PromptPostProcessor from ppp_classes import IFWILDCARDS_CHOICES, ONWARNING_CHOICES, SUPPORTED_APPS, PPPStateOptions -from ppp_logging import DEBUG_LEVEL, PromptPostProcessorLogFactory +from ppp_logging import DEBUG_LEVEL, PromptPostProcessorLogFactory, log from ppp_utils import escape_single_quotes from ppp_wildcards import PPPWildcards from ppp_enmappings import PPPExtraNetworkMappings @@ -69,7 +70,12 @@ class PromptPostProcessorComfyUINode: self.grammar_content = file.read() self.wildcards_obj = PPPWildcards(lf.log) self.extranetwork_mappings_obj = PPPExtraNetworkMappings(lf.log) - self.logger.info(f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized") + log( + self.logger, + DEBUG_LEVEL.minimal, + logging.INFO, + f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized", + ) class SmartType(str): def __ne__(self, other): @@ -134,14 +140,14 @@ class PromptPostProcessorComfyUINode: "on_warnings": ( [e.value for e in ONWARNING_CHOICES], { - "default": PromptPostProcessor.DEFAULT_ONWARNING, + "default": PromptPostProcessor.DEFAULT_ON_WARNING, "tooltip": "How to handle invalid content warnings", }, ), "process_wildcards": ( "BOOLEAN", { - "default": PromptPostProcessor.DEFAULT_WC_PROCESS, + "default": PromptPostProcessor.DEFAULT_PROCESS_WILDCARDS, "tooltip": "Process wildcards in the prompt", "label_on": "Yes", "label_off": "No", @@ -249,9 +255,19 @@ class PromptPostProcessorComfyUINode: model.model.model_config.__class__.__name__ if model is not None and not isinstance(model, str) else model ) or "" if modelclass == "": - self.logger.warning("Model class is not provided. System variables might not be properly set.") + log( + self.logger, + DEBUG_LEVEL.minimal, + logging.WARNING, + "Model class is not provided. System variables might not be properly set.", + ) if modelname == "": - self.logger.warning("Modelname is not provided. System variables will not be properly set.") + log( + self.logger, + DEBUG_LEVEL.minimal, + logging.WARNING, + "Modelname is not provided. System variables will not be properly set.", + ) # model class values in ComfyUI\comfy\supported_models.py env_info = { "app": SUPPORTED_APPS.comfyui.value, @@ -265,13 +281,13 @@ class PromptPostProcessorComfyUINode: options = PPPStateOptions( debug_level=DEBUG_LEVEL(debug_level), - gen_onwarning=ONWARNING_CHOICES(on_warnings) if on_warnings else PromptPostProcessor.DEFAULT_ONWARNING, - wil_process_wildcards=process_wildcards, - wil_ifwildcards=(wc_options["wc_if_wildcards"] if wc_options else IFWILDCARDS_CHOICES.stop.value), - wil_choice_separator=( + on_warning=ONWARNING_CHOICES(on_warnings) if on_warnings else PromptPostProcessor.DEFAULT_ON_WARNING, + process_wildcards=process_wildcards, + if_wildcards=(wc_options["wc_if_wildcards"] if wc_options else IFWILDCARDS_CHOICES.stop.value), + choice_separator=( wc_options["wc_choice_separator"] if wc_options else PromptPostProcessor.DEFAULT_CHOICE_SEPARATOR ), - wil_keep_choices_order=( + keep_choices_order=( wc_options["wc_keep_choices_order"] if wc_options else PromptPostProcessor.DEFAULT_KEEP_CHOICES_ORDER ), stn_separator=stn_options["stn_separator"] if stn_options else PromptPostProcessor.DEFAULT_STN_SEPARATOR, @@ -280,21 +296,21 @@ class PromptPostProcessorComfyUINode: ), cup_do_cleanup=do_cleanup, cup_cleanup_variables=cleanup_variables, - cup_extraspaces=( + cup_extra_spaces=( cup_options["cup_extra_spaces"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SPACES ), - cup_emptyconstructs=( + cup_empty_constructs=( cup_options["cup_empty_constructs"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EMPTY_CONSTRUCTS ), - cup_extraseparators=( + cup_extra_separators=( cup_options["cup_extra_separators"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS ), - cup_extraseparators2=( + cup_extra_separators2=( cup_options["cup_extra_separators2"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS2 ), - cup_extraseparators_include_eol=( + cup_extra_separators_include_eol=( cup_options["cup_extra_separators_include_eol"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS_INCLUDE_EOL @@ -305,15 +321,15 @@ class PromptPostProcessorComfyUINode: ), cup_ands=cup_options["cup_ands"] if cup_options else PromptPostProcessor.DEFAULT_CUP_ANDS, cup_ands_eol=(cup_options["cup_ands_eol"] if cup_options else PromptPostProcessor.DEFAULT_CUP_ANDS_EOL), - cup_extranetworktags=( + cup_extranetwork_tags=( cup_options["cup_extranetwork_tags"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRANETWORK_TAGS ), - cup_mergeattention=( + cup_merge_attention=( cup_options["cup_merge_attention"] if cup_options else PromptPostProcessor.DEFAULT_CUP_MERGE_ATTENTION ), - rem_removeextranetworktags=( + cup_remove_extranetwork_tags=( cup_options["cup_remove_extranetwork_tags"] if cup_options else PromptPostProcessor.DEFAULT_CUP_REMOVE_EXTRANETWORK_TAGS @@ -321,7 +337,7 @@ class PromptPostProcessorComfyUINode: ) self.wildcards_obj.refresh_wildcards( options.debug_level, - wildcards_folders if options.wil_process_wildcards else None, + wildcards_folders if options.process_wildcards else None, wc_options["wc_wildcards_input"] if wc_options else "", ) self.extranetwork_mappings_obj.refresh_extranetwork_mappings( diff --git a/ppp_common.py b/ppp_common.py index 0eefe88..07c0942 100644 --- a/ppp_common.py +++ b/ppp_common.py @@ -6,7 +6,7 @@ import textwrap import time import lark -from ppp_logging import DEBUG_LEVEL +from ppp_logging import log from ppp_classes import ONWARNING_CHOICES, PPPInterrupt, PPPState from ppp_utils import escape_single_quotes, format_output @@ -33,12 +33,12 @@ def parse_prompt( t1 = time.monotonic_ns() parsed_prompt = None try: - if state.options.debug_level == DEBUG_LEVEL.full: - state.logger.debug( - format_output( - f"Parsing {prompt_description}: '{escape_single_quotes(prompt)}'" - ) - ) + log( + state.logger, + state.options.debug_level, + logging.DEBUG, + f"Parsing {prompt_description}: '{escape_single_quotes(prompt)}'", + ) parsed_prompt = parser.parse(prompt) # we store the contents so we can use them later even if the meta position is not valid anymore if isinstance(parsed_prompt, lark.Tree): @@ -51,42 +51,43 @@ def parse_prompt( except lark.exceptions.UnexpectedInput: if raise_parsing_error: raise - state.logger.exception( - format_output(f"Parsing failed on prompt!: {escape_single_quotes(prompt)}") + log( + state.logger, + state.options.debug_level, + logging.ERROR, + f"Parsing failed on prompt!: {escape_single_quotes(prompt)}", ) t2 = time.monotonic_ns() - if state.options.debug_level == DEBUG_LEVEL.full: - state.logger.debug( - f"Parse {prompt_description} time: {(t2 - t1) / 1_000_000_000:.3f} seconds" + log( + state.logger, + state.options.debug_level, + logging.DEBUG, + f"Parse {prompt_description} time: {(t2 - t1) / 1_000_000_000:.3f} seconds", + ) + if parsed_prompt: + log( + state.logger, + state.options.debug_level, + logging.DEBUG, + "Tree:\n" + + textwrap.indent( + re.sub(r"\n$", "", (parsed_prompt.pretty() if isinstance(parsed_prompt, lark.Tree) else parsed_prompt)), + " ", + ), ) - if parsed_prompt: - state.logger.debug( - "Tree:\n" - + textwrap.indent( - re.sub( - r"\n$", - "", - ( - parsed_prompt.pretty() - if isinstance(parsed_prompt, lark.Tree) - else parsed_prompt - ), - ), - " ", - ) - ) return parsed_prompt def warn_or_stop(state: PPPState, is_negative: bool, message: str, e: Exception = None): INVALID_CONTENT_STOP = "INVALID CONTENT! {0}\nBREAK " - if state.options.gen_onwarning == ONWARNING_CHOICES.stop: + if state.options.on_warning == ONWARNING_CHOICES.stop: raise PPPInterrupt( message, INVALID_CONTENT_STOP.format(message) if not is_negative else "", INVALID_CONTENT_STOP.format(message) if is_negative else "", ) from e - state.logger.warning(format_output(message)) + log(state.logger, state.options.debug_level, logging.WARNING, format_output(message)) + def load_grammar() -> str: # Process with lark (debug with https://www.lark-parser.org/ide/) @@ -96,13 +97,15 @@ def load_grammar() -> str: return grammar_content -def preprocess_grammar(grammar_content: str, options: dict[str, bool], logger: logging.Logger = None) -> str: +def preprocess_grammar(grammar_content: str, options: dict[str, bool], logger: logging.Logger, debug_level: int) -> str: """ Preprocesses the grammar content to handle conditional compilation directives. Args: grammar_content (str): The raw grammar content. options (dict[str,bool]): Options for preprocessing. + logger (logging.Logger): The logger object. + debug_level (int): The debug level for logging. Returns: str: The preprocessed grammar content. @@ -153,8 +156,7 @@ def preprocess_grammar(grammar_content: str, options: dict[str, bool], logger: l all_blocks_skipped.append(skip_current_block[-1]) elif stripped_line.startswith("//#elif"): if not skip_current_block: - if logger: - logger.warning("Unmatched //#elif directive found in grammar content.") + log(logger, debug_level, logging.WARNING, "Unmatched //#elif directive found in grammar content.") elif all_blocks_skipped[-1]: # Extract condition from the #elif directive conditions = stripped_line[7:].strip() @@ -166,22 +168,24 @@ def preprocess_grammar(grammar_content: str, options: dict[str, bool], logger: l skip_current_block[-1] = True elif stripped_line.startswith("//#else"): if not skip_current_block: - if logger: - logger.warning("Unmatched //#else directive found in grammar content.") + log(logger, debug_level, logging.WARNING, "Unmatched //#else directive found in grammar content.") elif all_blocks_skipped[-1]: skip_current_block[-1] = False else: skip_current_block[-1] = True elif stripped_line.startswith("//#endif"): if not skip_current_block: - if logger: - logger.warning("Unmatched //#endif directive found in grammar content.") + log(logger, debug_level, logging.WARNING, "Unmatched //#endif directive found in grammar content.") else: skip_current_block.pop() all_blocks_skipped.pop() elif stripped_line.startswith("//#"): - if logger: - logger.warning(f"Unrecognized directive found in grammar content: {stripped_line}") + log( + logger, + debug_level, + logging.WARNING, + f"Unrecognized directive found in grammar content: {stripped_line}", + ) elif not any(skip_current_block): # Include the line if we're not skipping any current block result_lines.append(stripped_line) diff --git a/ppp_enmappings.py b/ppp_enmappings.py index efd7ae3..e69b219 100644 --- a/ppp_enmappings.py +++ b/ppp_enmappings.py @@ -3,7 +3,7 @@ from typing import Optional import logging import yaml -from ppp_logging import DEBUG_LEVEL +from ppp_logging import DEBUG_LEVEL, log from ppp_utils import deep_freeze, escape_single_quotes @@ -95,8 +95,7 @@ class PPPExtraNetworkMappings: """ self.__debug_level = debug_level self.__enmappings_folders = enmappings_folders or [] - # if self.__debug_level != DEBUG_LEVEL.none and self.__logger: - # self.__logger.info("Refreshing extra network mappings...") + # log(self.__logger, self.__debug_level, logging.INFO, "Refreshing extra network mappings...") # t1 = time.monotonic_ns() self.cached_mappings = {} for fullpath in list(self.__enmappings_files.keys()): @@ -118,8 +117,7 @@ class PPPExtraNetworkMappings: self.extranetwork_mappings = {} self.__enmappings_files = {} # t2 = time.monotonic_ns() - # if self.__debug_level != DEBUG_LEVEL.none and self.__logger: - # self.__logger.info(f"Extra network mappings refresh time: {(t2 - t1) / 1_000_000_000:.3f} seconds") + # log(self.__logger, self.__debug_level, logging.INFO, f"Extra network mappings refresh time: {(t2 - t1) / 1_000_000_000:.3f} seconds") # def get_extranetwork_mappings(self, key: str) -> list[PPPENMapping]: # """ @@ -143,11 +141,16 @@ class PPPExtraNetworkMappings: debug (bool): Whether to print debug messages or not. """ last_modified_cached = self.__enmappings_files.get(full_path, None) # a time or a hash - if debug and last_modified_cached is not None and self.__debug_level != DEBUG_LEVEL.none and self.__logger: + if debug and last_modified_cached is not None: if full_path == self.LOCALINPUT_FILENAME: - self.__logger.debug("Removing extra network mappings from input") + log(self.__logger, self.__debug_level, logging.DEBUG, "Removing extra network mappings from input") else: - self.__logger.debug(f"Removing extra network mappings from file: {full_path}") + log( + self.__logger, + self.__debug_level, + logging.DEBUG, + f"Removing extra network mappings from file: {full_path}", + ) if full_path in self.__enmappings_files.keys(): del self.__enmappings_files[full_path] for key in list(self.extranetwork_mappings.keys()): @@ -170,8 +173,13 @@ class PPPExtraNetworkMappings: if extension not in (".yaml", ".yml", ".json"): return self.__remove_extranetwork_mappings_from_path(full_path, False) - if last_modified_cached is not None and self.__debug_level != DEBUG_LEVEL.none and self.__logger: - self.__logger.debug(f"Updating extra network mappings from file: {full_path}") + if last_modified_cached is not None: + log( + self.__logger, + self.__debug_level, + logging.DEBUG, + f"Updating extra network mappings from file: {full_path}", + ) self.__get_extranetwork_mappings_in_structured_file(full_path) self.__enmappings_files[full_path] = last_modified @@ -187,15 +195,19 @@ class PPPExtraNetworkMappings: if h == new_h: return self.__remove_extranetwork_mappings_from_path(self.LOCALINPUT_FILENAME, False) - if h is not None and self.__debug_level != DEBUG_LEVEL.none and self.__logger: - self.__logger.debug("Updating extra network mappings from input") + if h is not None: + log(self.__logger, self.__debug_level, logging.DEBUG, "Updating extra network mappings from input") enmappings_input = enmappings_input.strip() if enmappings_input != "": try: content = yaml.safe_load(enmappings_input) except yaml.YAMLError as e: - if self.__logger: - self.__logger.warning(f"Invalid format for input extra network mappings: {e}") + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid format for input extra network mappings: {e}", + ) return if content is not None: self.__add_extranetwork_mapping(content, self.LOCALINPUT_FILENAME) @@ -210,33 +222,45 @@ class PPPExtraNetworkMappings: full_path (str): The path to the file that contains it. """ if not isinstance(content, dict): - if self.__logger: - self.__logger.warning(f"Invalid extra network mapping in file '{escape_single_quotes(full_path)}'!") + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid extra network mapping in file '{escape_single_quotes(full_path)}'!", + ) return for kind, maps in content.items(): if not isinstance(maps, dict): - if self.__logger: - self.__logger.warning( - f"Invalid extra network mapping definition for '{escape_single_quotes(kind)}:*' in file '{escape_single_quotes(full_path)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid extra network mapping definition for '{escape_single_quotes(kind)}:*' in file '{escape_single_quotes(full_path)}'!", + ) else: for name, variants in maps.items(): key = f"{kind}:{name}" if not isinstance(variants, list): - if self.__logger: - self.__logger.warning( - f"Invalid extra network mapping definition for '{escape_single_quotes(key)}' in file '{escape_single_quotes(full_path)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid extra network mapping definition for '{escape_single_quotes(key)}' in file '{escape_single_quotes(full_path)}'!", + ) elif self.extranetwork_mappings.get(key, None) is not None: - if self.__logger: - self.__logger.warning( - f"Duplicate extra network mapping '{escape_single_quotes(key)}' in file '{escape_single_quotes(full_path)}' and '{escape_single_quotes(self.extranetwork_mappings[key].file)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Duplicate extra network mapping '{escape_single_quotes(key)}' in file '{escape_single_quotes(full_path)}' and '{escape_single_quotes(self.extranetwork_mappings[key].file)}'!", + ) elif not isinstance(variants, list) or not all(isinstance(v, dict) for v in variants): - if self.__logger: - self.__logger.warning( - f"Invalid extra network mapping definition for '{escape_single_quotes(key)}' in file '{escape_single_quotes(full_path)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid extra network mapping definition for '{escape_single_quotes(key)}' in file '{escape_single_quotes(full_path)}'!", + ) else: self.extranetwork_mappings[key] = PPPENMapping(full_path, kind, name, variants) @@ -253,18 +277,22 @@ class PPPExtraNetworkMappings: with open(full_path, "r", encoding="utf-8") as file: content = yaml.safe_load(file) except: # pylint: disable=bare-except - if self.__logger: - self.__logger.warning( - f"Could not read file '{escape_single_quotes(full_path)}' with utf-8 encoding, trying windows-1252..." - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Could not read file '{escape_single_quotes(full_path)}' with utf-8 encoding, trying windows-1252...", + ) with open(full_path, "r", encoding="windows-1252") as file: content = yaml.safe_load(file) self.__add_extranetwork_mapping(content, full_path) except Exception as e: # pylint: disable=broad-except - if self.__logger: - self.__logger.error( - f"Error reading extra network mappings from file '{escape_single_quotes(full_path)}': {e}" - ) + log( + self.__logger, + self.__debug_level, + logging.ERROR, + f"Error reading extra network mappings from file '{escape_single_quotes(full_path)}': {e}", + ) def __get_extranetwork_mappings_in_directory(self, directory: str): """ @@ -274,10 +302,12 @@ class PPPExtraNetworkMappings: directory (str): The path to the directory. """ if not os.path.exists(directory): - if self.__logger: - self.__logger.warning( - f"Extra network mappings directory '{escape_single_quotes(directory)}' does not exist!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Extra network mappings directory '{escape_single_quotes(directory)}' does not exist!", + ) return for filename in os.listdir(directory): full_path = os.path.abspath(os.path.join(directory, filename)) diff --git a/ppp_logging.py b/ppp_logging.py index 1574d1e..130d857 100644 --- a/ppp_logging.py +++ b/ppp_logging.py @@ -3,6 +3,8 @@ import logging import sys import copy +from ppp_utils import format_output + class DEBUG_LEVEL(Enum): none = "none" @@ -95,3 +97,17 @@ class PromptPostProcessorLogCustomAdapter(logging.LoggerAdapter): tuple: A tuple containing the processed log message and keyword arguments. """ return f"[PPP] {msg}", kwargs + +def log(logger: logging.Logger, debug_level: DEBUG_LEVEL, kind: int, message: str, min_level: DEBUG_LEVEL | None = None): + if logger: + if min_level is None: + if kind == logging.DEBUG: + min_level = DEBUG_LEVEL.full + elif kind == logging.INFO: + min_level = DEBUG_LEVEL.minimal + else: + min_level = DEBUG_LEVEL.none + i_debug_level = list(DEBUG_LEVEL).index(debug_level) + i_min_level = list(DEBUG_LEVEL).index(min_level) + if i_debug_level >= i_min_level: + logger.log(kind, format_output(message)) diff --git a/ppp_tree.py b/ppp_tree.py index 5b8265f..673884b 100644 --- a/ppp_tree.py +++ b/ppp_tree.py @@ -1,4 +1,5 @@ from collections import namedtuple +import logging import math import re import textwrap @@ -9,8 +10,8 @@ import numpy as np from ppp_classes import IFWILDCARDS_CHOICES, PPPState from ppp_enmappings import PPPENMappingVariant -from ppp_logging import DEBUG_LEVEL -from ppp_utils import escape_single_quotes, format_output +from ppp_logging import DEBUG_LEVEL, log +from ppp_utils import escape_single_quotes from ppp_common import parse_prompt, warn_or_stop from ppp_wildcards import PPPWildcard @@ -48,6 +49,9 @@ class TreeProcessor(lark.visitors.Interpreter): self.detectedWildcards: list[str] = [] self.result = "" + def log(self, kind, message: str, min_level: DEBUG_LEVEL | None = None): + log(self.__state.logger, self.__state.options.debug_level, kind, message, min_level) + def warn_or_stop(self, message: str, e: Exception = None): warn_or_stop(self.__state, self.__is_negative, message, e) @@ -65,12 +69,10 @@ class TreeProcessor(lark.visitors.Interpreter): """ t1 = time.monotonic_ns() self.__is_negative = is_negative - if self.__debug_level != DEBUG_LEVEL.none: - self.__state.logger.info(f"Processing {prompt_description}...") + self.log(logging.INFO, f"Processing {prompt_description}...") self.visit(parsed_prompt) t2 = time.monotonic_ns() - if self.__debug_level != DEBUG_LEVEL.none: - self.__state.logger.info(f"Process {prompt_description} time: {(t2 - t1) / 1_000_000_000:.3f} seconds") + self.log(logging.INFO, f"Process {prompt_description} time: {(t2 - t1) / 1_000_000_000:.3f} seconds") return self.result def __visit( @@ -91,11 +93,9 @@ class TreeProcessor(lark.visitors.Interpreter): str: The result of the visit. """ backup_result = self.result - # if self.__debug_level == DEBUG_LEVEL.full: - # self.__state.logger.debug(f"Visiting node {node}.") + # self.log(logging.DEBUG, f"Visiting node {node}.") if restore_state: - # if self.__debug_level == DEBUG_LEVEL.full: - # self.__state.logger.debug("Backing up state before visiting.") + # self.log(logging.DEBUG, "Backing up state before visiting.") backup_shell = self.__shell.copy() backup_negtags = self.__negtags.copy() backup_already_processed = self.__already_processed.copy() @@ -120,8 +120,7 @@ class TreeProcessor(lark.visitors.Interpreter): if discard_content or restore_state: self.result = backup_result if restore_state: - # if self.__debug_level == DEBUG_LEVEL.full: - # self.__state.logger.debug("Restoring state after visiting.") + # self.log(logging.DEBUG, "Restoring state after visiting.") self.__shell = backup_shell self.__negtags = backup_negtags self.__already_processed = backup_already_processed @@ -210,9 +209,7 @@ class TreeProcessor(lark.visitors.Interpreter): output = self.result[len(start_result) :] if output != "": output = f" >> '{escape_single_quotes(output)}'" - self.__state.logger.debug( - format_output(f"TreeProcessor.{construct} {info}({duration / 1_000_000_000:.3f} seconds){output}") - ) + self.log(logging.DEBUG, f"TreeProcessor.{construct} {info}({duration / 1_000_000_000:.3f} seconds){output}") def __resolve_cond_value(self, c: str): """Resolve a condition value: try int first, fall back to variable lookup.""" @@ -322,7 +319,7 @@ class TreeProcessor(lark.visitors.Interpreter): Returns: bool: The result of the if condition evaluation. """ - # self.__state.logger.debug(f"__eval_condition {condition.data}") + # self.log(logging.DEBUG, f"__eval_condition {condition.data}") if condition.data == "operation_and": cond_result = True for c in condition.children: @@ -388,8 +385,7 @@ class TreeProcessor(lark.visitors.Interpreter): + and_replacements[and_processing][1] + self.__visit(tree.children[i + 1], False, True).lstrip() ) - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"AND construct {and_replacements[and_processing][0]}") + self.log(logging.DEBUG, f"AND construct {and_replacements[and_processing][0]}") elif and_processing == "error": self.warn_or_stop("AND constructs are not allowed!") else: # and_processing == "ok": @@ -423,43 +419,37 @@ class TreeProcessor(lark.visitors.Interpreter): pos = int(pos) scheduling_processing = self.__state.host_config.get("scheduling", "ok") if scheduling_processing == "before": - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Scheduling construct removed, taking before option") + self.log(logging.DEBUG, "Scheduling construct removed, taking before option") if before is not None: self.__visit(before) elif scheduling_processing == "after": - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Scheduling construct removed, taking after option") + self.log(logging.DEBUG, "Scheduling construct removed, taking after option") if after is not None: self.__visit(after) elif scheduling_processing == "first": - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Scheduling construct removed, taking first option") + self.log(logging.DEBUG, "Scheduling construct removed, taking first option") if before is not None: self.__visit(before) elif after is not None: self.__visit(after) elif scheduling_processing == "remove": - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Scheduling construct removed") + self.log(logging.DEBUG, "Scheduling construct removed") elif scheduling_processing == "error": self.warn_or_stop("Scheduling constructs are not allowed!") else: # scheduling_processing == "ok" # self.__shell.append(self.AccumulatedShell("sc", pos)) self.result += "[" if before is not None: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Shell scheduled before with position {pos}") + self.log(logging.DEBUG, f"Shell scheduled before with position {pos}") self.__shell.append(self.AccumulatedShell("scb", pos)) self.__visit(before) self.__shell.pop() - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Shell scheduled after with position {pos}") + self.log(logging.DEBUG, f"Shell scheduled after with position {pos}") self.__shell.append(self.AccumulatedShell("sca", pos)) self.result += ":" self.__visit(after) self.__shell.pop() - if self.__state.options.cup_emptyconstructs and re.fullmatch( + if self.__state.options.cup_empty_constructs and re.fullmatch( re.escape(start_result) + r"\[:\s*", self.result ): self.result = start_result @@ -477,27 +467,24 @@ class TreeProcessor(lark.visitors.Interpreter): t1 = time.monotonic_ns() alternation_processing = self.__state.host_config.get("alternation", "ok") if alternation_processing == "first": - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Alternation construct removed, taking first option") + self.log(logging.DEBUG, "Alternation construct removed, taking first option") self.__visit(tree.children[0]) elif alternation_processing == "remove": - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Alternation construct removed") + self.log(logging.DEBUG, "Alternation construct removed") elif alternation_processing == "error": self.warn_or_stop("Alternation constructs are not allowed!") else: # alternation_processing == "ok" # self.__shell.append(self.AccumulatedShell("al", len(tree.children))) self.result += "[" for i, opt in enumerate(tree.children): - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Shell alternate option {i+1}") + self.log(logging.DEBUG, f"Shell alternate option {i+1}") self.__shell.append(self.AccumulatedShell("alo", {"pos": i + 1, "len": len(tree.children)})) if i > 0: self.result += "|" self.__visit(opt) self.__shell.pop() self.result += "]" - if self.__state.options.cup_emptyconstructs and re.fullmatch( + if self.__state.options.cup_empty_constructs and re.fullmatch( re.escape(start_result) + r"\[\s*\]", self.result ): self.result = start_result @@ -525,10 +512,9 @@ class TreeProcessor(lark.visitors.Interpreter): weight_kind = 1 # decrease attention weight = 0.9 weight_str = "0.9" - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Shell attention with weight {weight}") + self.log(logging.DEBUG, f"Shell attention with weight {weight}") current_tree = tree.children[0] - if self.__state.options.cup_mergeattention: + if self.__state.options.cup_merge_attention: while isinstance(current_tree, lark.Tree) and current_tree.data == "attention": # we merge the weights if len(current_tree.children) == 2: @@ -554,16 +540,13 @@ class TreeProcessor(lark.visitors.Interpreter): if weight_kind == 1: weight_kind = 3 weight_str = "0.9" - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Converted to parentheses format") + self.log(logging.DEBUG, "Converted to parentheses format") elif attention_processing == "disable": weight_kind = 0 - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Attention construct disabled") + self.log(logging.DEBUG, "Attention construct disabled") elif attention_processing == "remove": weight_kind = -1 - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Attention construct removed") + self.log(logging.DEBUG, "Attention construct removed") elif attention_processing == "error": self.warn_or_stop("Attention constructs are not allowed!") # else: attention_processing == "ok": @@ -590,7 +573,7 @@ class TreeProcessor(lark.visitors.Interpreter): self.result += starttag self.__visit(current_tree) endtag = f":{weight_str})" - if self.__state.options.cup_emptyconstructs and re.fullmatch( + if self.__state.options.cup_empty_constructs and re.fullmatch( re.escape(start_result + starttag) + r"\s*", self.result ): self.result = start_result @@ -737,10 +720,7 @@ class TreeProcessor(lark.visitors.Interpreter): value = self.__get_user_variable_value(variable, True, True) if value is None: if default is not None: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug( - f"Variable '{escape_single_quotes(variable)}' not found, using default value" - ) + self.log(logging.DEBUG, f"Variable '{escape_single_quotes(variable)}' not found, using default value") v = self.__visit(default, False, True) self.result += v default_value = v @@ -799,7 +779,7 @@ class TreeProcessor(lark.visitors.Interpreter): t1 = time.monotonic_ns() start_result = self.result extnet = "(ignored)" - if not self.__state.options.rem_removeextranetworktags: + if not self.__state.options.cup_remove_extranetwork_tags: extnet_type: str = (tree.children[0].children[0] or "") + str(tree.children[0].children[1]) is_mapping = extnet_type.startswith("$") if is_mapping: @@ -871,9 +851,10 @@ class TreeProcessor(lark.visitors.Interpreter): self.__state.extranetwork_mappings_obj.cached_mappings[extnet_id] = found if found: if found.name: - if not found_in_cache and self.__debug_level != DEBUG_LEVEL.none: - self.__state.logger.info( - f"Mapping extranetwork '{escape_single_quotes(extnet_id)}' to '{escape_single_quotes(extnet_type)}:{escape_single_quotes(found.name)}'" + if not found_in_cache: + self.log( + logging.INFO, + f"Mapping extranetwork '{escape_single_quotes(extnet_id)}' to '{escape_single_quotes(extnet_type)}:{escape_single_quotes(found.name)}'", ) extnet_id = f"{extnet_type}:{found.name}" f_parameters = found.parameters @@ -889,15 +870,16 @@ class TreeProcessor(lark.visitors.Interpreter): elif f_parameters is not None and parameters_defaulted: parameters = f_parameters elif found.triggers: - if not found_in_cache and self.__debug_level != DEBUG_LEVEL.none: - self.__state.logger.info( - f"Mapping extranetwork '{escape_single_quotes(extnet_id)}' to just triggers" + if not found_in_cache: + self.log( + logging.INFO, + f"Mapping extranetwork '{escape_single_quotes(extnet_id)}' to just triggers", ) extnet_id = None else: - if not found_in_cache and self.__debug_level != DEBUG_LEVEL.none: - self.__state.logger.info( - f"Mapping extranetwork '{escape_single_quotes(extnet_id)}' to nothing" + if not found_in_cache: + self.log( + logging.INFO, f"Mapping extranetwork '{escape_single_quotes(extnet_id)}' to nothing" ) extnet_id = None if found.triggers: @@ -921,7 +903,7 @@ class TreeProcessor(lark.visitors.Interpreter): extnet = "(only triggers)" if triggers or compiled_extra_triggers: if extnet_id: - if not self.__state.options.cup_extranetworktags: + if not self.__state.options.cup_extranetwork_tags: self.result += " " else: self.result += ", " @@ -950,14 +932,12 @@ class TreeProcessor(lark.visitors.Interpreter): filter_object = tree.children[1].children[1] if tree.children[1] is not None else None if filter_object is None: for wc in selected_wildcards: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Removed default filter for wildcard '{escape_single_quotes(wc)}'") + self.log(logging.DEBUG, f"Removed default filter for wildcard '{escape_single_quotes(wc)}'") self.__state.wildcards_obj.set_wildcard_default_filter(wc, None) else: filter_specifier = self.__extract_filter_specifiers(filter_object) for wc in selected_wildcards: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Set default filter for wildcard '{escape_single_quotes(wc)}'") + self.log(logging.DEBUG, f"Set default filter for wildcard '{escape_single_quotes(wc)}'") self.__state.wildcards_obj.set_wildcard_default_filter(wc, filter_specifier) t2 = time.monotonic_ns() self.__debug_end("commandsetwcdeffilter", start_result, t2 - t1) @@ -968,7 +948,7 @@ class TreeProcessor(lark.visitors.Interpreter): """ t1 = time.monotonic_ns() start_result = self.result - if not self.__state.options.rem_removeextranetworktags: + if not self.__state.options.cup_remove_extranetwork_tags: self.result += f"<{tree.children[0]}" self.__visit(tree.children[1]) self.result += ">" @@ -1035,11 +1015,8 @@ class TreeProcessor(lark.visitors.Interpreter): ) continue self.__seen_wildcards.append(wc.key) - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Seen wildcard '{escape_single_quotes(wc.key)}'") - self.__state.logger.debug( - f"Including choices from wildcard '{escape_single_quotes(wc.key)}'" - ) + self.log(logging.DEBUG, f"Seen wildcard '{escape_single_quotes(wc.key)}'") + self.log(logging.DEBUG, f"Including choices from wildcard '{escape_single_quotes(wc.key)}'") (_, choice_values) = self.__check_wildcard_initialization(wc) if choice_values is not None: ch_values = self.__get_choices_internal_get(choice_values, None, wc.key) @@ -1087,7 +1064,7 @@ class TreeProcessor(lark.visitors.Interpreter): else: from_value: int = options.get("from", 1) to_value: int = options.get("to", 1) - separator: str = options.get("separator", self.__state.options.wil_choice_separator) + separator: str = options.get("separator", self.__state.options.choice_separator) msg_where = f"wildcard '{escape_single_quotes(wildcard_key)}'" if wildcard_key else "choices" if sampler != "~": self.warn_or_stop(f"Unsupported sampler '{escape_single_quotes(sampler)}' at {msg_where} options!") @@ -1132,21 +1109,19 @@ class TreeProcessor(lark.visitors.Interpreter): self.warn_or_stop(f"Not enough choices found for {msg_where}!") if num_choices < 2: repeating = False - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug( - format_output( - f"Selecting {'optional ' if optional else ''}{'repeating ' if repeating else ''}{num_choices} choice" - + ("s" if num_choices != 1 else "") - + (f" and separating with '{escape_single_quotes(separator)}'" if num_choices > 1 else "") - ) - ) + self.log( + logging.DEBUG, + f"Selecting {'optional ' if optional else ''}{'repeating ' if repeating else ''}{num_choices} choice" + + ("s" if num_choices != 1 else "") + + (f" and separating with '{escape_single_quotes(separator)}'" if num_choices > 1 else ""), + ) if num_choices > 0: selected_choices: list[dict] = ( list(self.__rng.choice(available_choices, size=num_choices, p=weights, replace=repeating)) if available_choices else [] ) - if self.__state.options.wil_keep_choices_order: + if self.__state.options.keep_choices_order: selected_choices = sorted(selected_choices, key=lambda x: x["choice_index"]) selected_choices_text = [] prefix: str = ( @@ -1164,11 +1139,11 @@ class TreeProcessor(lark.visitors.Interpreter): else: choice_content = self.__visit(choice_content_obj, False, True) t2 = time.monotonic_ns() - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug( - f"Adding choice {i+1} ({(t2-t1) / 1_000_000_000:.3f} seconds):\n" - + textwrap.indent(re.sub(r"\n$", "", choice_content), " ") - ) + self.log( + logging.DEBUG, + f"Adding choice {i+1} ({(t2-t1) / 1_000_000_000:.3f} seconds):\n" + + textwrap.indent(re.sub(r"\n$", "", choice_content), " "), + ) selected_choices_text.append(choice_content) suffix: str = ( self.__visit(options.get("suffix", None), False, True) @@ -1183,9 +1158,11 @@ class TreeProcessor(lark.visitors.Interpreter): prefix = "" suffix = "" results = [] - if self.__debug_level == DEBUG_LEVEL.full: - list_unseen = [f"'{escape_single_quotes(x)}'" for x in self.__seen_wildcards[seen_wildcards_len:]] - self.__state.logger.debug(f"Unseen wildcards: {', '.join(list_unseen)}") + self.log( + logging.DEBUG, + "Unseen wildcards: " + + ", ".join([f"'{escape_single_quotes(x)}'" for x in self.__seen_wildcards[seen_wildcards_len:]]), + ) self.__seen_wildcards = self.__seen_wildcards[:seen_wildcards_len] return (prefix, results, separator, suffix) @@ -1319,8 +1296,7 @@ class TreeProcessor(lark.visitors.Interpreter): ) cv["content"] = None if cv["content"] is not None: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Processed choice {cv}") + self.log(logging.DEBUG, f"Processed choice {cv}") choice_values.append(cv) else: self.warn_or_stop( @@ -1342,10 +1318,10 @@ class TreeProcessor(lark.visitors.Interpreter): ) wildcard.choices = choice_values t2 = time.monotonic_ns() - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug( - f"Processed choices for wildcard '{escape_single_quotes(wildcard.key)}' ({(t2-t1) / 1_000_000_000:.3f} seconds)" - ) + self.log( + logging.DEBUG, + f"Processed choices for wildcard '{escape_single_quotes(wildcard.key)}' ({(t2-t1) / 1_000_000_000:.3f} seconds)", + ) return (self.__clean_wildcard_options(options), choice_values) def get_wildcard_options(self, wildcard: PPPWildcard) -> tuple[dict | None, int]: @@ -1394,8 +1370,8 @@ class TreeProcessor(lark.visitors.Interpreter): n = 1 except lark.exceptions.UnexpectedInput: options = None - if options is None and self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Does not have options") + if options is None: + self.log(logging.DEBUG, "Does not have options") wildcard.options = options return options, n @@ -1416,9 +1392,8 @@ class TreeProcessor(lark.visitors.Interpreter): applied_options = self.__clean_wildcard_options(self.__convert_choices_options(tree.children[0], False)) wildcard_key: str = self.__visit(tree.children[1], False, True) wc = self.__get_original_node_content(tree, f"?__{wildcard_key}__") - if self.__state.options.wil_process_wildcards: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Processing wildcard: {wildcard_key}") + if self.__state.options.process_wildcards: + self.log(logging.DEBUG, f"Processing wildcard: {wildcard_key}") selected_wildcards = self.__state.wildcards_obj.get_wildcards(wildcard_key) if not selected_wildcards: self.detectedWildcards.append(wc) @@ -1436,32 +1411,29 @@ class TreeProcessor(lark.visitors.Interpreter): ): filter_wildcard_key = self.__visit(filter_object.children[2], False, True) filter_specifier = self.__wildcard_filters.get(filter_wildcard_key, None) - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Filtering choices with inherited filter") + self.log(logging.DEBUG, "Filtering choices with inherited filter") else: filter_specifier = self.__extract_filter_specifiers(filter_object.children[2]) - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Filtering choices") + self.log(logging.DEBUG, "Filtering choices") self.__wildcard_filters[wildcard_key] = filter_specifier if filter_object.children[1] is not None and "#" in str( filter_object.children[1] ): # means do not use the filter in this wildcard - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Ignoring filter") + self.log(logging.DEBUG, "Ignoring filter") filter_specifier = None else: filter_specifier = self.__state.wildcards_obj.get_wildcard_default_filter(wildcard_key) if filter_specifier is not None: self.__wildcard_filters[wildcard_key] = filter_specifier - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Applying default filter") + self.log(logging.DEBUG, "Applying default filter") if ( len(selected_wildcards) > 1 and filter_specifier is not None and any(y[0].isdecimal() for x in filter_specifier for y in x) ): - self.__state.logger.warning( - f"Using a globbing wildcard '{escape_single_quotes(wildcard_key)}' with positional index filters is not recommended!" + self.log( + logging.WARNING, + f"Using a globbing wildcard '{escape_single_quotes(wildcard_key)}' with positional index filters is not recommended!", ) var_object = tree.children[3] variablename = None @@ -1486,17 +1458,15 @@ class TreeProcessor(lark.visitors.Interpreter): ) continue self.__seen_wildcards.append(wildcard.key) - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(f"Seen wildcard '{escape_single_quotes(wildcard.key)}'") + self.log(logging.DEBUG, f"Seen wildcard '{escape_single_quotes(wildcard.key)}'") (options, choice_values) = self.__check_wildcard_initialization(wildcard) if options is not None: if applied_options is None: applied_options = options else: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug( - f"Options for wildcard '{escape_single_quotes(wildcard.key)}' are ignored!" - ) + self.log( + logging.DEBUG, f"Options for wildcard '{escape_single_quotes(wildcard.key)}' are ignored!" + ) choice_values_all += choice_values self.result += self.__get_choices(applied_options, choice_values_all, filter_specifier, wildcard_key) if wildcard_key in self.__wildcard_filters: @@ -1505,12 +1475,12 @@ class TreeProcessor(lark.visitors.Interpreter): self.__remove_user_variable(variablename) if variablebackup is not None: self.__state.user_variables[variablename] = variablebackup - elif self.__state.options.wil_ifwildcards != IFWILDCARDS_CHOICES.remove: + elif self.__state.options.if_wildcards != IFWILDCARDS_CHOICES.remove: self.detectedWildcards.append(wc) self.result += wc if self.__debug_level == DEBUG_LEVEL.full: list_unseen = [f"'{escape_single_quotes(x)}'" for x in self.__seen_wildcards[seen_wildcards_len:]] - self.__state.logger.debug(f"Unseen wildcards: {', '.join(list_unseen)}") + self.log(logging.DEBUG, f"Unseen wildcards: {', '.join(list_unseen)}") self.__seen_wildcards = self.__seen_wildcards[:seen_wildcards_len] t2 = time.monotonic_ns() self.__debug_end("wildcard", start_result, t2 - t1, f"'{escape_single_quotes(wc)}'") @@ -1538,11 +1508,10 @@ class TreeProcessor(lark.visitors.Interpreter): options = self.__convert_choices_options(tree.children[0], False) choice_values = [self.__convert_choice(c) for c in tree.children[1::]] ch = self.__get_original_node_content(tree, "?{...}") - if self.__state.options.wil_process_wildcards: - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug("Processing choices:") + if self.__state.options.process_wildcards: + self.log(logging.DEBUG, "Processing choices:") self.result += self.__get_choices(options, choice_values) - elif self.__state.options.wil_ifwildcards != IFWILDCARDS_CHOICES.remove: + elif self.__state.options.if_wildcards != IFWILDCARDS_CHOICES.remove: self.detectedWildcards.append(ch) self.result += ch t2 = time.monotonic_ns() @@ -1562,7 +1531,7 @@ class TreeProcessor(lark.visitors.Interpreter): attention_processing = self.__state.host_config.get("attention", "ok") # process the found negative tags for negtag in self.__negtags: - if self.__state.options.cup_mergeattention: + if self.__state.options.cup_merge_attention: # join consecutive attention elements for i in range(len(negtag.shell) - 1, 0, -1): if negtag.shell[i].type == "at" and negtag.shell[i - 1].type == "at": @@ -1615,8 +1584,7 @@ class TreeProcessor(lark.visitors.Interpreter): if content not in self.__already_processed: if self.__state.options.stn_ignore_repeats: self.__already_processed.append(content) - if self.__debug_level == DEBUG_LEVEL.full: - self.__state.logger.debug(format_output(f"Adding content at position {position}: {content}")) + self.log(logging.DEBUG, f"Adding content at position {position}: {content}") if position == "e": self.add_at["end"].append(content) elif position.startswith("p"): @@ -1625,6 +1593,6 @@ class TreeProcessor(lark.visitors.Interpreter): else: # position == "s" or invalid self.add_at["start"].append(content) else: - self.__state.logger.warning(format_output(f"Ignoring repeated content: {content}")) + self.log(logging.WARNING, f"Ignoring repeated content: {content}") t2 = time.monotonic_ns() self.__debug_end("start", "", t2 - t1) diff --git a/ppp_wildcards.py b/ppp_wildcards.py index 08dbc75..6f89af1 100644 --- a/ppp_wildcards.py +++ b/ppp_wildcards.py @@ -4,7 +4,7 @@ from typing import Optional import logging import yaml -from ppp_logging import DEBUG_LEVEL +from ppp_logging import DEBUG_LEVEL, log from ppp_utils import deep_freeze, escape_single_quotes @@ -77,8 +77,7 @@ class PPPWildcards: """ self.__debug_level = debug_level self.__wildcards_folders = wildcards_folders or [] - # if self.__debug_level != DEBUG_LEVEL.none and self.__logger: - # self.__logger.info("Refreshing wildcards...") + # log(self.__logger, self.__debug_level, logging.INFO, "Refreshing wildcards...") # t1 = time.monotonic_ns() for fullpath in list(self.__wildcard_files.keys()): if fullpath != self.LOCALINPUT_FILENAME: @@ -108,8 +107,7 @@ class PPPWildcards: self.wildcards = {} self.__wildcard_files = {} # t2 = time.monotonic_ns() - # if self.__debug_level != DEBUG_LEVEL.none and self.__logger: - # self.__logger.info(f"Wildcards refresh time: {(t2 - t1) / 1_000_000_000:.3f} seconds") + # log(self.__logger, self.__debug_level, logging.INFO, f"Wildcards refresh time: {(t2 - t1) / 1_000_000_000:.3f} seconds") def get_wildcards(self, key: str) -> list[PPPWildcard]: """ @@ -171,11 +169,16 @@ class PPPWildcards: debug (bool): Whether to print debug messages or not. """ last_modified_cached = self.__wildcard_files.get(full_path, None) # a time or a hash - if debug and last_modified_cached is not None and self.__debug_level != DEBUG_LEVEL.none and self.__logger: + if debug and last_modified_cached is not None: if full_path == self.LOCALINPUT_FILENAME: - self.__logger.debug("Removing from memory wildcards from input") + log(self.__logger, self.__debug_level, logging.DEBUG, "Removing from memory wildcards from input") else: - self.__logger.debug(f"Removing from memory wildcards from file: {full_path}") + log( + self.__logger, + self.__debug_level, + logging.DEBUG, + f"Removing from memory wildcards from file: {full_path}", + ) if full_path in self.__wildcard_files.keys(): del self.__wildcard_files[full_path] for key in list(self.wildcards.keys()): @@ -200,16 +203,20 @@ class PPPWildcards: if extension not in (".txt", ".json", ".yaml", ".yml"): return self.__remove_wildcards_from_path(full_path, False) - if last_modified_cached is not None and self.__debug_level != DEBUG_LEVEL.none and self.__logger: - self.__logger.debug(f"Updating wildcards from file: {full_path}") + if last_modified_cached is not None: + log(self.__logger, self.__debug_level, logging.DEBUG, f"Updating wildcards from file: {full_path}") if extension == ".txt": self.__get_wildcards_in_text_file(full_path, base) elif extension in (".json", ".yaml", ".yml"): self.__get_wildcards_in_structured_file(full_path, base) self.__wildcard_files[full_path] = last_modified except Exception as e: # pylint: disable=broad-except - if self.__logger: - self.__logger.error(f"Error reading wildcard file '{escape_single_quotes(full_path)}': {e}") + log( + self.__logger, + self.__debug_level, + logging.ERROR, + f"Error reading wildcard file '{escape_single_quotes(full_path)}': {e}", + ) def __get_wildcards_in_input(self, wildcards_input: str): """ @@ -224,22 +231,20 @@ class PPPWildcards: if h == new_h: return self.__remove_wildcards_from_path(self.LOCALINPUT_FILENAME, False) - if h is not None and self.__debug_level != DEBUG_LEVEL.none and self.__logger: - self.__logger.debug("Updating wildcards from input") + if h is not None: + log(self.__logger, self.__debug_level, logging.DEBUG, "Updating wildcards from input") wildcards_input = wildcards_input.strip() if wildcards_input != "": try: content = yaml.safe_load(wildcards_input) except yaml.YAMLError as e: - if self.__logger: - self.__logger.warning(f"Invalid format for input wildcards: {e}") + log(self.__logger, self.__debug_level, logging.WARNING, f"Invalid format for input wildcards: {e}") return if content is not None: self.__add_wildcard(content, self.LOCALINPUT_FILENAME, [self.LOCALINPUT_FILENAME]) self.__wildcard_files[self.LOCALINPUT_FILENAME] = new_h except Exception as e: # pylint: disable=broad-except - if self.__logger: - self.__logger.error(f"Error reading wildcards input: {e}") + log(self.__logger, self.__debug_level, logging.ERROR, f"Error reading wildcards input: {e}") # NOTE wcdef and choice options should not have properties in common @@ -301,8 +306,11 @@ class PPPWildcards: if isinstance(obj, (int, float, bool)): return [str(obj)] if not isinstance(obj, list) or len(obj) == 0: - self.__logger.warning( - f"Invalid format in wildcard '{escape_single_quotes('/'.join(key_parts))}' in file '{escape_single_quotes(full_path)}'!" + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid format in wildcard '{escape_single_quotes('/'.join(key_parts))}' in file '{escape_single_quotes(full_path)}'!", ) return None choices = [] @@ -315,8 +323,11 @@ class PPPWildcards: elif isinstance(c, dict): choices.append(self.__process_dict_choice(c, full_path, key_parts, i)) else: - self.__logger.warning( - f"Invalid choice {i+1} in wildcard '{escape_single_quotes('/'.join(key_parts))}' in file '{escape_single_quotes(full_path)}'!" + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid choice {i+1} in wildcard '{escape_single_quotes('/'.join(key_parts))}' in file '{escape_single_quotes(full_path)}'!", ) return choices @@ -349,10 +360,12 @@ class PPPWildcards: # we assume it is an anonymous wildcard with options firstkey = list(c.keys())[0] return self.__create_anonymous_wildcard(full_path, key_parts, i, c[firstkey], firstkey) - if self.__logger: - self.__logger.warning( - f"Invalid choice {i+1} in wildcard '{escape_single_quotes('/'.join(key_parts))}' in file '{escape_single_quotes(full_path)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid choice {i+1} in wildcard '{escape_single_quotes('/'.join(key_parts))}' in file '{escape_single_quotes(full_path)}'!", + ) return None def __create_anonymous_wildcard(self, full_path, key_parts, i, content, options=None): @@ -394,23 +407,29 @@ class PPPWildcards: tmp_key_parts.extend(key.split("/")) fullkey = "/".join(tmp_key_parts) if self.wildcards.get(fullkey, None) is not None: - if self.__logger: - self.__logger.warning( - f"Duplicate wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}' and '{escape_single_quotes(self.wildcards[fullkey].file)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Duplicate wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}' and '{escape_single_quotes(self.wildcards[fullkey].file)}'!", + ) else: obj = self.__get_nested(content, key) choices = self.__get_choices(obj, full_path, tmp_key_parts) if choices is None: - if self.__logger: - self.__logger.warning( - f"Invalid wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'!", + ) elif fullkey.startswith("_"): - if self.__logger: - self.__logger.warning( - f"Invalid wildcard name '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'! (cannot start with underscore)" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid wildcard name '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'! (cannot start with underscore)", + ) else: self.wildcards[fullkey] = PPPWildcard(full_path, fullkey, choices) return @@ -419,27 +438,37 @@ class PPPWildcards: elif isinstance(content, (int, float, bool)): content = [str(content)] if not isinstance(content, list): - if self.__logger: - self.__logger.warning(f"Invalid wildcard in file '{escape_single_quotes(full_path)}'!") + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid wildcard in file '{escape_single_quotes(full_path)}'!", + ) return fullkey = "/".join(key_parts) if self.wildcards.get(fullkey, None) is not None: - if self.__logger: - self.__logger.warning( - f"Duplicate wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}' and '{escape_single_quotes(self.wildcards[fullkey].file)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Duplicate wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}' and '{escape_single_quotes(self.wildcards[fullkey].file)}'!", + ) else: choices = self.__get_choices(content, full_path, key_parts) if choices is None: - if self.__logger: - self.__logger.warning( - f"Invalid wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'!" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid wildcard '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'!", + ) elif fullkey.startswith("_"): - if self.__logger: - self.__logger.warning( - f"Invalid wildcard name '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'! (cannot start with underscore)" - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Invalid wildcard name '{escape_single_quotes(fullkey)}' in file '{escape_single_quotes(full_path)}'! (cannot start with underscore)", + ) else: self.wildcards[fullkey] = PPPWildcard(full_path, fullkey, choices) @@ -457,10 +486,12 @@ class PPPWildcards: with open(full_path, "r", encoding="utf-8") as file: content = yaml.safe_load(file) except: # pylint: disable=bare-except - if self.__logger: - self.__logger.warning( - f"Could not read file '{escape_single_quotes(full_path)}' with utf-8 encoding, trying windows-1252..." - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Could not read file '{escape_single_quotes(full_path)}' with utf-8 encoding, trying windows-1252...", + ) with open(full_path, "r", encoding="windows-1252") as file: content = yaml.safe_load(file) self.__add_wildcard(content, full_path, external_key_parts) @@ -479,10 +510,12 @@ class PPPWildcards: with open(full_path, "r", encoding="utf-8") as file: text_content = map(lambda x: x.strip("\n\r"), file.readlines()) except: # pylint: disable=bare-except - if self.__logger: - self.__logger.warning( - f"Could not read file '{escape_single_quotes(full_path)}' with utf-8 encoding, trying windows-1252..." - ) + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Could not read file '{escape_single_quotes(full_path)}' with utf-8 encoding, trying windows-1252...", + ) with open(full_path, "r", encoding="windows-1252") as file: text_content = map(lambda x: x.strip("\n\r"), file.readlines()) text_content = list(filter(lambda x: x.strip() != "" and not x.strip().startswith("#"), text_content)) @@ -498,8 +531,12 @@ class PPPWildcards: directory (str): The path to the directory. """ if not os.path.exists(directory): - if self.__logger: - self.__logger.warning(f"Wildcard directory '{escape_single_quotes(directory)}' does not exist!") + log( + self.__logger, + self.__debug_level, + logging.WARNING, + f"Wildcard directory '{escape_single_quotes(directory)}' does not exist!", + ) return for filename in os.listdir(directory): full_path = os.path.abspath(os.path.join(directory, filename)) diff --git a/scripts/ppp_script.py b/scripts/ppp_script.py index 319617d..5cf5b03 100644 --- a/scripts/ppp_script.py +++ b/scripts/ppp_script.py @@ -1,6 +1,7 @@ if __name__ == "__main__": raise SystemExit("This script must be run from a Stable Diffusion WebUI") +import logging import sys import os import time @@ -16,7 +17,7 @@ from modules.paths import models_path # pylint: disable=import-error import gradio as gr # pylint: disable=import-error from ppp import PromptPostProcessor from ppp_classes import IFWILDCARDS_CHOICES, ONWARNING_CHOICES, SUPPORTED_APPS, SUPPORTED_APPS_NAMES, PPPStateOptions -from ppp_logging import DEBUG_LEVEL, PromptPostProcessorLogFactory +from ppp_logging import DEBUG_LEVEL, PromptPostProcessorLogFactory, log from ppp_cache import PPPLRUCache from ppp_wildcards import PPPWildcards from ppp_enmappings import PPPExtraNetworkMappings @@ -177,32 +178,30 @@ class PromptPostProcessorA1111Script(scripts.Script): ) options = PPPStateOptions( debug_level=DEBUG_LEVEL(getattr(opts, "ppp_gen_debug_level", PromptPostProcessor.DEFAULT_DEBUG_LEVEL)), - gen_onwarning=ONWARNING_CHOICES(getattr(opts, "ppp_gen_onwarning", PromptPostProcessor.DEFAULT_ONWARNING)), - wil_process_wildcards=getattr(opts, "ppp_wil_processwildcards", PromptPostProcessor.DEFAULT_WC_PROCESS), - wil_ifwildcards=IFWILDCARDS_CHOICES( + on_warning=ONWARNING_CHOICES(getattr(opts, "ppp_gen_onwarning", PromptPostProcessor.DEFAULT_ON_WARNING)), + process_wildcards=getattr(opts, "ppp_wil_processwildcards", PromptPostProcessor.DEFAULT_PROCESS_WILDCARDS), + if_wildcards=IFWILDCARDS_CHOICES( getattr(opts, "ppp_wil_ifwildcards", PromptPostProcessor.DEFAULT_IF_WILDCARDS) ), - wil_choice_separator=getattr( - opts, "ppp_wil_choice_separator", PromptPostProcessor.DEFAULT_CHOICE_SEPARATOR - ), - wil_keep_choices_order=getattr( + choice_separator=getattr(opts, "ppp_wil_choice_separator", PromptPostProcessor.DEFAULT_CHOICE_SEPARATOR), + keep_choices_order=getattr( opts, "ppp_wil_keep_choices_order", PromptPostProcessor.DEFAULT_KEEP_CHOICES_ORDER ), stn_separator=getattr(opts, "ppp_stn_separator", PromptPostProcessor.DEFAULT_STN_SEPARATOR), stn_ignore_repeats=getattr(opts, "ppp_stn_ignorerepeats", PromptPostProcessor.DEFAULT_STN_IGNORE_REPEATS), cup_do_cleanup=True, cup_cleanup_variables=True, - cup_extraspaces=getattr(opts, "ppp_cup_extraspaces", PromptPostProcessor.DEFAULT_CUP_EXTRA_SPACES), - cup_emptyconstructs=getattr( + cup_extra_spaces=getattr(opts, "ppp_cup_extraspaces", PromptPostProcessor.DEFAULT_CUP_EXTRA_SPACES), + cup_empty_constructs=getattr( opts, "ppp_cup_emptyconstructs", PromptPostProcessor.DEFAULT_CUP_EMPTY_CONSTRUCTS ), - cup_extraseparators=getattr( + cup_extra_separators=getattr( opts, "ppp_cup_extraseparators", PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS ), - cup_extraseparators2=getattr( + cup_extra_separators2=getattr( opts, "ppp_cup_extraseparators2", PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS2 ), - cup_extraseparators_include_eol=getattr( + cup_extra_separators_include_eol=getattr( opts, "ppp_cup_extraseparators_include_eol", PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS_INCLUDE_EOL, @@ -211,11 +210,13 @@ class PromptPostProcessorA1111Script(scripts.Script): cup_breaks_eol=getattr(opts, "ppp_cup_breaks_eol", PromptPostProcessor.DEFAULT_CUP_BREAKS_EOL), cup_ands=getattr(opts, "ppp_cup_ands", PromptPostProcessor.DEFAULT_CUP_ANDS), cup_ands_eol=getattr(opts, "ppp_cup_ands_eol", PromptPostProcessor.DEFAULT_CUP_ANDS_EOL), - cup_extranetworktags=getattr( + cup_extranetwork_tags=getattr( opts, "ppp_cup_extranetworktags", PromptPostProcessor.DEFAULT_CUP_EXTRANETWORK_TAGS ), - cup_mergeattention=getattr(opts, "ppp_cup_mergeattention", PromptPostProcessor.DEFAULT_CUP_MERGE_ATTENTION), - rem_removeextranetworktags=getattr( + cup_merge_attention=getattr( + opts, "ppp_cup_mergeattention", PromptPostProcessor.DEFAULT_CUP_MERGE_ATTENTION + ), + cup_remove_extranetwork_tags=getattr( opts, "ppp_rem_removeextranetworktags", PromptPostProcessor.DEFAULT_CUP_REMOVE_EXTRANETWORK_TAGS ), ) @@ -226,19 +227,21 @@ class PromptPostProcessorA1111Script(scripts.Script): self.lru_cache = PPPLRUCache(1000, logger=self.ppp_logger, debug_level=self.ppp_debug_level) self.wildcards_obj = PPPWildcards(self.ppp_logger) self.extranetwork_mappings_obj = PPPExtraNetworkMappings(self.ppp_logger) - self.ppp_logger.info( - f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized, running on {SUPPORTED_APPS_NAMES[app]}" + log( + self.ppp_logger, + DEBUG_LEVEL.minimal, + logging.INFO, + f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized, running on {SUPPORTED_APPS_NAMES[app]}", ) t1 = time.monotonic_ns() if getattr(opts, "prompt_attention", "") == "Compel parser": - self.ppp_logger.warning("Compel parser is not supported!") + log(self.ppp_logger, self.ppp_debug_level, logging.WARNING, "Compel parser is not supported!") init_images = getattr(p, "init_images", [None]) or [None] is_i2i = bool(init_images[0]) do_i2i = getattr(opts, "ppp_gen_doi2i", False) add_prompts = getattr(opts, "ppp_gen_addpromptstometadata", True) if is_i2i and not do_i2i: - if self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info("Not processing the prompt for i2i") + log(self.ppp_logger, self.ppp_debug_level, logging.INFO, "Not processing the prompt for i2i") return p.extra_generation_params.update( @@ -250,8 +253,12 @@ class PromptPostProcessorA1111Script(scripts.Script): } ) - if self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info(f"Post-processing prompts ({'i2i' if is_i2i else 't2i'})") + log( + self.ppp_logger, + self.ppp_debug_level, + logging.INFO, + f"Post-processing prompts ({'i2i' if is_i2i else 't2i'})", + ) env_info = { "app": app.value, "models_path": models_path, @@ -281,7 +288,7 @@ class PromptPostProcessorA1111Script(scripts.Script): if f.strip() != "" ] self.wildcards_obj.refresh_wildcards( - self.ppp_debug_level, wildcards_folders if options.wil_process_wildcards else None + self.ppp_debug_level, wildcards_folders if options.process_wildcards else None ) self.extranetwork_mappings_obj.refresh_extranetwork_mappings(self.ppp_debug_level, enmappings_folders) ppp = PromptPostProcessor( @@ -298,16 +305,14 @@ class PromptPostProcessorA1111Script(scripts.Script): prompts_list = [] if input_force_equal_seeds: - if self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info("Forcing equal seeds") + log(self.ppp_logger, self.ppp_debug_level, logging.INFO, "Forcing equal seeds") seeds = getattr(p, "all_seeds", []) subseeds = getattr(p, "all_subseeds", []) p.all_seeds = [seeds[0] for _ in seeds] p.all_subseeds = [subseeds[0] for _ in subseeds] if input_unlink_seed: - if self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info("Using unlinked seed") + log(self.ppp_logger, self.ppp_debug_level, logging.INFO, "Using unlinked seed") num_seeds = len(getattr(p, "all_seeds", [])) if input_incremental_seed: first_seed = np.random.randint(0, 2**32, dtype=np.int64) if input_seed == -1 else input_seed @@ -355,8 +360,7 @@ class PromptPostProcessorA1111Script(scripts.Script): # processes prompts for i, (prompttype, seed, prompt, negative_prompt) in enumerate(prompts_list): - if self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info(f"processing prompts[{i+1}] ({prompttype})") + log(self.ppp_logger, self.ppp_debug_level, logging.INFO, f"processing prompts[{i+1}] ({prompttype})") if ( self.lru_cache.get( (hash_envinfo, hash_options, seed, hash(self.wildcards_obj), prompt, negative_prompt) @@ -371,8 +375,8 @@ class PromptPostProcessorA1111Script(scripts.Script): self.lru_cache.put( (hash_envinfo, hash_options, seed, hash(self.wildcards_obj), posp, negp), (posp, negp) ) - elif self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info("result already in cache") + else: + log(self.ppp_logger, self.ppp_debug_level, logging.INFO, "result already in cache") # updates the prompts rpr_copy = None @@ -426,8 +430,12 @@ class PromptPostProcessorA1111Script(scripts.Script): p.extra_generation_params[k] = v t2 = time.monotonic_ns() - if self.ppp_debug_level != DEBUG_LEVEL.none: - self.ppp_logger.info(f"process time: {(t2 - t1) / 1_000_000_000:.3f} seconds") + log( + self.ppp_logger, + self.ppp_debug_level, + logging.INFO, + f"process time: {(t2 - t1) / 1_000_000_000:.3f} seconds", + ) def ppp_interrupt(self): """ diff --git a/tests/base_tests.py b/tests/base_tests.py index 8ec08cd..acee10d 100644 --- a/tests/base_tests.py +++ b/tests/base_tests.py @@ -45,27 +45,27 @@ class TestPromptPostProcessorBase(unittest.TestCase): self.interrupted = False self.defopts = PPPStateOptions( debug_level=DEBUG_LEVEL.full, - gen_onwarning=ONWARNING_CHOICES.stop, - wil_process_wildcards=True, - wil_ifwildcards=IFWILDCARDS_CHOICES.ignore, - wil_choice_separator=", ", - wil_keep_choices_order=False, + on_warning=ONWARNING_CHOICES.stop, + process_wildcards=True, + if_wildcards=IFWILDCARDS_CHOICES.ignore, + choice_separator=", ", + keep_choices_order=False, stn_separator=", ", stn_ignore_repeats=True, cup_do_cleanup=True, cup_cleanup_variables=True, - cup_emptyconstructs=True, - cup_extraseparators=True, - cup_extraseparators2=True, - cup_extraseparators_include_eol=False, - cup_extraspaces=True, + cup_empty_constructs=True, + cup_extra_separators=True, + cup_extra_separators2=True, + cup_extra_separators_include_eol=False, + cup_extra_spaces=True, cup_breaks=True, cup_breaks_eol=False, cup_ands=True, cup_ands_eol=False, - cup_extranetworktags=True, - cup_mergeattention=True, - rem_removeextranetworktags=False, + cup_extranetwork_tags=True, + cup_merge_attention=True, + cup_remove_extranetwork_tags=False, ) self.def_env_info = { "app": "tests", @@ -137,17 +137,17 @@ class TestPromptPostProcessorBase(unittest.TestCase): self.defopts, cup_do_cleanup=False, cup_cleanup_variables=False, - cup_emptyconstructs=False, - cup_extraseparators=False, - cup_extraseparators2=False, - cup_extraseparators_include_eol=False, - cup_extraspaces=False, + cup_empty_constructs=False, + cup_extra_separators=False, + cup_extra_separators2=False, + cup_extra_separators_include_eol=False, + cup_extra_spaces=False, cup_breaks=False, cup_breaks_eol=False, cup_ands=False, cup_ands_eol=False, - cup_extranetworktags=False, - cup_mergeattention=False, + cup_extranetwork_tags=False, + cup_merge_attention=False, ), self.grammar_content, self.interrupt, diff --git a/tests/tests_choices.py b/tests/tests_choices.py index 4f553b6..de5f3f2 100644 --- a/tests/tests_choices.py +++ b/tests/tests_choices.py @@ -86,7 +86,7 @@ class TestChoices(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - rem_removeextranetworktags=True, + cup_remove_extranetwork_tags=True, ), self.grammar_content, self.interrupt, diff --git a/tests/tests_cleanup.py b/tests/tests_cleanup.py index abf0c6f..8211771 100644 --- a/tests/tests_cleanup.py +++ b/tests/tests_cleanup.py @@ -42,7 +42,7 @@ class TestCleanup(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - rem_removeextranetworktags=True, + cup_remove_extranetwork_tags=True, ), self.grammar_content, self.interrupt, @@ -60,8 +60,8 @@ class TestCleanup(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - cup_extraseparators2=False, - cup_extraseparators_include_eol=False, + cup_extra_separators2=False, + cup_extra_separators_include_eol=False, ), self.grammar_content, self.interrupt, @@ -89,17 +89,17 @@ class TestCleanup(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - cup_emptyconstructs=False, - cup_extraseparators=True, - cup_extraseparators2=False, - cup_extraseparators_include_eol=False, - cup_extraspaces=False, + cup_empty_constructs=False, + cup_extra_separators=True, + cup_extra_separators2=False, + cup_extra_separators_include_eol=False, + cup_extra_spaces=False, cup_breaks=False, cup_breaks_eol=False, cup_ands=False, cup_ands_eol=False, - cup_extranetworktags=False, - cup_mergeattention=False, + cup_extranetwork_tags=False, + cup_merge_attention=False, ), self.grammar_content, self.interrupt, diff --git a/tests/tests_variables.py b/tests/tests_variables.py index 37e2ab9..d7e6ada 100644 --- a/tests/tests_variables.py +++ b/tests/tests_variables.py @@ -52,7 +52,7 @@ class TestVariables(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - gen_onwarning=ONWARNING_CHOICES.warn, + on_warning=ONWARNING_CHOICES.warn, ), self.grammar_content, self.interrupt, @@ -156,7 +156,7 @@ class TestVariables(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - gen_onwarning=ONWARNING_CHOICES.warn, + on_warning=ONWARNING_CHOICES.warn, ), self.grammar_content, self.interrupt, @@ -187,7 +187,7 @@ class TestVariables(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - gen_onwarning=ONWARNING_CHOICES.warn, + on_warning=ONWARNING_CHOICES.warn, ), self.grammar_content, self.interrupt, @@ -218,7 +218,7 @@ class TestVariables(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - gen_onwarning=ONWARNING_CHOICES.warn, + on_warning=ONWARNING_CHOICES.warn, ), self.grammar_content, self.interrupt, diff --git a/tests/tests_variants.py b/tests/tests_variants.py index 80115b0..444c839 100644 --- a/tests/tests_variants.py +++ b/tests/tests_variants.py @@ -65,7 +65,7 @@ class TestModelVariants(TestPromptPostProcessorBase): }, replace( self.defopts, - gen_onwarning=ONWARNING_CHOICES.warn, + on_warning=ONWARNING_CHOICES.warn, ), self.grammar_content, self.interrupt, diff --git a/tests/tests_wildcards.py b/tests/tests_wildcards.py index 3af4362..a304334 100644 --- a/tests/tests_wildcards.py +++ b/tests/tests_wildcards.py @@ -24,8 +24,8 @@ class TestWildcards(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - wil_process_wildcards=False, - wil_ifwildcards=IFWILDCARDS_CHOICES.ignore, + process_wildcards=False, + if_wildcards=IFWILDCARDS_CHOICES.ignore, ), self.grammar_content, self.interrupt, @@ -49,8 +49,8 @@ class TestWildcards(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - wil_process_wildcards=False, - wil_ifwildcards=IFWILDCARDS_CHOICES.remove, + process_wildcards=False, + if_wildcards=IFWILDCARDS_CHOICES.remove, ), self.grammar_content, self.interrupt, @@ -68,8 +68,8 @@ class TestWildcards(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - wil_process_wildcards=False, - wil_ifwildcards=IFWILDCARDS_CHOICES.warn, + process_wildcards=False, + if_wildcards=IFWILDCARDS_CHOICES.warn, ), self.grammar_content, self.interrupt, @@ -90,8 +90,8 @@ class TestWildcards(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - wil_process_wildcards=False, - wil_ifwildcards=IFWILDCARDS_CHOICES.stop, + process_wildcards=False, + if_wildcards=IFWILDCARDS_CHOICES.stop, ), self.grammar_content, self.interrupt, @@ -110,8 +110,8 @@ class TestWildcards(TestPromptPostProcessorBase): self.def_env_info, replace( self.defopts, - wil_process_wildcards=False, - wil_ifwildcards=IFWILDCARDS_CHOICES.warn, + process_wildcards=False, + if_wildcards=IFWILDCARDS_CHOICES.warn, ), self.grammar_content, self.interrupt,