Connect your model to Nextmv with the Python SDK¶

#  ___________________________________________________________________________
#
#  Pyomo: Python Optimization Modeling Objects
#  Copyright (c) 2015-2025
#  National Technology and Engineering Solutions of Sandia, LLC
#  Under the terms of Contract DE-NA0003525 with National Technology and
#  Engineering Solutions of Sandia, LLC, the U.S. Government retains certain
#  rights in this software.
#  This software is distributed under the 3-clause BSD License.
#  ___________________________________________________________________________


#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Import
from pyomo.environ import *

# Creation of a Concrete Model
model = ConcreteModel()

## Define sets ##
#  Sets
#       i   canning plants   / seattle, san-diego /
#       j   markets          / new-york, chicago, topeka / ;
model.i = Set(initialize=['seattle','san-diego'], doc='Canning plants')
model.j = Set(initialize=['new-york','chicago', 'topeka'], doc='Markets')

## Define parameters ##
#   Parameters
#       a(i)  capacity of plant i in cases
#         /    seattle     350
#              san-diego   600  /
#       b(j)  demand at market j in cases
#         /    new-york    325
#              chicago     300
#              topeka      275  / ;
model.a = Param(model.i, initialize={'seattle':350,'san-diego':600}, doc='Capacity of plant i in cases')
model.b = Param(model.j, initialize={'new-york':325,'chicago':300,'topeka':275}, doc='Demand at market j in cases')
#  Table d(i,j)  distance in thousands of miles
#                    new-york       chicago      topeka
#      seattle          2.5           1.7          1.8
#      san-diego        2.5           1.8          1.4  ;
dtab = {
    ('seattle',  'new-york') : 2.5,
    ('seattle',  'chicago')  : 1.7,
    ('seattle',  'topeka')   : 1.8,
    ('san-diego','new-york') : 2.5,
    ('san-diego','chicago')  : 1.8,
    ('san-diego','topeka')   : 1.4,
    }
model.d = Param(model.i, model.j, initialize=dtab, doc='Distance in thousands of miles')
#  Scalar f  freight in dollars per case per thousand miles  /90/ ;
model.f = Param(initialize=90, doc='Freight in dollars per case per thousand miles')
#  Parameter c(i,j)  transport cost in thousands of dollars per case ;
#            c(i,j) = f * d(i,j) / 1000 ;
def c_init(model, i, j):
  return model.f * model.d[i,j] / 1000
model.c = Param(model.i, model.j, initialize=c_init, doc='Transport cost in thousands of dollar per case')

## Define variables ##
#  Variables
#       x(i,j)  shipment quantities in cases
#       z       total transportation costs in thousands of dollars ;
#  Positive Variable x ;
model.x = Var(model.i, model.j, bounds=(0.0,None), doc='Shipment quantities in case')

## Define constraints ##
# supply(i)   observe supply limit at plant i
# supply(i) .. sum (j, x(i,j)) =l= a(i)
def supply_rule(model, i):
  return sum(model.x[i,j] for j in model.j) <= model.a[i]
model.supply = Constraint(model.i, rule=supply_rule, doc='Observe supply limit at plant i')
# demand(j)   satisfy demand at market j ;  
# demand(j) .. sum(i, x(i,j)) =g= b(j);
def demand_rule(model, j):
  return sum(model.x[i,j] for i in model.i) >= model.b[j]  
model.demand = Constraint(model.j, rule=demand_rule, doc='Satisfy demand at market j')

## Define Objective and solve ##
#  cost        define objective function
#  cost ..        z  =e=  sum((i,j), c(i,j)*x(i,j)) ;
#  Model transport /all/ ;
#  Solve transport using lp minimizing z ;
def objective_rule(model):
  return sum(model.c[i,j]*model.x[i,j] for i in model.i for j in model.j)
model.objective = Objective(rule=objective_rule, sense=minimize, doc='Define objective function')


## Display of the output ##
# Display x.l, x.m ;
def pyomo_postprocess(options=None, instance=None, results=None):
  model.x.display()

