Source code for concepts.dm.pdsketch.strips.atomic_strips_regression_search

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
# File   : atomic_strips_regression_search.py
# Author : Jiayuan Mao
# Email  : maojiayuan@gmail.com
# Date   : 07/13/2023
#
# This file is part of Project Concepts.
# Distributed under terms of the MIT license.

import itertools
from typing import List, Set

import jacinle

from concepts.dm.pdsketch.strips.strips_expression import SProposition
from concepts.dm.pdsketch.strips.atomic_strips_domain import (
    AtomicStripsDomain, AtomicStripsProblem,
    AtomicStripsOperatorApplier, AtomicStripsGroundedAchieveExpression
)


[docs] def gen_all_grounded_actions_and_rules(domain: AtomicStripsDomain, problem: AtomicStripsProblem, mode: str) -> List[AtomicStripsOperatorApplier]: assert mode in ('action', 'rule') all_actions = list() if mode == 'action': operator_list = domain.operators elif mode == 'rule': operator_list = domain.regression_rules else: raise ValueError('Unknown mode: {}.'.format(mode)) for operator in operator_list.values(): candidate_arguments = [ problem.objects_type2names[operator.arguments[i].dtype.typename] for i in range(len(operator.arguments)) ] for arg_list in itertools.product(*candidate_arguments): action = operator(*arg_list) static_check = True for i, prop in enumerate(action.preconditions): if domain.predicates[operator.preconditions[i].name].is_static: if prop not in problem.initial_state: static_check = False break if not static_check: continue all_actions.append(action) return all_actions
[docs] def astrips_regression_search_1(domain: AtomicStripsDomain, problem: AtomicStripsProblem, verbose: bool = False) -> List[AtomicStripsOperatorApplier]: """Search for a plan using regression search.""" assert problem.conjunctive_goal is not None, "Only conjunctive goals are supported." assert len(problem.conjunctive_goal) == 1, "Only single-goal problems are supported." for operator in domain.operators.values(): for precondition in operator.preconditions: if precondition.negated: raise NotImplementedError('astrips_regression_search does not support negated preconditions.') for regression_rule in domain.regression_rules.values(): for precondition in regression_rule.preconditions: if precondition.negated: raise NotImplementedError('astrips_regression_search does not support negated preconditions.') all_rules = gen_all_grounded_actions_and_rules(domain, problem, 'rule') def find_applicable_rules(state: Set[SProposition], goal: SProposition, maintains: Set[SProposition]): applicable_rules = list() for rule in all_rules: if rule.goal == goal and state.issuperset(rule.preconditions): applicable_rules.append(rule) assert len(applicable_rules) == 1, "Only one applicable rule is allowed." return applicable_rules[0] @jacinle.log_function(verbose=False) def dfs(state: Set[SProposition], goal: SProposition, maintains: Set[SProposition]): rule = find_applicable_rules(state, goal, maintains) if verbose: jacinle.log_function.print(f'Current state: {state}, goal: {goal}, maintains: {maintains} => rule: {rule}') actions = list() for item in rule.body: if isinstance(item, AtomicStripsGroundedAchieveExpression): if item.goal not in state: if verbose: jacinle.log_function.print(f'{str(item)}') state, sub_actions = dfs(state, item.goal, maintains.union(item.maintains)) actions.extend(sub_actions) else: if verbose: jacinle.log_function.print(f'{str(item)} (skipped)') elif isinstance(item, AtomicStripsOperatorApplier): if verbose: jacinle.log_function.print(f'do({str(item)})') actions.append(item) assert state.issuperset(item.preconditions), "The preconditions of an action should be satisfied." state = (state - frozenset(item.del_effects)).union(frozenset(item.add_effects)) else: raise ValueError('Unknown item: {}.'.format(item)) return state, actions end_state, actions = dfs(problem.initial_state, problem.conjunctive_goal[0], set()) return actions