AWS Sagemaker использует SM_USER_ARGS (как задокументировано здесь ) в качестве переменной среды, в которой он содержит строку (список) аргументов, передаваемых пользователем. Таким образом, значение переменной среды выглядит следующим образом: '["--test_size","0.2","--random_seed","42", "--not_optmize"]'.
С json.loads() я могу преобразовать эту строку в список python. Хотя я хочу создать абстрактный модуль, который возвращает argparse Namespace таким образом, чтобы остальная часть кода оставалась неизменной, независимо от того, запускаю ли я его локально или в службе AWS Sagemaker.
Итак, по сути, мне нужен код, который получает входные данные ["--test_size","0.2","--random_seed","42", "--not_optmize"] и выходные данные Namespace(test_size=0.2, random_seed='42', not_optmize=True, <other_arguments>... ]).
Помогает ли мне пакет python argparse ? Я пытаюсь выяснить способ, которым мне не нужно повторно реализовывать анализатор argparse.
Вот пример, у меня есть этот файл config.ini:
[Docker]
home_dir = /opt
SM_MODEL_DIR = %(home_dir)s/ml/model
SM_CHANNELS = ["training"]
SM_NUM_GPUS = 1
SM_NUM_CPUS =
SM_LOG_LEVEL = 20
SM_USER_ARGS = ["--test_size","0.2","--random_seed","42"]
SM_INPUT_DIR = %(home_dir)s/ml/input
SM_INPUT_CONFIG_DIR = %(home_dir)s/ml/input/config
SM_OUTPUT_DIR = %(home_dir)s/ml/output
SM_OUTPUT_INTERMEDIATE_DIR = %(home_dir)s/ml/output/intermediate
У меня есть этот класс Argparser:
import argparse
import configparser
import datetime
import json
import multiprocessing
import os
import time
from pathlib import Path
from typing import Any, Dict
from .files import JsonFile, YAMLFile
class ArgParser(ABC):
@abstractmethod
def get_arguments(self) -> Dict[str, Any]:
pass
class AWSArgParser(ArgParser):
def __init__(self):
configuration_file_path = 'config.ini'
self.environment = "Sagemaker" \
if os.environ.get("SM_MODEL_DIR", False) \
else os.environ.get("ENVIRON", "Default")
config = configparser.ConfigParser()
config.read(configuration_file_path)
if self.environment == "Local":
config[self.environment]["home_dir"] = str(pathlib.Path(__file__).parent.absolute())
if self.environment != 'Sagemaker':
config[self.environment]["SM_NUM_CPUS"] = str(multiprocessing.cpu_count())
for key, value in config[self.environment].items():
os.environ[key.upper()] = value
self.parser = argparse.ArgumentParser()
# AWS Sagemaker default environmental arguments
self.parser.add_argument(
'--model_dir',
type=str,
default=os.environ['SM_MODEL_DIR'],
)
self.parser.add_argument(
'--channel_names',
default=json.loads(os.environ['SM_CHANNELS']),
)
self.parser.add_argument(
'--num_gpus',
type=int,
default=os.environ['SM_NUM_GPUS'],
)
self.parser.add_argument(
'--num_cpus',
type=int,
default=os.environ['SM_NUM_CPUS'],
)
self.parser.add_argument(
'--user_args',
default=json.loads(os.environ['SM_USER_ARGS']),
)
self.parser.add_argument(
'--input_dir',
type=str,
default=os.environ['SM_INPUT_DIR'],
)
self.parser.add_argument(
'--input_config_dir',
type=Path,
default=os.environ['SM_INPUT_CONFIG_DIR'],
)
self.parser.add_argument(
'--output_dir',
type=Path,
default=os.environ['SM_OUTPUT_DIR'],
)
# Extra arguments
self.run_tag = datetime.datetime \
.fromtimestamp(time.time()) \
.strftime('%Y-%m-%d-%H-%M-%S')
self.parser.add_argument(
'--run_tag',
default=self.run_tag,
type=str,
help=f"Run tag (default: 'datetime.fromtimestamp')",
)
self.parser.add_argument(
'--environment',
type=str,
default=self.environment,
)
self.args = self.parser.parse_args()
def get_arguments(self) -> Dict[str, Any]:
<parse self.args.user_args>
return self.args
, тогда у меня есть сценарий train:
from utils.arg_parser import AWSArgParser
if __name__ == '__main__':
logger.info(f"Begin train.py")
if os.environ["ENVIRON"] == "Sagemaker":
arg_parser = AWSArgParser()
args = arg_parser.get_arguments()
else:
args = <normal local parse>