nextmv 0.10.0

The Python SDK for Nextmv

Nextmv Python SDK

This is the Python SDK for the Nextmv Platform.

Installation

Requires Python >=3.8. Install using pip:

pip install nextmv

Usage

The Nextmv Python SDK is used to interact with various parts of the Nextmv Platform:

• Local: Run decision models locally in the same way they would be run in Cloud.
• Cloud: Interact with the Nextmv Cloud API.

Local

To run a model locally, you can use the various helper functionality provided by the SDK.

Options

Use options to capture parameters (i.e.: configurations) for the run:

import nextmv

options = nextmv.Options(
    nextmv.Parameter("str_option", str, "default value", "A string option", required=True),
    nextmv.Parameter("int_option", int, 1, "An int option", required=False),
    nextmv.Parameter("float_option", float, 1.0, "A float option", required=False),
    nextmv.Parameter("bool_option", bool, True, "A bool option", required=True),
)

print(options.str_option)
print(options.int_option)
print(options.float_option)
print(options.bool_option)
print(options.to_dict())

By using options, you are able to pass in the values of the parameters with CLI arguments or environment variables.

$ python main.py --help
usage: main.py [options]

Options for main.py. Use command-line arguments (highest precedence) or environment variables.

optiTo exclude the `markdownlint` rule start and end block, you can use the
following syntax in your markdown file:STR_OPTION
        [env var: STR_OPTION] (required) (default: default value) (type: str): A string option
  -int_optRemember to replace `Your markdown content here` with your actual markdown
content.(type: int): An int option
  -float_option FLOAT_OPTION, --float_option FLOAT_OPTION
        [env var: FLOAT_OPTION] (default: 1.0) (type: float): A float option
  -bool_option BOOL_OPTION, --bool_option BOOL_OPTION
        [env var: BOOL_OPTION] (required) (default: True) (type: bool): A bool option

Input

Capture the input data for the run.

• Work with JSONinputs.

  import nextmv
  
  # Read JSON from stdin.
  json_input_1 = nextmv.load_local()
  print(json_input_1.data)
  
  # Can also specify JSON format directly, and read from a file.
  json_input_2 = nextmv.load_local(input_format=nextmv.InputFormat.JSON, path="input.json")
  print(json_input_2.data)
  

• Work with plain, utf-8 encoded, text inputs.

  import nextmv
  
  # Read text from stdin.
  text_input_1 = nextmv.load_local(input_format=nextmv.InputFormat.TEXT)
  print(text_input_1.data)
  
  # Can also read from a file.
  text_input_2 = nextmv.load_local(input_format=nextmv.InputFormat.TEXT, path="input.txt")
  print(text_input_2.data)
  

• Work with multiple CSV files.

  import nextmv
  
  # Read multiple CSV files from a dir named "input".
  csv_archive_input_1 = nextmv.load_local(input_format=nextmv.InputFormat.CSV_ARCHIVE)
  print(csv_archive_input_1.data)
  
  # Read multiple CSV files from a custom dir.
  csv_archive_input_2 = nextmv.load_local(input_format=nextmv.InputFormat.CSV_ARCHIVE, path="custom_dir")
  print(csv_archive_input_2.data)
  

Logging

The Nextmv platform captures logs via stderr. Use the provided functionality to record logs.

import sys

import nextmv

print("0. I do nothing")

nextmv.redirect_stdout()

nextmv.log("1. I log a message to stderr")

print("2. I print a message to stdout")

nextmv.reset_stdout()

print("3. I print another message to stdout")

print("4. I print yet another message to stderr without the logger", file=sys.stderr)

nextmv.log("5. I log a message to stderr using the nextmv module directly")

print("6. I print a message to stdout, again")

After executing it, here are the messages printed to the different streams.

• stdout

  1. I do nothing
  2. I print another message to stdout
  3. I print a message to stdout, again
  

• stderr

  1. I log a message to stderr
  2. I print a message to stdout
  3. I print yet another message to stderr without the logger
  4. I log a message to stderr using the nextmv module directly
  

Output

Write the output data after a run is completed.

• Work with JSON outputs.

  import nextmv
  
  output = nextmv.Output(
      solution={"foo": "bar"},
      statistics=nextmv.Statistics(
          result=nextmv.ResultStatistics(
              duration=1.0,
              value=2.0,
              custom={"custom": "result_value"},
          ),
          run=nextmv.RunStatistics(
              duration=3.0,
              iterations=4,
              custom={"custom": "run_value"},
          ),
      ),
  )
  
  # Write to stdout.
  nextmv.write_local(output)
  
  # Write to a file.
  nextmv.write_local(output, path="output.json")
  

• Work with multple CSV files.

  import nextmv
  
  output = nextmv.Output(
      output_format=nextmv.OutputFormat.CSV_ARCHIVE,
      solution={
          "output": [
              {"name": "Alice", "age": 30},
              {"name": "Bob", "age": 40},
          ],
      },
      statistics=nextmv.Statistics(
          result=nextmv.ResultStatistics(
              duration=1.0,
              value=2.0,
              custom={"custom": "result_value"},
          ),
          run=nextmv.RunStatistics(
              duration=3.0,
              iterations=4,
              custom={"custom": "run_value"},
          ),
      ),
  )
  
  # Write multiple CSV fiules to a dir named "output".
  nextmv.write_local(output)
  
  # Write multiple CSV files to a custom dir.
  nextmv.write_local(output, "custom_dir")
  

Cloud

Before starting:

1. Sign up for a Nextmv account.
2. Get your API key. Go to Team > API Key.

Visit the docs for more information. Make sure that you have your API key set as an environment variable:

export NEXTMV_API_KEY="<YOUR-API-KEY>"

Additionally, you must have a valid app in Nextmv Cloud.

• Make a run and get the results.

  import os
  
  from nextmv.cloud import Application, Client, PollingOptions
  
  input = {
      "defaults": {"vehicles": {"speed": 20}},
      "stops": [
          {
              "id": "Nijō Castle",
              "location": {"lon": 135.748134, "lat": 35.014239},
              "quantity": -1,
          },
          {
              "id": "Kyoto Imperial Palace",
              "location": {"lon": 135.762057, "lat": 35.025431},
              "quantity": -1,
          },
      ],
      "vehicles": [
          {
              "id": "v2",
              "capacity": 2,
              "start_location": {"lon": 135.728898, "lat": 35.039705},
          },
      ],
  }
  
  client = Client(api_key=os.getenv("NEXTMV_API_KEY"))
  app = Application(client=client, id="<YOUR-APP-ID>")
  result = app.new_run_with_result(
      input=input,
      instance_id="latest",
      run_options={"solve.duration": "1s"},
      polling_options=PollingOptions(),  # Customize the polling options.
  )
  print(result.to_dict())