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

Merge branch 'feature/gemma-image-recognition'

Browse Source
This commit is contained in:
every_holiday
2026-06-09 02:37:36 +09:00
parent cc424dca56 0521af2933
commit 072e819b0b
5 changed files with 279 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/app.py
View File

@@ -17,6 +17,7 @@ from discord_control.actions import setDiscordMute
from ocr.ocr_actions import runOcrFromScreen
from translate.translate_actions import saveTranslationText
from translate.translate_actions import translateTextToJapanese
from vision.vision_actions import runVisionFromScreen
from vrc_log.log_actions import collectVrchatLog
from vrc_log.log_actions import startSelfMonitor
from vrc_log.log_actions import startVrcLogMonitor
@@ -27,6 +28,7 @@ PARAM_DISCORD_MUTE = "DiscordSend"
PARAM_LOG_GREP = "vrc_log"
PARAM_OCR = "ocrEnabled"
PARAM_TRANSLATE_OCR = "translation"
PARAM_VISION = "visionEnabled"
def create_gateway():
@@ -96,10 +98,20 @@ def create_gateway():
saved = saveTranslationText(stem, translated)
gateway.log("INFO", f"translation saved={saved}")
def on_vision(address, *args):
if not args:
gateway.log("ERROR", "visionEnabled args empty")
return
if gateway.is_rising_edge(address, args[0]):
gateway.log("INFO", f"received {address} args={args}")
runVisionFromScreen()
gateway.map(PARAM_DISCORD_MUTE, on_discord_mute)
gateway.map(PARAM_LOG_GREP, on_log_grep)
gateway.map(PARAM_OCR, on_ocr)
gateway.map(PARAM_TRANSLATE_OCR, on_translate_ocr)
gateway.map(PARAM_VISION, on_vision)
gateway.set_default_debug(enabled=False)
return gateway

38
src/common/config_loader.py
View File

@@ -56,16 +56,41 @@ 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)),
}
def loadVisionConfig():
config = loadTomlFile(CONFIG_FILE)
vision = config.get("vision", {})
return {
"model_name": str(vision.get("model_name", "google/gemma-4-31B-it")).strip(),
"prompt": str(
vision.get(
"prompt",
"Describe the image briefly and extract any readable text.",
)
).strip(),
"max_new_tokens": int(vision.get("max_new_tokens", 256)),
"temperature": float(vision.get("temperature", 0.2)),
}
@@ -108,6 +133,14 @@ def loadVrcLogConfig():
notice_section = event_config.get("notice", {})
missing_count = notice_section.get("missing_count", 0)
log_patterns = (
vrc_log.get("patterns")
or vrc_log.get("log_patterns")
or event_config.get("patterns")
or event_config.get("log_patterns")
or []
)
if not missing_count:
missing_count = event_config.get("missing_count", 0)
@@ -127,4 +160,5 @@ def loadVrcLogConfig():
"staff_names": [str(name).strip() for name in staff_names if str(name).strip()],
"missing_count": int(missing_count),
"self_name": str(self_name).strip(),
"log_patterns": [str(pattern).strip() for pattern in log_patterns if str(pattern).strip()],
}

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)

116
src/vision/vision_actions.py Normal file
View File

@@ -0,0 +1,116 @@
from datetime import datetime
from common.config_loader import loadVisionConfig
from common.runtime_log import appendRuntimeLog
from screen.screen_actions import captureOcrRegion
_vision_pipeline = None
def log(level, message):
now = datetime.now().strftime("%H:%M:%S")
print(f"[{level}] {now} {message}", flush=True)
def getVisionPipeline():
global _vision_pipeline
if _vision_pipeline is not None:
return _vision_pipeline
config = loadVisionConfig()
model_name = config["model_name"]
if not model_name:
raise RuntimeError("vision.model_name is not set")
try:
from transformers import AutoProcessor
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 = Gemma4ForConditionalGeneration.from_pretrained(
model_name,
torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
device_map="auto",
).eval()
_vision_pipeline = (processor, model)
return _vision_pipeline
def saveVisionText(image_path, text):
return appendRuntimeLog(
"VISION TEXT",
f"image={image_path}\n{text}",
)
def runVisionFromImage(image_path):
config = loadVisionConfig()
processor, model = getVisionPipeline()
prompt = config["prompt"]
messages = [
{
"role": "system",
"content": [
{"type": "text", "text": "You are a helpful vision-language assistant."},
],
},
{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": prompt},
],
}
]
try:
import torch
from PIL import Image
except Exception as e:
log("ERROR", f"Vision runtime imports failed detail={e}")
return None
try:
image = Image.open(image_path).convert("RGB")
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"],
do_sample=config["temperature"] > 0,
)
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
text_path = saveVisionText(image_path, text)
log("INFO", f"Vision text appended={text_path}")
print(text, flush=True)
return {
"image_path": image_path,
"text_path": text_path,
"text": text,
}
def runVisionFromScreen():
log("ACTION", "runVisionFromScreen called")
image_path = captureOcrRegion()
return runVisionFromImage(image_path)

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: