zero width space markers and base64 encoded data in sgf files are probably terrible ideas
This commit is contained in:
1 parent
71d19a2c4e
commit
8dcfc1cd59
26 files changed
+153
-43
No files matched your search
@@ -1,3 +1,4 @@
|
||||
PROGRAM_NAME = "KaTrain"
|
||||
VERSION = "1.7.0"
|
||||
HOMEPAGE = "https://github.com/sanderland/katrain"
|
||||
CONFIG_MIN_VERSION = "1.7.0" # keep config files from this version
|
||||
@@ -237,3 +238,7 @@ TOP_MOVE_OPTIONS = [
|
||||
TOP_MOVE_VISITS,
|
||||
TOP_MOVE_NOTHING,
|
||||
]
|
||||
|
||||
SGF_COMMENTS_MARKER = "\u200b\u3164"
|
||||
SGF_NOTE_MARKER = "\u200b\u200b"
|
||||
SGF_SEPARATOR_MARKER = "\u3164\u3164"
|
||||
+27
-10
@@ -16,6 +16,7 @@ from katrain.core.constants import (
|
||||
STATUS_INFO,
|
||||
STATUS_TEACHING,
|
||||
PLAYER_HUMAN,
|
||||
VERSION, PROGRAM_NAME,
|
||||
)
|
||||
from katrain.core.engine import KataGoEngine
|
||||
from katrain.core.game_node import GameNode
|
||||
@@ -35,7 +36,7 @@ class KaTrainSGF(SGF):
|
||||
class Game:
|
||||
"""Represents a game of go, including an implementation of capture rules."""
|
||||
|
||||
DEFAULT_PROPERTIES = {"GM": 1, "FF": 4, "AP": f"KaTrain:{HOMEPAGE}", "CA": "UTF-8"}
|
||||
DEFAULT_PROPERTIES = {"GM": 1, "FF": 4, "AP": f"{PROGRAM_NAME}:{VERSION}", "CA": "UTF-8"}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -44,6 +45,7 @@ class Game:
|
||||
move_tree: GameNode = None,
|
||||
analyze_fast=False,
|
||||
game_properties: Optional[Dict] = None,
|
||||
loaded_from_file = None
|
||||
):
|
||||
self.katrain = katrain
|
||||
self._lock = threading.Lock()
|
||||
@@ -51,6 +53,8 @@ class Game:
|
||||
engine = {"B": engine, "W": engine}
|
||||
self.engines = engine
|
||||
self.game_id = datetime.strftime(datetime.now(), "%Y-%m-%d %H %M %S")
|
||||
self.loaded_from_file = loaded_from_file
|
||||
|
||||
|
||||
if move_tree:
|
||||
self.root = move_tree
|
||||
@@ -82,13 +86,16 @@ class Game:
|
||||
self.set_current_node(self.root)
|
||||
self.main_time_used = 0
|
||||
threading.Thread(
|
||||
target=lambda: self.analyze_all_nodes(-1_000_000, analyze_fast=analyze_fast), daemon=True
|
||||
target=lambda: self.analyze_all_nodes(-1_000_000, analyze_fast=analyze_fast, even_if_present=False),
|
||||
daemon=True,
|
||||
).start() # return faster, but bypass Kivy Clock
|
||||
|
||||
def analyze_all_nodes(self, priority=0, analyze_fast=False):
|
||||
def analyze_all_nodes(self, priority=0, analyze_fast=False, even_if_present=True):
|
||||
for node in self.root.nodes_in_tree:
|
||||
node.clear_analysis()
|
||||
node.analyze(self.engines[node.next_player], priority=priority, analyze_fast=analyze_fast)
|
||||
if even_if_present or not node.analysis_loaded:
|
||||
print(even_if_present, node.analysis_loaded, node.move)
|
||||
node.clear_analysis()
|
||||
node.analyze(self.engines[node.next_player], priority=priority, analyze_fast=analyze_fast)
|
||||
|
||||
# -- move tree functions --
|
||||
def _calculate_groups(self):
|
||||
@@ -292,9 +299,11 @@ class Game:
|
||||
self, path: str, trainer_config: Optional[Dict] = None,
|
||||
):
|
||||
if trainer_config is None:
|
||||
trainer_config = self.katrain.config("trainer")
|
||||
save_feedback = trainer_config["save_feedback"]
|
||||
trainer_config = self.katrain.config("trainer", {})
|
||||
save_feedback = trainer_config.get("save_feedback", False)
|
||||
eval_thresholds = trainer_config["eval_thresholds"]
|
||||
print(trainer_config)
|
||||
save_analysis = trainer_config.get("save_analysis", False)
|
||||
|
||||
def player_name(player_info):
|
||||
if player_info.name and player_info.player_type == PLAYER_HUMAN:
|
||||
@@ -314,15 +323,23 @@ class Game:
|
||||
x_properties["RE"] = self.end_result
|
||||
self.root.properties = {**root_properties, **{k: [v] for k, v in x_properties.items()}}
|
||||
player_names = {bw: re.sub(r"['<>:\"/\\|?*]", "", self.root.get_property("P" + bw, bw)) for bw in "BW"}
|
||||
game_name = f"katrain_{player_names['B']} vs {player_names['W']} {self.game_id}"
|
||||
file_name = os.path.abspath(os.path.join(path, f"{game_name}.sgf"))
|
||||
base_game_name = f"katrain_{player_names['B']} vs {player_names['W']}"
|
||||
game_name = f"{base_game_name} {self.game_id}"
|
||||
|
||||
if self.loaded_from_file and base_game_name in self.loaded_from_file and PROGRAM_NAME in self.root.get_property("AP","") and self.loaded_from_file.endswith('sgf'):
|
||||
file_name = self.loaded_from_file
|
||||
else:
|
||||
file_name = os.path.abspath(os.path.join(path, f"{game_name}.sgf"))
|
||||
os.makedirs(os.path.dirname(file_name), exist_ok=True)
|
||||
|
||||
show_dots_for = {
|
||||
bw: trainer_config.get("eval_show_ai", True) or self.katrain.players_info[bw].human for bw in "BW"
|
||||
}
|
||||
sgf = self.root.sgf(
|
||||
save_comments_player=show_dots_for, save_comments_class=save_feedback, eval_thresholds=eval_thresholds
|
||||
save_comments_player=show_dots_for,
|
||||
save_comments_class=save_feedback,
|
||||
eval_thresholds=eval_thresholds,
|
||||
save_analysis=save_analysis,
|
||||
)
|
||||
with open(file_name, "w", encoding="utf-8") as f:
|
||||
f.write(sgf)
|
||||
|
||||
+51
-10
@@ -2,10 +2,13 @@ import copy
|
||||
import random
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from katrain.core.constants import HOMEPAGE, VERSION, SGF_COMMENTS_MARKER, SGF_SEPARATOR_MARKER, SGF_NOTE_MARKER, PROGRAM_NAME
|
||||
from katrain.core.lang import i18n
|
||||
from katrain.core.sgf_parser import Move, SGFNode
|
||||
from katrain.core.utils import evaluation_class, var_to_grid
|
||||
from katrain.gui.style import INFO_PV_COLOR
|
||||
import base64
|
||||
import gzip
|
||||
|
||||
|
||||
class GameNode(SGFNode):
|
||||
@@ -22,19 +25,56 @@ class GameNode(SGFNode):
|
||||
self.time_used = 0
|
||||
self.undo_threshold = random.random() # for fractional undos
|
||||
self.end_state = None
|
||||
self.analysis_loaded = False
|
||||
self.clear_analysis()
|
||||
|
||||
def add_list_property(self, property: str, values: List):
|
||||
if property == "KT":
|
||||
try:
|
||||
import bson
|
||||
|
||||
self.analysis = bson.loads(gzip.decompress(base64.standard_b64decode(values[0])))
|
||||
self.analysis_loaded = True
|
||||
except Exception as e:
|
||||
print(e)
|
||||
elif property == "C":
|
||||
comments = [ # strip out all previously auto generated comments
|
||||
c for v in values for c in v.split(SGF_SEPARATOR_MARKER)
|
||||
]
|
||||
self.note = "".join(c for c in comments if SGF_NOTE_MARKER in c)
|
||||
super().add_list_property(
|
||||
property,
|
||||
[c for c in comments if c.strip() and SGF_COMMENTS_MARKER not in c and SGF_NOTE_MARKER not in c],
|
||||
)
|
||||
else:
|
||||
return super().add_list_property(property, values)
|
||||
|
||||
def clear_analysis(self):
|
||||
self.analysis_visits_requested = 0
|
||||
self.analysis = {"moves": {}, "root": None, "completed": False}
|
||||
|
||||
def sgf_properties(self, save_comments_player=None, save_comments_class=None, eval_thresholds=None):
|
||||
def sgf_properties(
|
||||
self, save_comments_player=None, save_comments_class=None, eval_thresholds=None, save_analysis=False
|
||||
):
|
||||
properties = copy.copy(super().sgf_properties())
|
||||
note = self.note.strip()
|
||||
print(save_analysis, self.analysis_complete)
|
||||
if save_analysis and self.analysis_complete:
|
||||
try:
|
||||
import bson
|
||||
|
||||
analysis = copy.deepcopy(self.analysis)
|
||||
for movedict in analysis["moves"].values():
|
||||
if "ownership" in movedict:
|
||||
del movedict["ownership"]
|
||||
properties["KT"] = [base64.standard_b64encode(gzip.compress(bson.dumps(analysis))).decode("utf-8")]
|
||||
except Exception as e:
|
||||
print(e)
|
||||
if self.points_lost and save_comments_class is not None and eval_thresholds is not None:
|
||||
show_class = save_comments_class[evaluation_class(self.points_lost, eval_thresholds)]
|
||||
else:
|
||||
show_class = False
|
||||
comments = properties.get("C", [])
|
||||
if (
|
||||
self.parent
|
||||
and self.parent.analysis_exists
|
||||
@@ -50,18 +90,19 @@ class GameNode(SGFNode):
|
||||
properties["SQ"] = best_sq
|
||||
if top_x and "MA" not in properties:
|
||||
properties["MA"] = [top_x]
|
||||
comment = self.comment(sgf=True, interactive=False)
|
||||
if comment:
|
||||
properties["C"] = ["\n".join(properties.get("C", "")) + comment]
|
||||
comments.append(self.comment(sgf=True, interactive=False))
|
||||
if self.is_root:
|
||||
properties["C"] = [
|
||||
i18n._("SGF start message")
|
||||
+ "\n"
|
||||
+ "\n".join(properties.get("C", ""))
|
||||
+ "\nSGF with review generated by KaTrain."
|
||||
comments = [
|
||||
i18n._("SGF start message") + SGF_COMMENTS_MARKER,
|
||||
*comments,
|
||||
f"SGF with generated by {PROGRAM_NAME} {VERSION}{SGF_COMMENTS_MARKER}.",
|
||||
]
|
||||
if note:
|
||||
properties["C"] = ["\n".join(properties.get("C", "")) + f"\nNote: {self.note}"]
|
||||
comments.append(f"\nNote: {self.note}{SGF_NOTE_MARKER}")
|
||||
if comments:
|
||||
properties["C"] = [SGF_SEPARATOR_MARKER.join(comments)]
|
||||
elif "C" in properties:
|
||||
del properties["C"]
|
||||
return properties
|
||||
|
||||
@staticmethod
|
||||
|
||||
Reference in new issue
Block a user