"""Finite symbolic assembly worlds; model output cannot supply its own effects. These recipes describe logical prerequisites only, not physical feasibility. Every grounded operation in a scene is supplied by this trusted generator. """ import hashlib import json import random from copy import deepcopy from .config import Config from .prompts import render from .structured import structured_answer from .language import answer_text, readable_scene, scene_summary, step_texts # Skills before release must retain the component; 'installed' is a logical # stage, not an assertion that real geometry or process parameters are safe. RECIPES = { 'manipulation': ('Перемести детали в сборочные позиции', ['pick', 'position', 'place', 'release', 'inspect'], None), 'mechanical': ('Собери механический узел', ['pick', 'align', 'insert', 'release', 'inspect'], None), 'fastening': ('Установи и затяни крепёж', ['pick', 'align', 'insert_fastener', 'tighten', 'release', 'inspect'], 'driver'), 'press_fit': ('Собери прессовые соединения', ['pick', 'align', 'press_fit', 'release', 'inspect'], 'press'), 'electrical': ('Выполни электромонтаж', ['pick', 'route_wire', 'connect', 'release', 'test_continuity', 'inspect'], 'tester'), 'electronics': ('Установи электронные компоненты', ['pick', 'align', 'seat_component', 'release', 'test_connection', 'inspect'], 'tester'), 'adhesive': ('Собери клеевые соединения', ['clean_surface', 'apply_adhesive', 'pick', 'align', 'bond', 'release', 'wait_cure', 'inspect'], 'dispenser'), 'welding': ('Соедини детали сваркой', ['pick', 'align', 'clamp', 'release', 'weld', 'inspect'], 'welder'), 'collaborative': ('Передай детали оператору для сборки', ['wait_operator', 'pick', 'handover', 'release', 'confirm_assembly', 'inspect'], None), 'operator_assistance': ('Удерживай коробку возле оператора', ['detect', 'approach', 'pick', 'move_to_operator', 'hold', 'wait_assembly', 'move_to_storage', 'release', 'return_home'], None), 'inspection': ('Проверь детали в контрольной позиции', ['pick', 'position', 'place', 'release', 'measure', 'inspect'], 'gauge'), } # Explicit perception and approach precede grasping in the human-level plan. for _domain, (_title, _recipe, _tool) in list(RECIPES.items()): if _domain != 'operator_assistance': _pick = _recipe.index('pick') RECIPES[_domain] = (_title, _recipe[:_pick] + ['detect', 'approach'] + _recipe[_pick:], _tool) def canonical(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')) def digest(value): return hashlib.sha256(canonical(value).encode()).hexdigest() def action_key(action): if not isinstance(action, dict) or set(action) != {'skill', 'object', 'target'}: raise ValueError('Действие должно иметь только skill, object, target') return canonical(action) def enabled(state, operation): return set(operation['requires']) <= state and not (set(operation['forbids']) & state) def apply(state, operation): if not enabled(state, operation): raise ValueError(f"Нарушены предусловия: {operation['action']}") return (state - set(operation['delete'])) | set(operation['add']) def solve(scene): state = set(scene['initial_state']) actions, trace = [], [sorted(state)] remaining = list(scene['operations']) while remaining: operation = next((op for op in remaining if enabled(state, op)), None) if operation is None: break state = apply(state, operation) actions.append(operation['action']) trace.append(sorted(state)) remaining.remove(operation) status = 'ready' if set(scene['goal']) <= state else 'blocked' return {'status': status, 'actions': actions}, trace def validate_plan(episode, plan): if not isinstance(plan, dict) or set(plan) != {'status', 'actions'}: raise ValueError('Ожидаются status и actions') if plan['status'] not in ('ready', 'blocked') or not isinstance(plan['actions'], list): raise ValueError('Неверный статус/список действий') operations = {action_key(op['action']): op for op in episode['scene']['operations']} state = set(episode['scene']['initial_state']) trace = [sorted(state)] for action in plan['actions']: operation = operations.get(action_key(action)) if operation is None: raise ValueError(f'Неизвестное действие или объект: {action}') state = apply(state, operation) trace.append(sorted(state)) achieved = set(episode['scene']['goal']) <= state if plan['status'] == 'ready' and not achieved: raise ValueError('План не достигает цели') if plan['status'] == 'blocked': # This domain has an acyclic recipe dependency graph. The deterministic # solver is complete for its grounded operations (not arbitrary PDDL). if achieved or solve(episode['scene'])[0]['status'] != 'blocked': raise ValueError('Необоснованная блокировка выполнимой задачи') return trace def branch_plan(plan): nodes = [] for index, action in enumerate(plan['actions']): nodes.append({'id': f'n{index}', 'action': action, 'observation': 'action_succeeded', 'on_success': f'n{index + 1}' if index + 1 < len(plan['actions']) else 'done', 'on_failure': 'replan'}) nodes.extend([ {'id': 'done', 'terminal': plan['status']}, {'id': 'replan', 'terminal': 'replan_required'}, ]) return {'entry': 'n0' if plan['actions'] else 'done', 'nodes': nodes} def validate_branch(episode, branch): if set(branch) != {'entry', 'nodes'} or not isinstance(branch['nodes'], list): raise ValueError('Неверная схема графа') nodes = {node['id']: node for node in branch['nodes']} if len(nodes) != len(branch['nodes']): raise ValueError('Повторяющиеся узлы') operations = {action_key(op['action']): op for op in episode['scene']['operations']} pending = [(branch['entry'], set(episode['scene']['initial_state']), frozenset(), False)] paths = 0 while pending: node_id, state, visited, failed = pending.pop() if node_id not in nodes or node_id in visited: raise ValueError('Отсутствующий узел или цикл') node = nodes[node_id] if failed and node.get('terminal') != 'replan_required': raise ValueError('После сбоя необходима новая сцена и перепланирование') if 'terminal' in node: if set(node) != {'id', 'terminal'}: raise ValueError('Неверный терминальный узел') status = node['terminal'] if status == 'ready': if not set(episode['scene']['goal']) <= state: raise ValueError('Ветка не достигает цели') elif status == 'blocked': if solve(episode['scene'])[0]['status'] != 'blocked': raise ValueError('Необоснованная блокировка') elif status == 'replan_required': if not failed: raise ValueError('Перепланирование без наблюдаемого сбоя') else: raise ValueError('Неизвестный терминальный статус') paths += 1 continue if set(node) != {'id', 'action', 'observation', 'on_success', 'on_failure'} or node['observation'] != 'action_succeeded': raise ValueError('Неполный узел или неизвестное наблюдение') operation = operations.get(action_key(node['action'])) if operation is None: raise ValueError('Неизвестная операция') success = apply(state, operation) visited = visited | {node_id} # The defined failure outcome has no success effects. Stop and obtain a # fresh scene; never blindly retry a physical action after a failure. pending.append((node['on_failure'], state, visited, True)) pending.append((node['on_success'], success, visited, failed)) if len(pending) + paths > 10000: raise ValueError('Слишком много путей в графе') return paths def make_episode(index: int, config: Config): rng = random.Random(f'{config.seed}:{index}') domain = config.domains[index % len(config.domains)] title, recipe, tool = RECIPES[domain] missing = set(recipe) - set(config.prompts.action_templates) if missing: raise ValueError(f'В action_templates отсутствуют навыки: {sorted(missing)}') count = rng.randint(config.min_components, config.max_components) state = {'gripper_empty', 'fixture_ready', 'operator_ready'} if domain == 'operator_assistance': state.add('robot_home') if tool: state.add(f'tool_available:{tool}') components, operations = [], [] blocked = rng.random() < config.blocked_fraction unavailable = rng.randrange(count) if blocked else -1 for i in range(count): obj, target = f'part_{i}', f'slot_{i}' dependencies = [f'part_{j}' for j in range(i) if rng.random() < 0.35] # Positions, part classes and progress have operational meaning, and are # included in the signature. No free-form names enter deduplication. location = rng.choice(['tray', 'buffer', 'conveyor']) components.append({'id': obj, 'target': target, 'type': rng.choice(['small', 'medium', 'large']), 'source_location': location, 'dependencies': dependencies, 'name': f"{config.prompts.object_names.get(domain, 'деталь')} №{i + 1}", 'target_name': f"{'зона хранения' if domain == 'operator_assistance' else 'сборочная позиция'} №{i + 1}"}) state.add(f'at:{obj}:{location}') if i != unavailable: state.add(f'available:{obj}') prefix = f'{obj}:stage:' # Some parts are already installed; final inspection is still required. start = len(recipe) - 1 if domain != 'operator_assistance' and i != unavailable and not dependencies and rng.random() < 0.2 else 0 state.add(f'{prefix}{start}') for stage, skill in enumerate(recipe): requires = [f'{prefix}{stage}', 'fixture_ready', f'available:{obj}'] adds, deletes = [f'{prefix}{stage + 1}'], [f'{prefix}{stage}'] if skill == 'pick': requires += ['gripper_empty', f'at:{obj}:{location}'] requires += [f'complete:{dependency}' for dependency in dependencies] adds += [f'holding:{obj}'] deletes += ['gripper_empty', f'at:{obj}:{location}'] elif skill == 'release': requires += [f'holding:{obj}'] adds += ['gripper_empty', f'at:{obj}:{target}'] deletes += [f'holding:{obj}'] elif 'pick' in recipe[:stage] and 'release' not in recipe[:stage]: requires += [f'holding:{obj}'] if skill in ('tighten', 'press_fit', 'apply_adhesive', 'weld', 'measure', 'test_connection', 'test_continuity') and tool: requires += [f'tool_available:{tool}'] if skill == 'approach': requires += ['gripper_empty'] if domain == 'operator_assistance': deletes += ['robot_home'] if skill == 'wait_assembly': adds += [f'operator_finished:{obj}'] if skill == 'move_to_storage': requires += [f'operator_finished:{obj}'] if skill == 'return_home': requires += ['gripper_empty'] adds += ['robot_home'] if skill in ('wait_operator', 'handover', 'confirm_assembly', 'move_to_operator', 'hold', 'wait_assembly'): requires += ['operator_ready'] if stage == len(recipe) - 1: adds += [f'complete:{obj}'] operations.append({'action': {'skill': skill, 'object': obj, 'target': target}, 'requires': requires, 'forbids': [], 'add': adds, 'delete': deletes, 'failure_effects': []}) if start: state.discard(f'at:{obj}:{location}') state.add(f'at:{obj}:{target}') scene = {'domain': domain, 'components': components, 'initial_state': sorted(state), 'goal': ['gripper_empty'] + [f'complete:part_{i}' for i in range(count)], 'operations': operations} if domain == 'operator_assistance': scene['goal'].append('robot_home') signature = digest(scene) plan, trace = solve(scene) episode = {'scenario_id': signature, 'signature': signature, 'candidate_index': index, 'instruction': config.prompts.task_instructions[domain], 'scene': scene, 'plan': plan, 'trace': trace, 'branch': branch_plan(plan), 'validation_level': 'logical_only', 'source': {'provider': 'symbolic', 'model': None}} validate_plan(episode, plan) validate_branch(episode, episode['branch']) return episode def split_for(signature, config): value = int(signature[:16], 16) / 2**64 return 'train' if value < config.split[0] else 'validation' if value < sum(config.split[:2]) else 'test' def training_rows(episode, config, *, output_format='text'): """All views share the same split; only pre-action state enters step input.""" if output_format not in ('text', 'structured'): raise ValueError('Неизвестный формат ответа') common = {'instruction': episode['instruction'], 'scene': readable_scene(episode['scene'], episode['scene']['initial_state'], config.prompts)} for mode in config.modes: if mode == 'step': actions = episode['plan']['actions'] indices = range(len(actions)) if config.step_sampling == 'all' else [int(episode['signature'][:8], 16) % max(1, len(actions))] if not actions: indices = [0] for step in indices: prompt = deepcopy(common) prompt['scene'] = readable_scene(episode['scene'], episode['trace'][step], config.prompts) prompt['history'] = step_texts(episode, config.prompts)[:step] answer = {'status': 'continue', 'action': actions[step]} if actions else {'status': episode['plan']['status'], 'action': None} yield row(episode, mode, step, prompt, answer, config, output_format) else: prompt = deepcopy(common) if mode == 'branch': prompt['observations'] = ['действие выполнено успешно', 'при выполнении произошла ошибка'] yield row(episode, mode, 0, prompt, episode['plan'] if mode == 'full' else episode['branch'], config, output_format) def row(episode, mode, index, prompt, answer, config, output_format): prompt['mode'] = mode structured = output_format == 'structured' public_answer = structured_answer(episode, mode, index, config.prompts) if structured else None return {'record_id': f"{episode['scenario_id']}:{mode}:{index}", 'scenario_id': episode['scenario_id'], 'split': split_for(episode['signature'], config), 'mode': mode, 'validation_level': 'logical_only', 'source': episode['source'], 'task': episode['instruction'], 'scene_summary': scene_summary(prompt['scene']), 'status': public_answer['status'] if structured else (('continue' if episode['plan']['actions'] else episode['plan']['status']) if mode == 'step' else episode['plan']['status']), 'text_source': episode.get('text_source', 'template'), 'output_format': output_format, 'messages': [ {'role': 'system', 'content': (config.prompts.structured_system if structured else config.prompts.training_system)[mode]}, {'role': 'user', 'content': render(config.prompts.training_user, canonical(prompt))}, {'role': 'assistant', 'content': canonical(public_answer) if structured else answer_text(episode, mode, index, config.prompts)}, ]}