115 lines
6.6 KiB
Python
115 lines
6.6 KiB
Python
"""Russian action descriptions and readable scenes for the planning model."""
|
||
import re
|
||
|
||
LOCATIONS = {'tray': 'лоток', 'buffer': 'буферная зона', 'conveyor': 'конвейер'}
|
||
SIZES = {'small': 'маленький', 'medium': 'средний', 'large': 'большой'}
|
||
|
||
|
||
def component_for(action, scene):
|
||
return next(c for c in scene['components'] if c['id'] == action['object'])
|
||
|
||
|
||
def object_name(component, scene, prompts):
|
||
return component.get('name') or f"{prompts.object_names.get(scene['domain'], 'деталь')} №{scene['components'].index(component) + 1}"
|
||
|
||
|
||
def target_name(component, scene):
|
||
return component.get('target_name') or f"сборочная позиция №{scene['components'].index(component) + 1}"
|
||
|
||
|
||
def action_text(action, scene, prompts):
|
||
component = component_for(action, scene)
|
||
return prompts.action_templates[action['skill']].format(
|
||
object=object_name(component, scene, prompts), target=target_name(component, scene))
|
||
|
||
|
||
def normalized_action_text(text):
|
||
words = re.findall(r'[\w]+', text.casefold())
|
||
while words and words[0] in {'сначала', 'затем', 'далее'}:
|
||
words.pop(0)
|
||
synonyms = {'перенести': 'переместить', 'взять': 'захватить',
|
||
'проконтролировать': 'проверить', 'возвратиться': 'вернуться',
|
||
'местоположение': 'положение'}
|
||
return [synonyms.get(word, word) for word in words]
|
||
|
||
|
||
def validate_texts(episode, texts, prompts):
|
||
if len(texts) != len(episode['plan']['actions']):
|
||
raise ValueError('Число русских описаний не совпадает с числом действий')
|
||
for action, text in zip(episode['plan']['actions'], texts):
|
||
if not text.strip() or not re.search('[А-Яа-яЁё]', text):
|
||
raise ValueError('Каждое действие должно быть описано на русском языке')
|
||
if re.search(r'\b(?:part_|slot_)\d+|\b(?:robot|camera)\.', text):
|
||
raise ValueError('В тексте плана используй названия объектов, а не код')
|
||
component = component_for(action, episode['scene'])
|
||
if action['skill'] != 'return_home' and component.get('name') and component['name'].casefold() not in text.casefold():
|
||
raise ValueError(f"В описании действия укажи название объекта: {component['name']}")
|
||
expected = action_text(action, episode['scene'], prompts)
|
||
if normalized_action_text(text) != normalized_action_text(expected):
|
||
raise ValueError(f'Описание не соответствует проверяемому действию. Используй: {expected}')
|
||
|
||
|
||
def step_texts(episode, prompts):
|
||
if 'step_texts' in episode:
|
||
validate_texts(episode, episode['step_texts'], prompts)
|
||
return episode['step_texts']
|
||
return [action_text(action, episode['scene'], prompts) for action in episode['plan']['actions']]
|
||
|
||
|
||
def terminal_text(status):
|
||
return ('Задание выполнено.' if status == 'ready' else
|
||
'Задание невозможно завершить в текущих условиях. Сообщить оператору о недоступных ресурсах или зависимостях и запросить уточнение.')
|
||
|
||
|
||
def answer_text(episode, mode, index, prompts):
|
||
steps = step_texts(episode, prompts)
|
||
status = episode['plan']['status']
|
||
if mode == 'step':
|
||
return steps[index] if steps else terminal_text(status)
|
||
if not steps:
|
||
return terminal_text(status)
|
||
if mode == 'full':
|
||
lines = [f'{i}. {text}' for i, text in enumerate(steps, 1)]
|
||
if status == 'blocked':
|
||
lines.append(terminal_text(status))
|
||
return '\n'.join(lines)
|
||
lines = []
|
||
for i, text in enumerate(steps, 1):
|
||
success = f'перейти к пункту {i + 1}' if i < len(steps) else terminal_text(status)
|
||
lines.append(f'{i}. {text}\n Если действие выполнено успешно — {success}.\n'
|
||
' Если произошла ошибка — остановиться, уточнить состояние сцены и запросить новый план.')
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def readable_scene(scene, state, prompts):
|
||
state = set(state)
|
||
names = {c['id']: object_name(c, scene, prompts) for c in scene['components']}
|
||
objects = []
|
||
for c in scene['components']:
|
||
prefix = f"at:{c['id']}:"
|
||
location = next((s[len(prefix):] for s in state if s.startswith(prefix)), None)
|
||
location = target_name(c, scene) if location == c['target'] else LOCATIONS.get(location, location)
|
||
if f"holding:{c['id']}" in state:
|
||
location = 'в захвате робота'
|
||
stage = next((int(s.rsplit(':', 1)[1]) for s in state if s.startswith(f"{c['id']}:stage:")), 0)
|
||
operations = [op['action'] for op in scene['operations'] if op['action']['object'] == c['id']]
|
||
objects.append({'название': names[c['id']], 'размер': SIZES.get(c['type'], c['type']),
|
||
'положение': location or 'не определено', 'назначение': target_name(c, scene),
|
||
'доступен': f"available:{c['id']}" in state,
|
||
'после_завершения': [names[d] for d in c['dependencies']],
|
||
'выполненные_действия': [action_text(a, scene, prompts) for a in operations[:stage]]})
|
||
return {'объекты': objects, 'захват_свободен': 'gripper_empty' in state,
|
||
'оператор_готов': 'operator_ready' in state,
|
||
'цель': 'Выполнить задание для всех перечисленных объектов и освободить захват.',
|
||
'ограничение': 'Соблюдать зависимости объектов и не начинать следующий этап до завершения предыдущего.'}
|
||
|
||
|
||
def scene_summary(scene):
|
||
lines = []
|
||
for obj in scene['объекты']:
|
||
lines.append(f"{obj['название']}: {obj['положение']}; назначение — {obj['назначение']}; "
|
||
f"{'доступен' if obj['доступен'] else 'недоступен'}.")
|
||
if obj['после_завершения']:
|
||
lines.append(' Начать после: ' + ', '.join(obj['после_завершения']) + '.')
|
||
return '\n'.join(lines)
|