# This is an optional code path that allows the script to be run outside of
# pyomo command-line.  For example:  python transport.py
if __name__ == '__main__':
    # This emulates what the pyomo command-line tools does
    from pyomo.opt import SolverFactory
    import pyomo.environ
    opt = SolverFactory("glpk")
    results = opt.solve(model)
    #sends results to stdout
    results.write()
    print("\nDisplaying Solution\n" + '-'*60)
    pyomo_postprocess(None, model, results)

# Example of a shortest path network flow in a graph
# Shows integration of highspy with networkx

import highspy
import networkx as nx

orig, dest = ('A', 'D')

# create directed graph with edge weights (distances)
G = nx.DiGraph()
G.add_weighted_edges_from([('A', 'B', 2.0), ('B', 'C', 3.0), ('A', 'C', 1.5), ('B', 'D', 2.5), ('C', 'D', 1.0)])

h = highspy.Highs()
h.silent()

x = h.addBinaries(G.edges, obj=nx.get_edge_attributes(G, 'weight'))

# add flow conservation constraints
#                        {  1  if n = orig
#   sum(out) - sum(in) = { -1  if n = dest
#                        {  0     otherwise
rhs  = lambda n: 1 if n == orig else -1 if n == dest else 0
flow = lambda E: h.qsum((x[e] for e in E))

h.addConstrs(flow(G.out_edges(n)) - flow(G.in_edges(n)) == rhs(n) for n in G.nodes)
h.minimize()

# Print the solution
print('Shortest path from', orig, 'to', dest, 'is: ', end = '')
sol = h.vals(x)

n = orig
while n != dest:
    print(n, end=' ')
    n = next(e[1] for e in G.out_edges(n) if sol[e] > 0.5)

print(dest)

#  ___________________________________________________________________________
#
#  Pyomo: Python Optimization Modeling Objects
#  Copyright (c) 2015-2025
#  National Technology and Engineering Solutions of Sandia, LLC
#  Under the terms of Contract DE-NA0003525 with National Technology and
#  Engineering Solutions of Sandia, LLC, the U.S. Government retains certain
#  rights in this software.
#  This software is distributed under the 3-clause BSD License.
#  ___________________________________________________________________________


#!/usr/bin/env python
# -*- coding: utf-8 -*-


import nextmv

# Import
from pyomo.environ import *

# Duration parameter for the solver.
SUPPORTED_PROVIDER_DURATIONS = {
    "cbc": "sec",
    "glpk": "tmlim",
    "scip": "limits/time",
}

# Creation of a Concrete Model
model = ConcreteModel()

nextmv.redirect_stdout()
input = nextmv.load()
options = input.options

## Define sets ##
#  Sets
#       i   canning plants   / seattle, san-diego /
#       j   markets          / new-york, chicago, topeka / ;
sets = input.data["sets"]
model.i = Set(initialize=sets["i"], doc="Canning plants")
model.j = Set(initialize=sets["j"], doc="Markets")

## Define parameters ##
#   Parameters
#       a(i)  capacity of plant i in cases
#         /    seattle     350
#              san-diego   600  /
#       b(j)  demand at market j in cases
#         /    new-york    325
#              chicago     300
#              topeka      275  / ;
params = input.data["parameters"]
model.a = Param(
    model.i,
    initialize=params["a"],
    doc="Capacity of plant i in cases",
)
model.b = Param(
    model.j,
    initialize=params["b"],
    doc="Demand at market j in cases",
)
#  Table d(i,j)  distance in thousands of miles
#                    new-york       chicago      topeka
#      seattle          2.5           1.7          1.8
#      san-diego        2.5           1.8          1.4  ;
dtab = {(edge["from"], edge["to"]): edge["cost"] for edge in params["d"]}
model.d = Param(model.i, model.j, initialize=dtab, doc="Distance in thousands of miles")
#  Scalar f  freight in dollars per case per thousand miles  /90/ ;
model.f = Param(
    initialize=params["f"], doc="Freight in dollars per case per thousand miles"
)


