every_holiday/VRCWT-OSC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Add Gemini OCR and fix VRC log world handling

Browse Source
This commit is contained in:
every_holiday
2026-06-09 02:37:06 +09:00
parent fad1a20089
commit 0521af2933
4 changed files with 148 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/common/config_loader.py
View File

@@ -56,16 +56,24 @@ def getSecretValue(section_name, key, default=""):
def loadOcrConfig():
config = loadTomlFile(CONFIG_FILE)
ocr = config.get("ocr", {})
secrets = loadSecretConfig()
ai_secrets = secrets.get("ai", {})
crop = ocr.get("crop", {})
return {
"provider": str(ai_secrets.get("provider", ocr.get("provider", "paddleocr"))).strip().lower(),
"model_name": str(ai_secrets.get("model_name", ocr.get("model_name", ""))).strip(),
"api_key": str(ai_secrets.get("GEMINI_API_KEY", "")).strip(),
"lang": str(ocr.get("lang", "japan")).strip(),
"crop_left": float(crop.get("left", 0.05)),
"crop_top": float(crop.get("top", 0.35)),
"crop_right": float(crop.get("right", 0.95)),
"crop_bottom": float(crop.get("bottom", 0.95)),
"preprocess": bool(ocr.get("preprocess", True)),
"scale": float(ocr.get("scale", 1.5)),
"use_angle_cls": bool(ocr.get("use_angle_cls", False)),
"scale": float(ocr.get("scale", 2.0)),
"use_angle_cls": bool(ocr.get("use_angle_cls", True)),
"delay_seconds": float(ocr.get("delay_seconds", 1.0)),
"show_log": bool(ocr.get("show_log", False)),
}
@@ -74,7 +82,7 @@ def loadVisionConfig():
vision = config.get("vision", {})
return {
"model_name": str(vision.get("model_name", "google/gemma-3-4b-it")).strip(),
"model_name": str(vision.get("model_name", "google/gemma-4-31B-it")).strip(),
"prompt": str(
vision.get(
"prompt",

89
src/ocr/ocr_actions.py
View File

@@ -1,4 +1,5 @@
import os
import time
from datetime import datetime
os.environ["FLAGS_enable_pir_api"] = "0"
@@ -33,9 +34,9 @@ def getOcrEngine():
log("INFO", "PaddleOCR initializing")
config = loadOcrConfig()
options = {
"lang": "japan",
"lang": config["lang"],
"use_angle_cls": config["use_angle_cls"],
"show_log": False,
"show_log": config["show_log"],
}
try:
@@ -47,6 +48,27 @@ def getOcrEngine():
return _ocr_engine
_gemini_client = None
def getGeminiClient(api_key):
global _gemini_client
if _gemini_client is not None:
return _gemini_client
if not api_key:
raise RuntimeError("Gemini API key is not set in secrets.toml [ai]")
try:
from google import genai
except Exception as e:
raise RuntimeError(f"Gemini SDK is missing detail={e}")
_gemini_client = genai.Client(api_key=api_key)
return _gemini_client
def saveOcrText(image_path, text):
return appendRuntimeLog(
"OCR TEXT",
@@ -149,18 +171,69 @@ def runPaddleOcr(engine, image_path):
return engine.ocr(str(image_path))
def runOcrFromImage(image_path, show_result=True):
log("ACTION", f"PaddleOCR start image={image_path}")
def runGeminiOcr(image_path):
config = loadOcrConfig()
model_name = config["model_name"] or "gemini-2.5-flash"
prompt = (
"Extract every readable text fragment from this image with maximum accuracy. "
"Do not summarize, explain, translate, or guess. "
"Return only the text exactly as it appears, preserving line breaks and order."
)
try:
from google.genai import types
except Exception as e:
raise RuntimeError(f"Gemini types import failed detail={e}")
client = getGeminiClient(config["api_key"])
mime_type = "image/png"
suffix = str(image_path).lower()
if suffix.endswith(".jpg") or suffix.endswith(".jpeg"):
mime_type = "image/jpeg"
elif suffix.endswith(".webp"):
mime_type = "image/webp"
image_bytes = image_path.read_bytes()
image_part = types.Part.from_bytes(data=image_bytes, mime_type=mime_type)
response = client.models.generate_content(
model=model_name,
contents=[prompt, image_part],
)
text = (response.text or "").strip()
return text
def runOcrFromImage(image_path, show_result=True):
try:
image_path = preprocessOcrImage(image_path)
config = loadOcrConfig()
provider = config["provider"]
log("ACTION", f"OCR start provider={provider} image={image_path}")
if provider == "gemini":
text = runGeminiOcr(image_path)
else:
engine = getOcrEngine()
result = runPaddleOcr(engine, image_path)
text = extractTextFromAny(result)
if not text:
raise RuntimeError("OCR text is empty")
except Exception as e:
log("ERROR", f"PaddleOCR failed detail={e}")
log("ERROR", f"OCR failed detail={e}")
if config.get("provider") == "gemini":
log("INFO", "Falling back to PaddleOCR")
try:
engine = getOcrEngine()
result = runPaddleOcr(engine, image_path)
text = extractTextFromAny(result)
except Exception as fallback_error:
log("ERROR", f"PaddleOCR fallback failed detail={fallback_error}")
return None
else:
return None
text = extractTextFromAny(result)
text_path = saveOcrText(image_path, text)
try:
@@ -183,4 +256,8 @@ def runOcrFromImage(image_path, show_result=True):
def runOcrFromScreen():
log("ACTION", "runOcrFromScreen called")
image_path = captureOcrRegion()
delay_seconds = max(0.0, loadOcrConfig().get("delay_seconds", 1.0))
if delay_seconds:
log("INFO", f"OCR delayed seconds={delay_seconds}")
time.sleep(delay_seconds)
return runOcrFromImage(image_path, show_result=True)

30
src/vision/vision_actions.py
View File

@@ -26,18 +26,18 @@ def getVisionPipeline():
try:
from transformers import AutoProcessor
from transformers import AutoModelForImageTextToText
from transformers import Gemma4ForConditionalGeneration
import torch
except Exception as e:
raise RuntimeError(f"vision dependencies are missing detail={e}")
log("INFO", f"Vision model initializing model={model_name}")
processor = AutoProcessor.from_pretrained(model_name)
model = AutoModelForImageTextToText.from_pretrained(
model = Gemma4ForConditionalGeneration.from_pretrained(
model_name,
torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
device_map="auto",
)
).eval()
_vision_pipeline = (processor, model)
return _vision_pipeline
@@ -55,6 +55,12 @@ def runVisionFromImage(image_path):
prompt = config["prompt"]
messages = [
{
"role": "system",
"content": [
{"type": "text", "text": "You are a helpful vision-language assistant."},
],
},
{
"role": "user",
"content": [
@@ -73,15 +79,23 @@ def runVisionFromImage(image_path):
try:
image = Image.open(image_path).convert("RGB")
inputs = processor(images=image, text=prompt, return_tensors="pt")
inputs = {key: value.to(model.device) for key, value in inputs.items()}
inputs = processor.apply_chat_template(
messages,
add_generation_prompt=True,
tokenize=True,
return_dict=True,
return_tensors="pt",
)
inputs = inputs.to(model.device, dtype=model.dtype)
with torch.no_grad():
output = model.generate(
**inputs,
max_new_tokens=config["max_new_tokens"],
temperature=config["temperature"],
do_sample=config["temperature"] > 0,
)
text = processor.batch_decode(output, skip_special_tokens=True)[0].strip()
input_len = inputs["input_ids"].shape[-1]
generated = output[0][input_len:]
text = processor.decode(generated, skip_special_tokens=True).strip()
except Exception as e:
log("ERROR", f"Vision inference failed detail={e}")
return None

28
src/vrc_log/log_actions.py
View File

@@ -352,6 +352,9 @@ def parseVrchatEvents(text, config):
output_lines = []
self_state = None
if not guest_names:
debug("guest_names is empty; guest comparison is disabled")
for raw_line in text.splitlines():
line_time = parseLineTime(raw_line)
if not line_time or line_time < start_time:
@@ -362,14 +365,22 @@ def parseVrchatEvents(text, config):
room_match = ENTERING_ROOM_PATTERN.search(raw_line)
world_name_match = WORLD_NAME_PATTERN.search(raw_line)
world_match = WORLD_PATTERN.search(raw_line)
should_update_world = bool(room_match or world_name_match)
location = ""
world_id = ""
instance_id = ""
if should_update_world and world_match:
if world_name_match:
world_id = world_name_match.group(1).strip()
location = world_id
elif room_match and world_match:
location = world_match.group(1).strip()
world_id, instance_id = splitWorldLocation(location)
elif world_match:
location = world_match.group(1).strip()
world_id, instance_id = splitWorldLocation(location)
if location != state["location"]:
state["location"] = location
if world_id and world_id != state["world_id"]:
state["location"] = location or world_id
state["world_id"] = world_id
state["instance_id"] = instance_id
state["guest_present_set"].clear()
@@ -377,9 +388,12 @@ def parseVrchatEvents(text, config):
state["member_present_set"].clear()
state["last_notice_key"] = None
output_lines = []
world_label = formatWorldLabel(world_id, world_name_map)
if world_label != "不明":
output_lines.append(f"ワールド入室: {world_label}")
join_match = JOIN_PATTERN.search(raw_line)
if join_match and state["world_id"] and line_matches:
if join_match and line_matches:
name = join_match.group(1).strip()
join_leave_line = None
if isSelf(name, self_name):
@@ -408,7 +422,7 @@ def parseVrchatEvents(text, config):
continue
left_match = LEFT_PATTERN.search(raw_line)
if left_match and state["world_id"] and line_matches:
if left_match and line_matches:
name = left_match.group(1).strip()
join_leave_line = None
if isSelf(name, self_name):
@@ -469,9 +483,9 @@ def collectVrchatLog(pattern=None, notify_changed=True):
text = readTextSafe(latest_log)
lines, self_state = parseVrchatEvents(text, config)
output_text = "\n".join(lines) if lines else ""
output_path = None
if lines:
output_text = "\n".join(lines)
output_path = appendRuntimeLog("VRC LOG", output_text)
join_leave_lines = [line for line in lines if "[join]" in line or "[leave]" in line]
if join_leave_lines: