Skip to content

Commit 177cba9

Browse files
authored
Merge pull request #849 from murrayrm/cds112_fixes-12Jan2023
Small fixes based on Caltech CDS 112 course
2 parents 08dcc95 + 574cf69 commit 177cba9

11 files changed

Lines changed: 455 additions & 140 deletions

File tree

‎control/flatsys/flatsys.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -436,6 +436,12 @@ def point_to_point(
436436
warnings.warn("basis too small; solution may not exist")
437437

438438
if cost is not None or trajectory_constraints is not None:
439+
# Make sure that we have enough timepoints to evaluate
440+
if timepts.size < 3:
441+
raise ControlArgument(
442+
"There must be at least three time points if trajectory"
443+
" cost or constraints are specified")
444+
439445
# Search over the null space to minimize cost/satisfy constraints
440446
N = sp.linalg.null_space(M)
441447

‎control/iosys.py‎

Lines changed: 65 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -676,6 +676,12 @@ def __init__(self, linsys, **kwargs):
676676
StateSpace.__init__(
677677
self, linsys, remove_useless_states=False, init_namedio=False)
678678

679+
# When sampling a LinearIO system, return a LinearIOSystem
680+
def sample(self, *args, **kwargs):
681+
return LinearIOSystem(StateSpace.sample(self, *args, **kwargs))
682+
683+
sample.__doc__ = StateSpace.sample.__doc__
684+
679685
# The following text needs to be replicated from StateSpace in order for
680686
# this entry to show up properly in sphinx doccumentation (not sure why,
681687
# but it was the only way to get it to work).
@@ -1373,13 +1379,11 @@ def unused_signals(self):
13731379
-------
13741380
13751381
unused_inputs : dict
1376-
13771382
A mapping from tuple of indices (isys, isig) to string
13781383
'{sys}.{sig}', for all unused subsystem inputs.
13791384
13801385
unused_outputs : dict
1381-
1382-
A mapping from tuple of indices (isys, isig) to string
1386+
A mapping from tuple of indices (osys, osig) to string
13831387
'{sys}.{sig}', for all unused subsystem outputs.
13841388
13851389
"""
@@ -1433,10 +1437,13 @@ def _find_outputs_by_basename(self, basename):
14331437
for sig, isig in sys.output_index.items()
14341438
if sig == (basename)}
14351439

1436-
def check_unused_signals(self, ignore_inputs=None, ignore_outputs=None):
1440+
def check_unused_signals(
1441+
self, ignore_inputs=None, ignore_outputs=None, warning=True):
14371442
"""Check for unused subsystem inputs and outputs
14381443
1439-
If any unused inputs or outputs are found, emit a warning.
1444+
Check to see if there are any unused signals and return a list of
1445+
unused input and output signal descriptions. If `warning` is True
1446+
and any unused inputs or outputs are found, emit a warning.
14401447
14411448
Parameters
14421449
----------
@@ -1454,6 +1461,16 @@ def check_unused_signals(self, ignore_inputs=None, ignore_outputs=None):
14541461
If the 'sig' form is used, all subsystem outputs with that
14551462
name are considered ignored.
14561463
1464+
Returns
1465+
-------
1466+
dropped_inputs: list of tuples
1467+
A list of the dropped input signals, with each element of the
1468+
list in the form of (isys, isig).
1469+
1470+
dropped_outputs: list of tuples
1471+
A list of the dropped output signals, with each element of the
1472+
list in the form of (osys, osig).
1473+
14571474
"""
14581475

14591476
if ignore_inputs is None:
@@ -1477,7 +1494,7 @@ def check_unused_signals(self, ignore_inputs=None, ignore_outputs=None):
14771494
ignore_input_map[self._parse_signal(
14781495
ignore_input, 'input')[:2]] = ignore_input
14791496

1480-
# (isys, isig) -> signal-spec
1497+
# (osys, osig) -> signal-spec
14811498
ignore_output_map = {}
14821499
for ignore_output in ignore_outputs:
14831500
if isinstance(ignore_output, str) and '.' not in ignore_output:
@@ -1496,30 +1513,32 @@ def check_unused_signals(self, ignore_inputs=None, ignore_outputs=None):
14961513
used_ignored_inputs = set(ignore_input_map) - set(unused_inputs)
14971514
used_ignored_outputs = set(ignore_output_map) - set(unused_outputs)
14981515

1499-
if dropped_inputs:
1516+
if warning and dropped_inputs:
15001517
msg = ('Unused input(s) in InterconnectedSystem: '
15011518
+ '; '.join(f'{inp}={unused_inputs[inp]}'
15021519
for inp in dropped_inputs))
15031520
warn(msg)
15041521

1505-
if dropped_outputs:
1522+
if warning and dropped_outputs:
15061523
msg = ('Unused output(s) in InterconnectedSystem: '
15071524
+ '; '.join(f'{out} : {unused_outputs[out]}'
15081525
for out in dropped_outputs))
15091526
warn(msg)
15101527

1511-
if used_ignored_inputs:
1528+
if warning and used_ignored_inputs:
15121529
msg = ('Input(s) specified as ignored is (are) used: '
15131530
+ '; '.join(f'{inp} : {ignore_input_map[inp]}'
15141531
for inp in used_ignored_inputs))
15151532
warn(msg)
15161533

1517-
if used_ignored_outputs:
1534+
if warning and used_ignored_outputs:
15181535
msg = ('Output(s) specified as ignored is (are) used: '
15191536
+ '; '.join(f'{out}={ignore_output_map[out]}'
15201537
for out in used_ignored_outputs))
15211538
warn(msg)
15221539

1540+
return dropped_inputs, dropped_outputs
1541+
15231542

15241543
class LinearICSystem(InterconnectedSystem, LinearIOSystem):
15251544

@@ -2580,9 +2599,10 @@ def tf2io(*args, **kwargs):
25802599

25812600

25822601
# Function to create an interconnected system
2583-
def interconnect(syslist, connections=None, inplist=None, outlist=None,
2584-
params=None, check_unused=True, ignore_inputs=None,
2585-
ignore_outputs=None, warn_duplicate=None, **kwargs):
2602+
def interconnect(
2603+
syslist, connections=None, inplist=None, outlist=None, params=None,
2604+
check_unused=True, add_unused=False, ignore_inputs=None,
2605+
ignore_outputs=None, warn_duplicate=None, **kwargs):
25862606
"""Interconnect a set of input/output systems.
25872607
25882608
This function creates a new system that is an interconnection of a set of
@@ -2653,8 +2673,8 @@ def interconnect(syslist, connections=None, inplist=None, outlist=None,
26532673
the system input connects to only one subsystem input, a single input
26542674
specification can be given (without the inner list).
26552675
2656-
If omitted, the input map can be specified using the
2657-
:func:`~control.InterconnectedSystem.set_input_map` method.
2676+
If omitted the `input` parameter will be used to identify the list
2677+
of input signals to the overall system.
26582678
26592679
outlist : list of output connections, optional
26602680
List of connections for how the outputs from the subsystems are mapped
@@ -2707,12 +2727,16 @@ def interconnect(syslist, connections=None, inplist=None, outlist=None,
27072727
System name (used for specifying signals). If unspecified, a generic
27082728
name <sys[id]> is generated with a unique integer id.
27092729
2710-
check_unused : bool
2730+
check_unused : bool, optional
27112731
If True, check for unused sub-system signals. This check is
27122732
not done if connections is False, and neither input nor output
27132733
mappings are specified.
27142734
2715-
ignore_inputs : list of input-spec
2735+
add_unused : bool, optional
2736+
If True, subsystem signals that are not connected to other components
2737+
are added as inputs and outputs of the interconnected system.
2738+
2739+
ignore_inputs : list of input-spec, optional
27162740
A list of sub-system inputs known not to be connected. This is
27172741
*only* used in checking for unused signals, and does not
27182742
disable use of the input.
@@ -2722,7 +2746,7 @@ def interconnect(syslist, connections=None, inplist=None, outlist=None,
27222746
signals from all sub-systems with that base name are
27232747
considered ignored.
27242748
2725-
ignore_outputs : list of output-spec
2749+
ignore_outputs : list of output-spec, optional
27262750
A list of sub-system outputs known not to be connected. This
27272751
is *only* used in checking for unused signals, and does not
27282752
disable use of the output.
@@ -2732,7 +2756,7 @@ def interconnect(syslist, connections=None, inplist=None, outlist=None,
27322756
outputs from all sub-systems with that base name are
27332757
considered ignored.
27342758
2735-
warn_duplicate : None, True, or False
2759+
warn_duplicate : None, True, or False, optional
27362760
Control how warnings are generated if duplicate objects or names are
27372761
detected. In `None` (default), then warnings are generated for
27382762
systems that have non-generic names. If `False`, warnings are not
@@ -2886,10 +2910,30 @@ def interconnect(syslist, connections=None, inplist=None, outlist=None,
28862910
outlist = new_outlist
28872911

28882912
newsys = InterconnectedSystem(
2889-
syslist, connections=connections, inplist=inplist, outlist=outlist,
2890-
inputs=inputs, outputs=outputs, states=states,
2913+
syslist, connections=connections, inplist=inplist,
2914+
outlist=outlist, inputs=inputs, outputs=outputs, states=states,
28912915
params=params, dt=dt, name=name, warn_duplicate=warn_duplicate)
28922916

2917+
# See if we should add any signals
2918+
if add_unused:
2919+
# Get all unused signals
2920+
dropped_inputs, dropped_outputs = newsys.check_unused_signals(
2921+
ignore_inputs, ignore_outputs, warning=False)
2922+
2923+
# Add on any unused signals that we aren't ignoring
2924+
for isys, isig in dropped_inputs:
2925+
inplist.append((isys, isig))
2926+
inputs.append(newsys.syslist[isys].input_labels[isig])
2927+
for osys, osig in dropped_outputs:
2928+
outlist.append((osys, osig))
2929+
outputs.append(newsys.syslist[osys].output_labels[osig])
2930+
2931+
# Rebuild the system with new inputs/outputs
2932+
newsys = InterconnectedSystem(
2933+
syslist, connections=connections, inplist=inplist,
2934+
outlist=outlist, inputs=inputs, outputs=outputs, states=states,
2935+
params=params, dt=dt, name=name, warn_duplicate=warn_duplicate)
2936+
28932937
# check for implicitly dropped signals
28942938
if check_unused:
28952939
newsys.check_unused_signals(ignore_inputs, ignore_outputs)

‎control/optimal.py‎

Lines changed: 35 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
from . import config
2020
from .exception import ControlNotImplemented
21-
from .timeresp import TimeResponseData
2221

2322
# Define module default parameter values
2423
_optimal_trajectory_methods = {'shooting', 'collocation'}
@@ -105,14 +104,14 @@ class OptimalControlProblem():
105104
106105
Notes
107106
-----
108-
To describe an optimal control problem we need an input/output system, a
109-
time horizon, a cost function, and (optionally) a set of constraints on
110-
the state and/or input, either along the trajectory and at the terminal
111-
time. This class sets up an optimization over the inputs at each point in
112-
time, using the integral and terminal costs as well as the trajectory and
113-
terminal constraints. The `compute_trajectory` method sets up an
114-
optimization problem that can be solved using
115-
:func:`scipy.optimize.minimize`.
107+
To describe an optimal control problem we need an input/output system,
108+
a set of time points over a a fixed horizon, a cost function, and
109+
(optionally) a set of constraints on the state and/or input, either
110+
along the trajectory and at the terminal time. This class sets up an
111+
optimization over the inputs at each point in time, using the integral
112+
and terminal costs as well as the trajectory and terminal constraints.
113+
The `compute_trajectory` method sets up an optimization problem that
114+
can be solved using :func:`scipy.optimize.minimize`.
116115
117116
The `_cost_function` method takes the information computes the cost of the
118117
trajectory generated by the proposed input. It does this by calling a
@@ -140,7 +139,8 @@ class OptimalControlProblem():
140139
def __init__(
141140
self, sys, timepts, integral_cost, trajectory_constraints=[],
142141
terminal_cost=None, terminal_constraints=[], initial_guess=None,
143-
trajectory_method=None, basis=None, log=False, **kwargs):
142+
trajectory_method=None, basis=None, log=False, kwargs_check=True,
143+
**kwargs):
144144
"""Set up an optimal control problem."""
145145
# Save the basic information for use later
146146
self.system = sys
@@ -183,7 +183,7 @@ def __init__(
183183
" discrete time systems")
184184

185185
# Make sure there were no extraneous keywords
186-
if kwargs:
186+
if kwargs_check and kwargs:
187187
raise TypeError("unrecognized keyword(s): ", str(kwargs))
188188

189189
self.trajectory_constraints = _process_constraints(
@@ -829,7 +829,7 @@ def compute_mpc(self, x, squeeze=None):
829829
return res.inputs[:, 0]
830830

831831
# Create an input/output system implementing an MPC controller
832-
def create_mpc_iosystem(self):
832+
def create_mpc_iosystem(self, **kwargs):
833833
"""Create an I/O system implementing an MPC controller"""
834834
# Check to make sure we are in discrete time
835835
if self.system.dt == 0:
@@ -857,11 +857,17 @@ def _output(t, x, u, params={}):
857857
res = self.compute_trajectory(u, print_summary=False)
858858
return res.inputs[:, 0]
859859

860+
# Define signal names, if they are not already given
861+
if not kwargs.get('inputs'):
862+
kwargs['inputs'] = self.system.state_labels
863+
if not kwargs.get('outputs'):
864+
kwargs['outputs'] = self.system.input_labels
865+
if not kwargs.get('states'):
866+
kwargs['states'] = self.system.ninputs * \
867+
(self.timepts.size if self.basis is None else self.basis.N)
868+
860869
return ct.NonlinearIOSystem(
861-
_update, _output, dt=self.system.dt,
862-
inputs=self.system.nstates, outputs=self.system.ninputs,
863-
states=self.system.ninputs * \
864-
(self.timepts.size if self.basis is None else self.basis.N))
870+
_update, _output, dt=self.system.dt, **kwargs)
865871

866872

867873
# Optimal control result
@@ -923,7 +929,7 @@ def __init__(
923929
print("* Final cost:", self.cost)
924930

925931
# Process data as a time response (with "outputs" = inputs)
926-
response = TimeResponseData(
932+
response = ct.TimeResponseData(
927933
ocp.timepts, inputs, states, issiso=ocp.system.issiso(),
928934
transpose=transpose, return_x=return_states, squeeze=squeeze)
929935

@@ -934,7 +940,7 @@ def __init__(
934940

935941
# Compute the input for a nonlinear, (constrained) optimal control problem
936942
def solve_ocp(
937-
sys, horizon, X0, cost, trajectory_constraints=None, terminal_cost=None,
943+
sys, timepts, X0, cost, trajectory_constraints=None, terminal_cost=None,
938944
terminal_constraints=[], initial_guess=None, basis=None, squeeze=None,
939945
transpose=None, return_states=True, print_summary=True, log=False,
940946
**kwargs):
@@ -946,7 +952,7 @@ def solve_ocp(
946952
sys : InputOutputSystem
947953
I/O system for which the optimal input will be computed.
948954
949-
horizon : 1D array_like
955+
timepts : 1D array_like
950956
List of times at which the optimal input should be computed.
951957
952958
X0: array-like or number, optional
@@ -984,9 +990,9 @@ def solve_ocp(
984990
985991
initial_guess : 1D or 2D array_like
986992
Initial inputs to use as a guess for the optimal input. The inputs
987-
should either be a 2D vector of shape (ninputs, horizon) or a 1D
988-
input of shape (ninputs,) that will be broadcast by extension of the
989-
time axis.
993+
should either be a 2D vector of shape (ninputs, len(timepts)) or a
994+
1D input of shape (ninputs,) that will be broadcast by extension of
995+
the time axis.
990996
991997
log : bool, optional
992998
If `True`, turn on logging messages (using Python logging module).
@@ -1063,7 +1069,7 @@ def solve_ocp(
10631069

10641070
# Set up the optimal control problem
10651071
ocp = OptimalControlProblem(
1066-
sys, horizon, cost, trajectory_constraints=trajectory_constraints,
1072+
sys, timepts, cost, trajectory_constraints=trajectory_constraints,
10671073
terminal_cost=terminal_cost, terminal_constraints=terminal_constraints,
10681074
initial_guess=initial_guess, basis=basis, log=log, **kwargs)
10691075

@@ -1075,12 +1081,12 @@ def solve_ocp(
10751081

10761082
# Create a model predictive controller for an optimal control problem
10771083
def create_mpc_iosystem(
1078-
sys, horizon, cost, constraints=[], terminal_cost=None,
1084+
sys, timepts, cost, constraints=[], terminal_cost=None,
10791085
terminal_constraints=[], log=False, **kwargs):
10801086
"""Create a model predictive I/O control system
10811087
10821088
This function creates an input/output system that implements a model
1083-
predictive control for a system given the time horizon, cost function and
1089+
predictive control for a system given the time points, cost function and
10841090
constraints that define the finite-horizon optimization that should be
10851091
carried out at each state.
10861092
@@ -1089,7 +1095,7 @@ def create_mpc_iosystem(
10891095
sys : InputOutputSystem
10901096
I/O system for which the optimal input will be computed.
10911097
1092-
horizon : 1D array_like
1098+
timepts : 1D array_like
10931099
List of times at which the optimal input should be computed.
10941100
10951101
cost : callable
@@ -1127,12 +1133,12 @@ def create_mpc_iosystem(
11271133
"""
11281134
# Set up the optimal control problem
11291135
ocp = OptimalControlProblem(
1130-
sys, horizon, cost, trajectory_constraints=constraints,
1136+
sys, timepts, cost, trajectory_constraints=constraints,
11311137
terminal_cost=terminal_cost, terminal_constraints=terminal_constraints,
1132-
log=log, **kwargs)
1138+
log=log, kwargs_check=False, **kwargs)
11331139

11341140
# Return an I/O system implementing the model predictive controller
1135-
return ocp.create_mpc_iosystem()
1141+
return ocp.create_mpc_iosystem(**kwargs)
11361142

11371143

11381144
#

0 commit comments

Comments
 (0)