zodiac

  1#  # # <!-- // /*  SPDX-License-Identifier: MPL-2.0  */ -->
  2#  # # <!-- // /*  d a r k s h a p e s */ -->
  3
  4import sys
  5import os
  6import argparse
  7import multiprocessing as mp
  8
  9mp.set_start_method("spawn", force=True)
 10sys.path.append(os.getcwd())
 11# import platform
 12
 13# if platform.system == "darwin":
 14#     import multiprocessing as mp
 15
 16#     mp.set_start_method("fork", force=True)
 17
 18
 19def start_trace():
 20    from viztracer import VizTracer
 21    from datetime import datetime
 22
 23    assembled_path = os.path.join("log", f".nnll{datetime.now().strftime('%Y%m%d')}_trace.json")
 24    os.makedirs("log", exist_ok=True)
 25    tracer = VizTracer()
 26    tracer.start()
 27    return tracer, assembled_path
 28
 29
 30def set_env(args: argparse.ArgumentParser):
 31    os.environ["TELEMETRY"] = "False"
 32    os.environ["TOKENIZERS_PARALLELISM"] = "false"
 33    try:
 34        import huggingface_hub
 35    except (ImportError, ModuleNotFoundError, Exception):  # pylint: disable=broad-exception-caught
 36        pass
 37    else:
 38        huggingface_hub.constants.HF_HUB_DISABLE_TELEMETRY = 1  # privacy
 39        huggingface_hub.constants.HF_HUB_DISABLE_IMPLICIT_TOKEN = 1
 40        huggingface_hub.constants.HF_XET_HIGH_PERFORMANCE = int(args.net or args.diag)  # download methods (if online)
 41        huggingface_hub.constants.HF_HUB_ENABLE_HF_TRANSFER = int(args.net or args.diag)
 42        huggingface_hub.constants.HF_HUB_DISABLE_PROGRESS_BARS = int(not args.diag)  # superficial/diagnostic
 43        huggingface_hub.constants.HF_HUB_OFFLINE = int(not args.net or not args.diag)  # -net = True -> hub offline = False/0
 44        os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"
 45        os.environ["HF_HUB_DISABLE_IMPLICIT_TOKEN"] = "1"
 46        os.environ["HF_XET_HIGH_PERFORMANCE"] = str(int(args.net or args.diag))
 47        os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = str(int(args.net or args.diag))
 48        os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = str(not args.diag)
 49        os.environ["HF_HUB_OFFLINE"] = str(int(not args.net or not args.diag))
 50
 51        os.environ["DISABLE_HF_TOKENIZER_DOWNLOAD"] = str(not args.net or not args.diag)  # litellm
 52        # huggingface_hub.constants.HF_HUB_VERBOSITY
 53
 54    try:
 55        import litellm
 56    except (ImportError, ModuleNotFoundError, Exception):  # pylint: disable=broad-exception-caught
 57        pass
 58    else:
 59        litellm.disable_token_counter = False
 60        litellm.disable_streaming_logging = True
 61        litellm.turn_off_message_logging = True
 62        litellm.suppress_debug_info = False
 63        litellm.json_logs = False  # type: ignore
 64        litellm.disable_end_user_cost_tracking = True
 65        litellm.telemetry = False
 66        litellm.disable_hf_tokenizer_download = not args.net  # -net = True -> disable download = False/0
 67        os.environ["DISABLE_END_USER_COST_TRACKING"] = "True"
 68    return True
 69
 70
 71def main() -> None:
 72    """Parse launch arguments (mostly turning down/disconnecting loud dependency packages)\n
 73    :param args: Launch arguments from command line
 74    """
 75
 76    parser = argparse.ArgumentParser(description="Multimodal generative media sequencer")
 77    parser.add_argument("-n", "--net", action="store_true", help="Allow network access (for downloading requirements)")
 78    parser.add_argument("-t", "--trace", action="store_true", help="Enable trace logs (generated in log folder)")
 79
 80    parser.add_argument("-d", "--diag", action="store_true", help="Process using diagnostic settings")
 81
 82    args = parser.parse_args()
 83
 84    parser = argparse.ArgumentParser(description="Multimodal generative media sequencer")
 85    parser.add_argument("-n", "--net", action="store_true", help="Allow network access (for downloading requirements)")
 86    parser.add_argument("-t", "--trace", action="store_true", help="Enable trace logs (generated in log folder)")
 87
 88    parser.add_argument("-d", "--diag", action="store_true", help="Process using diagnostic settings")
 89
 90    args = parser.parse_args()
 91
 92    env_ready = set_env(args)
 93    if env_ready:
 94        tracer, assembled_path = start_trace() if args.trace else None, None
 95        return tracer, assembled_path
 96
 97
 98if __name__ == "__main__":
 99    tracer, assembled_path = main()
100    if tracer and hasattr(tracer.stop):
101        tracer.stop()
102        tracer.save(output_file=assembled_path)