Source code for immuneML.dsl.InstructionParser

import logging
import os
from pathlib import Path

from immuneML.dsl.DefaultParamsLoader import DefaultParamsLoader
from immuneML.dsl.definition_parsers.DefinitionParserOutput import DefinitionParserOutput
from immuneML.dsl.symbol_table.SymbolTable import SymbolTable
from immuneML.dsl.symbol_table.SymbolType import SymbolType
from immuneML.environment.EnvironmentSettings import EnvironmentSettings
from immuneML.hyperparameter_optimization.config.ReportConfig import ReportConfig
from immuneML.hyperparameter_optimization.config.SplitConfig import SplitConfig
from immuneML.util.Logger import log
from immuneML.util.ParameterValidator import ParameterValidator
from immuneML.util.PathBuilder import PathBuilder
from immuneML.util.ReflectionHandler import ReflectionHandler
from immuneML.workflows.instructions.Instruction import Instruction
from immuneML.workflows.instructions.TrainMLModelInstruction import TrainMLModelInstruction
from scripts.DocumentatonFormat import DocumentationFormat
from scripts.specification_util import write_class_docs


[docs]class InstructionParser: keyword = "instructions"
[docs] @staticmethod def parse(definition_output: DefinitionParserOutput, path): specification = definition_output.specification symbol_table = definition_output.symbol_table if InstructionParser.keyword in specification: if len(specification[InstructionParser.keyword].keys()) > 1: logging.warning(f"InstructionParser: multiple instructions were listed in the specification (under keys " f"{str(list(specification[InstructionParser.keyword].keys()))[1:-1]}). " "These instructions are independent and results from one instruction are not available to others. " "If this is the intended behavior, please ignore this warning. If the output of one instruction is needed for the " "other, please use separate YAML specifications and separate runs to perform the analysis.") for key in specification[InstructionParser.keyword]: assert type(specification[InstructionParser.keyword][key]) == dict, f"InstructionParser: instructions are incorrectly defined. Please make sure each instruction is defined under a unique keyword, like this:\n" \ f"instructions:\n" \ f" my_instruction:\n" \ f" type: InstructionType\n" \ f" ... other parameters\n\n" \ f"Found keyword '{key}' containing '{specification[InstructionParser.keyword][key]}', which is a {type(specification[InstructionParser.keyword][key]).__name__}, " \ f"expected a dictionary instead." specification[InstructionParser.keyword][key], symbol_table = \ InstructionParser.parse_instruction(key, specification[InstructionParser.keyword][key], symbol_table, path) else: specification[InstructionParser.keyword] = {} return symbol_table, specification[InstructionParser.keyword]
@staticmethod @log def parse_instruction(key: str, instruction: dict, symbol_table: SymbolTable, path) -> tuple: ParameterValidator.assert_keys_present(list(instruction.keys()), ["type"], InstructionParser.__name__, key) valid_instructions = [cls[:-6] for cls in ReflectionHandler.discover_classes_by_partial_name("Parser", "dsl/instruction_parsers/")] ParameterValidator.assert_in_valid_list(instruction["type"], valid_instructions, "InstructionParser", "type") default_params = DefaultParamsLoader.load("instructions/", instruction["type"]) instruction = {**default_params, **instruction} parser = ReflectionHandler.get_class_by_name("{}Parser".format(instruction["type"]), "instruction_parsers/")() instruction_object = parser.parse(key, instruction, symbol_table, path) symbol_table.add(key, SymbolType.INSTRUCTION, instruction_object) return instruction, symbol_table
[docs] @staticmethod def generate_docs(path: Path): inst_path = PathBuilder.build(path / "instructions") instructions = sorted(ReflectionHandler.all_nonabstract_subclasses(Instruction, "Instruction", subdirectory='instructions/'), key=lambda x: x.__name__) inst_paths = {} for instruction in instructions: instruction_name = instruction.__name__[:-11] if hasattr(InstructionParser, f"make_{instruction_name.lower()}_docs"): fn = getattr(InstructionParser, f"make_{instruction_name.lower()}_docs") file_path = fn(inst_path) else: file_path = InstructionParser.make_docs(instruction, instruction_name, inst_path) inst_paths[instruction_name] = file_path inst_file_path = inst_path / "instructions.rst" with inst_file_path.open('w') as file: for key, item in inst_paths.items(): lines = f"{key}\n---------------------------\n.. include:: {os.path.relpath(item, EnvironmentSettings.source_docs_path)}\n" file.writelines(lines)
[docs] @staticmethod def make_docs(instruction, name, path: Path): file_path = path / f"{name}.rst" with file_path.open("w") as file: write_class_docs(DocumentationFormat(instruction, "", DocumentationFormat.LEVELS[1]), file) return file_path
[docs] @staticmethod def make_trainmlmodel_docs(path): file_path = path / "hp.rst" with file_path.open("w") as file: write_class_docs(DocumentationFormat(TrainMLModelInstruction, "", DocumentationFormat.LEVELS[1]), file) write_class_docs(DocumentationFormat(SplitConfig, "SplitConfig", DocumentationFormat.LEVELS[1]), file) write_class_docs(DocumentationFormat(ReportConfig, "ReportConfig", DocumentationFormat.LEVELS[1]), file) return file_path