Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_validate_valid_path():
tempdir = tempfile.mkdtemp()
assert get_validated_path(tempdir, "out", "default") == tempdir
def test_validate_if_default_is_valid():
tempdir = tempfile.mkdtemp()
assert get_validated_path(None, "out", tempdir) == tempdir
def split_nlu_data(args: argparse.Namespace) -> None:
from rasa.nlu.training_data.loading import load_data
from rasa.nlu.training_data.util import get_file_format
data_path = rasa.cli.utils.get_validated_path(args.nlu, "nlu", DEFAULT_DATA_PATH)
data_path = data.get_nlu_directory(data_path)
nlu_data = load_data(data_path)
fformat = get_file_format(data_path)
train, test = nlu_data.train_test_split(args.training_fraction, args.random_seed)
train.persist(args.out, filename=f"training_data.{fformat}")
test.persist(args.out, filename=f"test_data.{fformat}")
def split_nlu_data(args):
from rasa.nlu.training_data.loading import load_data
from rasa.nlu.training_data.util import get_file_format
data_path = get_validated_path(args.nlu, "nlu", DEFAULT_DATA_PATH)
data_path = data.get_nlu_directory(data_path)
nlu_data = load_data(data_path)
fformat = get_file_format(data_path)
train, test = nlu_data.train_test_split(args.training_fraction)
train.persist(args.out, filename=f"training_data.{fformat}")
test.persist(args.out, filename=f"test_data.{fformat}")
def _get_credentials_and_endpoints_paths(
args: argparse.Namespace
) -> Tuple[Optional[Text], Optional[Text]]:
config_endpoint = args.config_endpoint
if config_endpoint:
loop = asyncio.get_event_loop()
endpoints_config_path, credentials_path = loop.run_until_complete(
_pull_runtime_config_from_server(config_endpoint)
)
else:
endpoints_config_path = cli_utils.get_validated_path(
args.endpoints, "endpoints", DEFAULT_ENDPOINTS_PATH, True
)
credentials_path = None
return credentials_path, endpoints_config_path
def train_nlu(
args: argparse.Namespace, train_path: Optional[Text] = None
) -> Optional[Text]:
from rasa.train import train_nlu
output = train_path or args.out
config = _get_valid_config(args.config, CONFIG_MANDATORY_KEYS_NLU)
nlu_data = get_validated_path(
args.nlu, "nlu", DEFAULT_DATA_PATH, none_is_valid=True
)
return train_nlu(
config=config,
nlu_data=nlu_data,
output=output,
train_path=train_path,
fixed_model_name=args.fixed_model_name,
persist_nlu_training_data=args.persist_nlu_data,
)
def get_provided_model(arg_model: Text):
model_path = get_validated_path(arg_model, "model", DEFAULT_MODELS_PATH)
if os.path.isdir(model_path):
model_path = get_latest_model(model_path)
return model_path
def perform_interactive_learning(args, zipped_model) -> None:
from rasa.core.train import do_interactive_learning
if zipped_model and os.path.exists(zipped_model):
args.model = zipped_model
with model.unpack_model(zipped_model) as model_path:
args.core, args.nlu = model.get_model_subdirectories(model_path)
stories_directory = data.get_core_directory(args.data)
args.endpoints = get_validated_path(
args.endpoints, "endpoints", DEFAULT_ENDPOINTS_PATH, True
)
do_interactive_learning(args, stories_directory)
else:
print_error(
"Interactive learning process cannot be started as no initial model was "
"found. Use 'rasa train' to train a model."
def _prepare_credentials_for_rasa_x(
credentials_path: Optional[Text], rasa_x_url: Optional[Text] = None
) -> Text:
credentials_path = cli_utils.get_validated_path(
credentials_path, "credentials", DEFAULT_CREDENTIALS_PATH, True
)
if credentials_path:
credentials = io_utils.read_config_file(credentials_path)
else:
credentials = {}
# this makes sure the Rasa X is properly configured no matter what
if rasa_x_url:
credentials["rasa"] = {"url": rasa_x_url}
dumped_credentials = yaml.dump(credentials, default_flow_style=False)
tmp_credentials = io_utils.create_temporary_file(dumped_credentials, "yml")
return tmp_credentials