#  Parameter c(i,j)  transport cost in thousands of dollars per case ;
#            c(i,j) = f * d(i,j) / 1000 ;
def c_init(model, i, j):
    return model.f * model.d[i, j] / 1000


model.c = Param(
    model.i,
    model.j,
    initialize=c_init,
    doc="Transport cost in thousands of dollar per case",
)

## Define variables ##
#  Variables
#       x(i,j)  shipment quantities in cases
#       z       total transportation costs in thousands of dollars ;
#  Positive Variable x ;
model.x = Var(model.i, model.j, bounds=(0.0, None), doc="Shipment quantities in case")


## Define constraints ##
# supply(i)   observe supply limit at plant i
# supply(i) .. sum (j, x(i,j)) =l= a(i)
def supply_rule(model, i):
    return sum(model.x[i, j] for j in model.j) <= model.a[i]


model.supply = Constraint(
    model.i, rule=supply_rule, doc="Observe supply limit at plant i"
)


# demand(j)   satisfy demand at market j ;
# demand(j) .. sum(i, x(i,j)) =g= b(j);
def demand_rule(model, j):
    return sum(model.x[i, j] for i in model.i) >= model.b[j]


model.demand = Constraint(model.j, rule=demand_rule, doc="Satisfy demand at market j")


## Define Objective and solve ##
#  cost        define objective function
#  cost ..        z  =e=  sum((i,j), c(i,j)*x(i,j)) ;
#  Model transport /all/ ;
#  Solve transport using lp minimizing z ;
def objective_rule(model):
    return sum(model.c[i, j] * model.x[i, j] for i in model.i for j in model.j)


model.objective = Objective(
    rule=objective_rule, sense=minimize, doc="Define objective function"
)


## Display of the output ##
# Display x.l, x.m ;
def pyomo_postprocess(options=None, instance=None, results=None):
    model.x.display()


# This is an optional code path that allows the script to be run outside of
# pyomo command-line.  For example:  python transport.py
if __name__ == "__main__":
    # This emulates what the pyomo command-line tools does
    from pyomo.opt import SolverFactory

    opt = SolverFactory(options.solver)
    opt.options[SUPPORTED_PROVIDER_DURATIONS[options.solver]] = options.duration
    results = opt.solve(model)
    # sends results to stdout
    results.write()
    print("\nDisplaying Solution\n" + "-" * 60)
    pyomo_postprocess(None, model, results)

    shipments = [
        {"from": ix[0], "to": ix[1], "quantity": x()}
        for ix, x in model.x.items()
        if x() > 0.01
    ]
    solution = {"shipments": shipments}
    metrics = {
        "duration": results.solver.time,
        "value": value(model.objective, exception=False),
        "variables": model.nvariables(),
        "constraints": model.nconstraints(),
        "num_edges_used": len(shipments),
    }
    nextmv.write(solution=solution, metrics=metrics)

# Example of a shortest path network flow in a graph
# Shows integration of highspy with networkx

import highspy
import networkx as nx
import nextmv

nextmv.redirect_stdout()
input = nextmv.load(
    data_files=[
        nextmv.json_data_file("nodes", input_data_key="nodes"),
        nextmv.csv_data_file("edges", input_data_key="edges"),
    ],
)
options = input.options

nodes = input.data["nodes"]
orig, dest = (nodes["origin"], nodes["destination"])

# create directed graph with edge weights (distances)
G = nx.DiGraph()
G.add_weighted_edges_from(
    [(edge["from"], edge["to"], float(edge["weight"])) for edge in input.data["edges"]]
)

h = highspy.Highs()
h.setOptionValue("time_limit", options.duration)

x = h.addBinaries(G.edges, obj=nx.get_edge_attributes(G, "weight"))

# add flow conservation constraints
#                        {  1  if n = orig
#   sum(out) - sum(in) = { -1  if n = dest
#                        {  0     otherwise
rhs = lambda n: 1 if n == orig else -1 if n == dest else 0
flow = lambda E: h.qsum((x[e] for e in E))

