Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 87 additions & 30 deletions dimos/utils/logging_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,92 @@ def _configure_structlog() -> Path:
return _LOG_FILE_PATH


_CONSOLE_PATH_WIDTH = 30
_CONSOLE_USE_COLORS = hasattr(sys.stdout, "isatty") and sys.stdout.isatty()

_CONSOLE_LEVEL_COLORS = {
"dbg": "\033[1;36;40m", # bold cyan on black
"inf": "\033[1;32;40m", # bold green on black
"war": "\033[1;33;40m", # bold yellow on black
"err": "\033[1;31;40m", # bold red on black
"cri": "\033[1;31;40m", # bold red on black
}
_CONSOLE_RESET = "\033[0m"
_CONSOLE_FIXED = "\033[1;30;40m" # bold dark gray on black
_CONSOLE_TEXT = "\033[0;34m" # blue
_CONSOLE_KEY = "\033[0;36m" # cyan
_CONSOLE_VAL = "\033[0;35m" # magenta
_CONSOLE_EQ = "\033[0;37m" # white


def _compact_console_processor(logger: Any, method_name: str, event_dict: Mapping[str, Any]) -> str:
"""Format log lines as: HH:MM:SS.mmm[lvl][file.py ] Event key=value ..."""
event_dict = dict(event_dict)

# Time — HH:MM:SS.mmm
timestamp = event_dict.pop("timestamp", "")
if timestamp:
try:
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
time_str = dt.strftime("%H:%M:%S") + f".{dt.microsecond // 1000:03d}"
except (ValueError, AttributeError):
time_str = str(timestamp)[:12]
else:
now = datetime.now()
time_str = now.strftime("%H:%M:%S") + f".{now.microsecond // 1000:03d}"

# Level — 3-letter lowercase abbreviation
level = event_dict.pop("level", "???")
level_short = level[:3].lower()

# File path — fixed width, truncated from the left, padded on the right
file_path = event_dict.pop("logger", "")
if len(file_path) > _CONSOLE_PATH_WIDTH:
file_path = file_path[-_CONSOLE_PATH_WIDTH:]
file_path = f"{file_path:<{_CONSOLE_PATH_WIDTH}s}"

# Event message
event = event_dict.pop("event", "")

# Remove internal / callsite / exception fields
for key in (
"func_name",
"lineno",
"exception",
"exc_info",
"exception_type",
"exception_message",
"traceback_lines",
"_record",
"_from_structlog",
):
event_dict.pop(key, None)

# Assemble the line
if _CONSOLE_USE_COLORS:
R = _CONSOLE_RESET
color = _CONSOLE_LEVEL_COLORS.get(level_short, "")
line = (
f"{_CONSOLE_FIXED}{time_str}{R}"
f"{color}[{level_short}]{R}"
f"{_CONSOLE_FIXED}[{file_path}]{R} "
f"{_CONSOLE_TEXT}{event}{R}"
)
if event_dict:
kv_parts = " ".join(
f"{_CONSOLE_KEY}{k}{_CONSOLE_EQ}={_CONSOLE_VAL}{v}{R}"
for k, v in sorted(event_dict.items())
)
line += " " + kv_parts
else:
kv_str = " ".join(f"{k}={v}" for k, v in sorted(event_dict.items()))
line = f"{time_str} [{level_short}][{file_path}] {event}"
if kv_str:
line += " " + kv_str

return line


def setup_logger(*, level: int | None = None) -> Any:
"""Set up a structured logger using structlog.

Expand Down Expand Up @@ -140,39 +226,10 @@ def setup_logger(*, level: int | None = None) -> Any:
stdlib_logger.setLevel(level)
stdlib_logger.propagate = False

# Create console handler with pretty formatting.
# We use exception_formatter=None because we handle exceptions
# separately with Rich in the global exception handler

console_renderer = structlog.dev.ConsoleRenderer(
colors=True,
pad_event=60,
force_colors=False,
sort_keys=True,
# Don't format exceptions in console logs
exception_formatter=None, # type: ignore[arg-type]
)

# Wrapper to remove callsite info and exception details before rendering to console.
def console_processor_without_callsite(
logger: Any, method_name: str, event_dict: Mapping[str, Any]
) -> str:
event_dict = dict(event_dict)
# Remove callsite info
event_dict.pop("func_name", None)
event_dict.pop("lineno", None)
# Remove exception fields since we handle them with Rich
event_dict.pop("exception", None)
event_dict.pop("exc_info", None)
event_dict.pop("exception_type", None)
event_dict.pop("exception_message", None)
event_dict.pop("traceback_lines", None)
return console_renderer(logger, method_name, event_dict)

console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(level)
console_formatter = structlog.stdlib.ProcessorFormatter(
processor=console_processor_without_callsite,
processor=_compact_console_processor,
)
console_handler.setFormatter(console_formatter)
stdlib_logger.addHandler(console_handler)
Expand Down