from __future__ import annotations
import re
import sys
import importlib_resources
from itertools import product
from omegaconf import OmegaConf
from .meta.base import MetaTaskBase
from ..sim import MineDojoSim, InventoryItem
from ..sim.wrappers import FastResetWrapper, ARNNWrapper
from .meta import (
HarvestMeta,
CombatMeta,
TechTreeMeta,
Playthrough,
SurvivalMeta,
CreativeMeta,
)
import logging
_logger = logging.getLogger(__name__)
_logger.setLevel(logging.INFO)
_stream_handler = logging.StreamHandler(stream=sys.stderr)
_stream_handler.setFormatter(logging.Formatter("[%(levelname)s:%(name)s] %(message)s"))
_logger.addHandler(_stream_handler)
def _resource_file_path(fname) -> str:
with importlib_resources.path("minedojo.tasks.description_files", fname) as p:
return str(p)
_MetaTaskName2Class = {
"Open-Ended": MineDojoSim,
"Harvest": HarvestMeta,
"Combat": CombatMeta,
"TechTree": TechTreeMeta,
"Playthrough": Playthrough,
"Survival": SurvivalMeta,
"Creative": CreativeMeta,
}
MetaTaskName2Class = {k.lower(): v for k, v in _MetaTaskName2Class.items()}
def _meta_task_make(meta_task: str, *args, **kwargs) -> MetaTaskBase | FastResetWrapper:
"""
Gym-style making tasks from names.
Usage example:
.. highlight:: python
.. code-block:: python
import minedojo
env = minedojo.make(task_name, *args, **kwargs)
Args:
meta_task: Name of the meta task. Can be one of (and their lower-cased equivalents)
``"Open-Ended"``, ``"Harvest"``, ``"Combat"``, ``"TechTree"``,
``"Playthrough"``, ``"Survival"``, ``"Creative"``.
*args: See corresponding task's docstring for more info.
**kwargs: See corresponding task's docstring for more info.
"""
meta_task = meta_task.lower()
assert (
meta_task in MetaTaskName2Class
), f"Invalid meta task name provided {meta_task}"
if meta_task == "open-ended":
if "fast_reset" in kwargs:
fast_reset = kwargs.pop("fast_reset")
fast_reset_random_teleport_range = kwargs.pop(
"fast_reset_random_teleport_range", None
)
if fast_reset is True:
return FastResetWrapper(
MineDojoSim(*args, **kwargs), fast_reset_random_teleport_range
)
return MetaTaskName2Class[meta_task](*args, **kwargs)
_ALL_TASKS_SPECS_UNFILLED = OmegaConf.load(_resource_file_path("tasks_specs.yaml"))
# check no duplicates
assert len(set(_ALL_TASKS_SPECS_UNFILLED.keys())) == len(_ALL_TASKS_SPECS_UNFILLED)
# all possible variables used to fill specs
_ALL_VARS = {
# Combat
"combat_biomes": ["forest", "plains", "extreme_hills"],
"regular_biomes_mob": [
"cow",
"pig",
"sheep",
"chicken",
],
"regular_biomes_night_mob": [
"zombie",
"spider",
"skeleton",
"creeper",
"witch",
"enderman",
],
"end_mob": ["shulker", "endermite", "enderman",],
"nether_mob": [
"blaze",
"ghast",
"wither_skeleton",
"zombie_pigman",
],
"plains_mob": ["horse", "donkey"],
"weapon_material": ["wooden", "iron", "diamond"],
"armor_material": ["leather", "iron", "diamond"],
# Harvest
"quantity": [1, 8],
## wool and milk
"cow_biomes": ["plains", "extreme_hills", "forest"],
"sheep_biomes": ["plains", "extreme_hills", "forest"],
## mine
"ore_type": ["iron_ore", "gold_ore", "diamond", "redstone", "coal", "cobblestone"],
"pickaxe_material": ["wooden", "stone", "iron", "golden", "diamond"],
## most supported items (default only)
"natural_items": [
"nether_star",
"blaze_rod",
"ghast_tear",
"nether_wart",
"netherrack",
"soul_sand",
"chorus_flower",
"chorus_fruit",
"chorus_plant",
"elytra",
"end_stone",
"ender_pearl",
"apple",
"beef",
"beetroot",
"beetroot_seeds",
"bone",
"brown_mushroom",
"cactus",
"carrot",
"chicken",
"dirt",
"egg",
"feather",
"fish",
"grass",
"leaves",
"log",
"monster_egg",
"mutton",
"porkchop",
"potato",
"prismarine_shard",
"pumpkin",
"rabbit",
"red_mushroom",
"reeds",
"sapling",
"skull",
"snowball",
"spawn_egg",
"sponge",
"string",
"totem_of_undying",
"vine",
"web",
"wheat_seeds",
"wheat",
],
"craft_items": [
"book",
"carrot_on_a_stick",
"clay",
"crafting_table",
"dye",
"end_bricks",
"end_rod",
"ender_eye",
"flint_and_steel",
"glowstone",
"gold_nugget",
"iron_nugget",
"iron_trapdoor",
"lever",
"nether_brick",
"planks",
"pumpkin_seeds",
"red_nether_brick",
"sandstone",
"shears",
"slime_ball",
"stick",
"stone_button",
"stonebrick",
"sugar",
"torch",
"trapped_chest",
"wooden_button",
"wool",
"stone_pressure_plate",
],
"crafting_table_items": [
"anvil",
"arrow",
"banner",
"beacon",
"bed",
"beetroot_soup",
"boat",
"bookshelf",
"bowl",
"bread",
"bucket",
"cake",
"cauldron",
"chest",
"cookie",
"end_crystal",
"ender_chest",
"fence",
"fence_gate",
"fire_charge",
"fishing_rod",
"flower_pot",
"furnace",
"glass_bottle",
"glass_pane",
"golden_apple",
"hopper",
"iron_bars",
"ladder",
"lead",
"map",
"minecart",
"mushroom_stew",
"painting",
"paper",
"pumpkin_pie",
"rabbit_stew",
"rail",
"sea_lantern",
"shield",
"sign",
"speckled_melon",
"stone_slab",
"trapdoor",
"tripwire_hook",
"wooden_door",
"writable_book",
],
"furnace_items": [
"baked_potato",
"brick",
"cooked_beef",
"cooked_chicken",
"cooked_fish",
"cooked_mutton",
"cooked_porkchop",
"cooked_rabbit",
"glass",
"gold_ingot",
"iron_ingot",
"quartz",
"stone",
"emerald",
"netherbrick",
],
## core items
### most of the items here need trees in the biomes
"biome_subset": ["plains", "jungle", "taiga", "forest", "swampland"],
"natural_core": [
"apple",
"beef",
"bone",
"chicken",
"log",
"reeds",
"web",
"wheat",
],
"hand_craft_core": [
"flint_and_steel",
"crafting_table",
"planks",
"shears",
"stick",
"sugar",
"torch",
],
"crafting_table_core": [
"arrow",
"chest",
"shield",
"fishing_rod",
"bucket",
"furnace",
],
"furnace_core": [
"cooked_beef",
"glass",
"gold_ingot",
"iron_ingot",
"brick",
"stone",
],
# Tech-tree
"from_barehand_tools": ["wooden", "stone"],
"from_barehand_tools_armor": ["iron", "golden", "diamond"],
"from_wood_tools": ["stone"],
"from_wood_tools_armor": ["iron", "golden", "diamond"],
"from_stone_tools_armor": [
"iron",
"golden",
"diamond",
],
"from_iron_tools_armor": [
"golden",
"diamond",
],
"from_gold_tools_armor": [
"diamond",
],
"target_tools": [
"sword",
"pickaxe",
"axe",
"hoe",
"shovel",
],
"target_armor": [
"boots",
"chestplate",
"helmet",
"leggings",
],
"target_tools_armor": [
"sword",
"pickaxe",
"axe",
"hoe",
"shovel",
"boots",
"chestplate",
"helmet",
"leggings",
],
"redstone_list": [
"redstone_block",
"clock",
"compass",
"dispenser",
"dropper",
"observer",
"piston",
"redstone_lamp",
"redstone_torch",
"repeater",
"detector_rail",
"comparator",
"activator_rail",
],
}
[docs]def product_dict(**kwargs):
keys = kwargs.keys()
vals = kwargs.values()
for instance in product(*vals):
yield dict(zip(keys, instance))
ALL_TASKS_SPECS = {}
for task_id, task_specs in _ALL_TASKS_SPECS_UNFILLED.items():
unfilled_vars = re.findall(r"\{(.*?)\}", task_id)
def _recursive_find_unfilled_vars(x):
if OmegaConf.is_dict(x):
return {k: _recursive_find_unfilled_vars(v) for k, v in x.items()}
elif isinstance(x, str):
unfilled_vars.extend(re.findall(r"\{(.*?)\}", x))
return x
_recursive_find_unfilled_vars(task_specs)
# deduplicate unfilled vars
unfilled_vars = list(set(unfilled_vars))
if len(unfilled_vars) == 0:
# no unfilled vars, just make the task
ALL_TASKS_SPECS[task_id] = task_specs
else:
unfilled_vars_values = {var: _ALL_VARS[var] for var in unfilled_vars}
for var_dict in product_dict(**unfilled_vars_values):
filled_task_id = task_id.format(**var_dict)
def _recursive_replace_var(x):
if OmegaConf.is_dict(x):
return {k: _recursive_replace_var(v) for k, v in x.items()}
elif isinstance(x, str):
return x.format(**var_dict)
return x
task_specs_filled = _recursive_replace_var(task_specs)
for k, v in task_specs_filled.items():
if k == "target_quantities":
task_specs_filled[k] = int(v)
task_specs_filled["prompt"] = task_specs_filled["prompt"].replace("_", " ")
task_specs_filled["prompt"] = task_specs_filled["prompt"].replace(" 1", "")
ALL_TASKS_SPECS[filled_task_id] = task_specs_filled
# check no duplicates
assert len(set(ALL_TASKS_SPECS.keys())) == len(ALL_TASKS_SPECS)
# load prompts and guidance for programmatic tasks
P_TASKS_PROMPTS_GUIDANCE = OmegaConf.load(
_resource_file_path("programmatic_tasks.yaml")
)
# check no duplicates
assert len(set(P_TASKS_PROMPTS_GUIDANCE.keys())) == len(P_TASKS_PROMPTS_GUIDANCE)
# load prompts and guidance for creative tasks
C_TASKS_PROMPTS_GUIDANCE = OmegaConf.load(_resource_file_path("creative_tasks.yaml"))
# check no duplicates
assert len(set(C_TASKS_PROMPTS_GUIDANCE.keys())) == len(C_TASKS_PROMPTS_GUIDANCE)
# load prompt and guidance for Playthrough task
PLAYTHROUGH_PROMPT_GUIDANCE = OmegaConf.load(
_resource_file_path("playthrough_task.yaml")
)
# check only one playthrough task
assert len(PLAYTHROUGH_PROMPT_GUIDANCE.keys()) == 1
ALL_PROGRAMMATIC_TASK_IDS = list(P_TASKS_PROMPTS_GUIDANCE.keys())
ALL_PROGRAMMATIC_TASK_INSTRUCTIONS = {
task_id: (
P_TASKS_PROMPTS_GUIDANCE[task_id]["prompt"],
P_TASKS_PROMPTS_GUIDANCE[task_id]["guidance"],
)
for task_id in ALL_PROGRAMMATIC_TASK_IDS
}
ALL_CREATIVE_TASK_IDS = list(C_TASKS_PROMPTS_GUIDANCE.keys())
ALL_CREATIVE_TASK_INSTRUCTIONS = {
task_id: (
C_TASKS_PROMPTS_GUIDANCE[task_id]["prompt"],
C_TASKS_PROMPTS_GUIDANCE[task_id]["guidance"],
)
for task_id in ALL_CREATIVE_TASK_IDS
}
PLAYTHROUGH_TASK_ID = list(PLAYTHROUGH_PROMPT_GUIDANCE.keys())[0]
PLAYTHROUGH_TASK_INSTRUCTION = (
PLAYTHROUGH_PROMPT_GUIDANCE[PLAYTHROUGH_TASK_ID]["prompt"],
PLAYTHROUGH_PROMPT_GUIDANCE[PLAYTHROUGH_TASK_ID]["guidance"],
)
ALL_TASK_IDS = ALL_PROGRAMMATIC_TASK_IDS + ALL_CREATIVE_TASK_IDS + [PLAYTHROUGH_TASK_ID]
ALL_TASK_INSTRUCTIONS = {
**ALL_PROGRAMMATIC_TASK_INSTRUCTIONS,
**ALL_CREATIVE_TASK_INSTRUCTIONS,
PLAYTHROUGH_TASK_ID: PLAYTHROUGH_TASK_INSTRUCTION,
}
_logger.info(
f"Loaded {len(P_TASKS_PROMPTS_GUIDANCE)} Programmatic tasks, "
f"{len(C_TASKS_PROMPTS_GUIDANCE)} Creative tasks, "
"""and 1 special task: "Playthrough". """
f"Totally {len(P_TASKS_PROMPTS_GUIDANCE) + len(C_TASKS_PROMPTS_GUIDANCE) + 1} tasks loaded."
)
def _parse_inventory_dict(inv_dict: dict[str, dict]) -> list[InventoryItem]:
return [InventoryItem(slot=k, **v) for k, v in inv_dict.items()]
def _specific_task_make(task_id: str, *args, **kwargs):
assert task_id in ALL_TASKS_SPECS, f"Invalid task id provided {task_id}"
task_specs = ALL_TASKS_SPECS[task_id].copy()
# handle list of inventory items
if "initial_inventory" in task_specs:
kwargs["initial_inventory"] = _parse_inventory_dict(
task_specs["initial_inventory"]
)
task_specs.pop("initial_inventory")
# pop prompt from task specs because it is set from programmatic yaml
task_specs.pop("prompt")
# meta task
meta_task_cls = task_specs.pop("__cls__")
task_obj = _meta_task_make(meta_task_cls, *args, **task_specs, **kwargs)
return task_obj
[docs]def make(task_id: str, *args, cam_interval: int | float = 15, **kwargs):
"""
Make a task. task_id can be one of the following:
1. a task id for Programmatic tasks, e.g., "combat_bat_extreme_hills_barehand"
2. format "creative:{idx}" for the idx-th Creative task
3. "playthrough" or "open-ended" for these two special tasks
4. one of "harvest", "combat", "techtree", and "survival" to creative meta task
"""
if task_id.startswith("creative:"):
creative_idx = int(task_id.split(":")[1])
assert len(C_TASKS_PROMPTS_GUIDANCE) > creative_idx >= 0
info = C_TASKS_PROMPTS_GUIDANCE[task_id]
env_obj = _meta_task_make("creative", *args, **kwargs)
env_obj.specify_prompt(info["prompt"])
env_obj.specify_guidance(info["guidance"])
env_obj.collection = info["collection"]
env_obj.source = info["source"]
elif task_id in P_TASKS_PROMPTS_GUIDANCE:
info = P_TASKS_PROMPTS_GUIDANCE[task_id]
env_obj = _specific_task_make(task_id, *args, **kwargs)
env_obj.specify_prompt(info["prompt"])
env_obj.specify_guidance(info["guidance"])
elif task_id.lower() == PLAYTHROUGH_TASK_ID.lower():
info = PLAYTHROUGH_PROMPT_GUIDANCE[PLAYTHROUGH_TASK_ID]
env_obj = _specific_task_make(task_id, *args, **kwargs)
env_obj.specify_prompt(info["prompt"])
env_obj.specify_guidance(info["guidance"])
elif task_id.lower() in [
"open-ended",
"harvest",
"combat",
"techtree",
"survival",
]:
env_obj = _meta_task_make(task_id, *args, **kwargs)
else:
raise ValueError(f"Invalid task id provided {task_id}")
return ARNNWrapper(env_obj, cam_interval=cam_interval)