h.addConstrs(flow(G.out_edges(n)) - flow(G.in_edges(n)) == rhs(n) for n in G.nodes)
h.minimize()

# Print the solution
shortest_path = []
print("Shortest path from", orig, "to", dest, "is: ", end="")
sol = h.vals(x)

n = orig
while n != dest:
    shortest_path.append(n)
    print(n, end=" ")
    n = next(e[1] for e in G.out_edges(n) if sol[e] > 0.5)

print(dest)

solution = {"shortest_path": shortest_path}
metrics = {
    "duration": h.getRunTime(),
    "value": h.getInfo().objective_function_value,
    "variables": h.numVariables,
    "constraints": h.numConstrs,
    "length_shortest_path": len(shortest_path),
}
nextmv.write(
    options=options,
    metrics=metrics,
    solution_files=[
        nextmv.json_solution_file(name="shortest_path", data=solution),
    ],
)

import os
import time

import nextmv
from nextmv import local

# Instantiate the local application.
local_app = local.Application(src=".")

# Provide any input you want for the app. This input can come from a file, for
# example.
problem = nextmv.load(path=os.path.join("inputs", "problem.json"))
run_id_1 = local_app.new_run(input=problem.data)

# Sleep and get metadata, output (results), logs.
time.sleep(5)

run_information_1 = local_app.run_information(run_id=run_id_1)
nextmv.write(run_information_1)

run_results_1 = local_app.run_result(run_id=run_id_1)
nextmv.write(run_results_1.output)

run_logs_1 = local_app.run_logs(run_id=run_id_1)
print(run_logs_1)

import time

import nextmv
from nextmv import local

# Instantiate the local application.
local_app = local.Application(src=".")

# Provide any input you want for the app. This input can come from a file, for
# example.
run_id_1 = local_app.new_run(input_dir_path="./inputs")

# Sleep and get metadata, output (results), logs.
time.sleep(5)

run_information_1 = local_app.run_information(run_id=run_id_1)
nextmv.write(run_information_1)

run_results_1 = local_app.run_result(run_id=run_id_1, output_dir_path="./outputs")

run_logs_1 = local_app.run_logs(run_id=run_id_1)
print(run_logs_1)

import os

import nextmv
from nextmv import local

# Instantiate the local application.
local_app = local.Application(src=".")

# Start a new run, loading the data from a file, and poll for results.
problem = nextmv.load(path=os.path.join("inputs", "problem.json"))
run_results_2 = local_app.new_run_with_result(input=problem.data)
nextmv.write(run_results_2)  # This time, print both metadata and output at once.

run_logs_2 = local_app.run_logs(run_id=run_results_2.id)
print(run_logs_2)

import nextmv
from nextmv import local

# Instantiate the local application.
local_app = local.Application(src=".")

# Start a new run, loading the file from a directory, and poll for results.
run_results_2 = local_app.new_run_with_result(
    input_dir_path="./inputs",
    output_dir_path="./outputs",
)
nextmv.write(run_results_2)  # This time, print both metadata and output at once.

run_logs_2 = local_app.run_logs(run_id=run_results_2.id)
print(run_logs_2)

import os

from nextmv import cloud

# Instantiate the cloud application.
client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
cloud_app = cloud.Application.new(
    client=client,
    name="test-pyomo-app",
    id="test-pyomo-app",
    exist_ok=True,
)
print("Cloud application created:", cloud_app.id)

import os

from nextmv import cloud

# Instantiate the cloud application.
client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
cloud_app = cloud.Application.new(
    client=client,
    name="test-highs-app",
    id="test-highs-app",
    exist_ok=True,
)
print("Cloud application created:", cloud_app.id)

import os

from nextmv import cloud, local

# Instantiate the local application.
local_app = local.Application(src=".")

# Instantiate the cloud application.
client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
cloud_app = cloud.Application(client=client, id="test-pyomo-app")

# Sync the local app's runs to the cloud app.
local_app.sync(target=cloud_app, verbose=True)