Skip to content

Fix encoding issue #20443

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Nov 3, 2024
Merged
Show file tree
Hide file tree
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
19 changes: 16 additions & 3 deletions keras/src/utils/io_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,10 +91,18 @@ def set_logging_verbosity(level):

def print_msg(message, line_break=True):
"""Print the message to absl logging or stdout."""
message = str(message)
if is_interactive_logging_enabled():
if line_break:
sys.stdout.write(message + "\n")
else:
message = message + "\n" if line_break else message
try:
sys.stdout.write(message)
except UnicodeEncodeError:
# If the encoding differs from UTF-8, `sys.stdout.write` may fail.
# To address this, replace special unicode characters in the
# message, and then encode and decode using the target encoding.
message = _replace_special_unicode_character(message)
message_bytes = message.encode(sys.stdout.encoding, errors="ignore")
message = message_bytes.decode(sys.stdout.encoding)
sys.stdout.write(message)
sys.stdout.flush()
else:
Expand Down Expand Up @@ -123,3 +131,8 @@ def ask_to_proceed_with_overwrite(filepath):
return False
print_msg("[TIP] Next time specify overwrite=True!")
return True


def _replace_special_unicode_character(message):
message = str(message).replace("━", "=") # Fall back to Keras2 behavior.
return message
12 changes: 12 additions & 0 deletions keras/src/utils/io_utils_test.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import sys
import tempfile
from unittest.mock import patch

from keras.src.testing import test_case
Expand Down Expand Up @@ -55,3 +57,13 @@ def test_ask_to_proceed_with_overwrite_invalid_then_yes(self, _):
@patch("builtins.input", side_effect=["invalid", "n"])
def test_ask_to_proceed_with_overwrite_invalid_then_no(self, _):
self.assertFalse(io_utils.ask_to_proceed_with_overwrite("test_path"))

def test_print_msg_with_different_encoding(self):
# https://github.com/keras-team/keras/issues/19386
io_utils.enable_interactive_logging()
self.assertTrue(io_utils.is_interactive_logging_enabled())
ori_stdout = sys.stdout
with tempfile.TemporaryFile(mode="w", encoding="cp1251") as tmp:
sys.stdout = tmp
io_utils.print_msg("━")
sys.stdout = ori_stdout