diff --git a/control/iosys.py b/control/iosys.py index 1fe10346f..4e1a6cc2e 100644 --- a/control/iosys.py +++ b/control/iosys.py @@ -76,7 +76,8 @@ class for a set of subclasses that are used to implement specific Parameter values for the systems. Passed to the evaluation functions for the system as default values, overriding internal defaults. name : string, optional - System name (used for specifying signals) + System name (used for specifying signals). If unspecified, a generic + name is generated with a unique integer id. Attributes ---------- @@ -108,6 +109,14 @@ class for a set of subclasses that are used to implement specific The default is to return the entire system state. """ + + idCounter = 0 + def name_or_default(self, name=None): + if name is None: + name = "sys[{}]".format(InputOutputSystem.idCounter) + InputOutputSystem.idCounter += 1 + return name + def __init__(self, inputs=None, outputs=None, states=None, params={}, dt=None, name=None): """Create an input/output system. @@ -143,7 +152,8 @@ def __init__(self, inputs=None, outputs=None, states=None, params={}, functions for the system as default values, overriding internal defaults. name : string, optional - System name (used for specifying signals) + System name (used for specifying signals). If unspecified, a generic + name is generated with a unique integer id. Returns ------- @@ -152,9 +162,9 @@ def __init__(self, inputs=None, outputs=None, states=None, params={}, """ # Store the input arguments - self.params = params.copy() # default parameters - self.dt = dt # timebase - self.name = name # system name + self.params = params.copy() # default parameters + self.dt = dt # timebase + self.name = self.name_or_default(name) # system name # Parse and store the number of inputs, outputs, and states self.set_inputs(inputs) @@ -204,10 +214,12 @@ def __mul__(sys2, sys1): if dt is False: raise ValueError("System timebases are not compabile") + inplist = [(0,i) for i in range(sys1.ninputs)] + outlist = [(1,i) for i in range(sys2.noutputs)] # Return the series interconnection between the systems - newsys = InterconnectedSystem((sys1, sys2)) + newsys = InterconnectedSystem((sys1, sys2), inplist=inplist, outlist=outlist) - # Set up the connecton map + # Set up the connection map manually newsys.set_connect_map(np.block( [[np.zeros((sys1.ninputs, sys1.noutputs)), np.zeros((sys1.ninputs, sys2.noutputs))], @@ -215,18 +227,6 @@ def __mul__(sys2, sys1): np.zeros((sys2.ninputs, sys2.noutputs))]] )) - # Set up the input map - newsys.set_input_map(np.concatenate( - (np.eye(sys1.ninputs), np.zeros((sys2.ninputs, sys1.ninputs))), - axis=0)) - # TODO: set up input names - - # Set up the output map - newsys.set_output_map(np.concatenate( - (np.zeros((sys2.noutputs, sys1.noutputs)), np.eye(sys2.noutputs)), - axis=1)) - # TODO: set up output names - # Return the newly created system return newsys @@ -271,18 +271,10 @@ def __add__(sys1, sys2): ninputs = sys1.ninputs noutputs = sys1.noutputs + inplist = [[(0,i),(1,i)] for i in range(ninputs)] + outlist = [[(0,i),(1,i)] for i in range(noutputs)] # Create a new system to handle the composition - newsys = InterconnectedSystem((sys1, sys2)) - - # Set up the input map - newsys.set_input_map(np.concatenate( - (np.eye(ninputs), np.eye(ninputs)), axis=0)) - # TODO: set up input names - - # Set up the output map - newsys.set_output_map(np.concatenate( - (np.eye(noutputs), np.eye(noutputs)), axis=1)) - # TODO: set up output names + newsys = InterconnectedSystem((sys1, sys2), inplist=inplist, outlist=outlist) # Return the newly created system return newsys @@ -301,16 +293,10 @@ def __neg__(sys): if sys.ninputs is None or sys.noutputs is None: raise ValueError("Can't determine number of inputs or outputs") + inplist = [(0,i) for i in range(sys.ninputs)] + outlist = [(0,i,-1) for i in range(sys.noutputs)] # Create a new system to hold the negation - newsys = InterconnectedSystem((sys,), dt=sys.dt) - - # Set up the input map (identity) - newsys.set_input_map(np.eye(sys.ninputs)) - # TODO: set up input names - - # Set up the output map (negate the output) - newsys.set_output_map(-np.eye(sys.noutputs)) - # TODO: set up output names + newsys = InterconnectedSystem((sys,), dt=sys.dt, inplist=inplist, outlist=outlist) # Return the newly created system return newsys @@ -482,10 +468,13 @@ def feedback(self, other=1, sign=-1, params={}): if dt is False: raise ValueError("System timebases are not compabile") + inplist = [(0,i) for i in range(self.ninputs)] + outlist = [(0,i) for i in range(self.noutputs)] # Return the series interconnection between the systems - newsys = InterconnectedSystem((self, other), params=params, dt=dt) + newsys = InterconnectedSystem((self, other), inplist=inplist, outlist=outlist, + params=params, dt=dt) - # Set up the connecton map + # Set up the connecton map manually newsys.set_connect_map(np.block( [[np.zeros((self.ninputs, self.noutputs)), sign * np.eye(self.ninputs, other.noutputs)], @@ -493,18 +482,6 @@ def feedback(self, other=1, sign=-1, params={}): np.zeros((other.ninputs, other.noutputs))]] )) - # Set up the input map - newsys.set_input_map(np.concatenate( - (np.eye(self.ninputs), np.zeros((other.ninputs, self.ninputs))), - axis=0)) - # TODO: set up input names - - # Set up the output map - newsys.set_output_map(np.concatenate( - (np.eye(self.noutputs), np.zeros((self.noutputs, other.noutputs))), - axis=1)) - # TODO: set up output names - # Return the newly created system return newsys @@ -564,9 +541,11 @@ def linearize(self, x0, u0, t=0, params={}, eps=1e-6): linsys = StateSpace(A, B, C, D, self.dt, remove_useless=False) return LinearIOSystem(linsys) - def copy(self): + def copy(self, newname=None): """Make a copy of an input/output system.""" - return copy.copy(self) + newsys = copy.copy(self) + newsys.name = self.name_or_default("copy of " + self.name if not newname else newname) + return newsys class LinearIOSystem(InputOutputSystem, StateSpace): @@ -610,7 +589,8 @@ def __init__(self, linsys, inputs=None, outputs=None, states=None, functions for the system as default values, overriding internal defaults. name : string, optional - System name (used for specifying signals) + System name (used for specifying signals). If unspecified, a generic + name is generated with a unique integer id. Returns ------- @@ -728,7 +708,8 @@ def __init__(self, updfcn, outfcn=None, inputs=None, outputs=None, * dt = True Discrete time with unspecified sampling time name : string, optional - System name (used for specifying signals). + System name (used for specifying signals). If unspecified, a generic + name is generated with a unique integer id. Returns ------- @@ -808,10 +789,13 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], syslist : array_like of InputOutputSystems The list of input/output systems to be connected - connections : tuple of connection specifications, optional + connections : list of tuple of connection specifications, optional Description of the internal connections between the subsystems. - Each element of the tuple describes an input to one of the - subsystems. The entries are are of the form: + + [connection1, connection2, ...] + + Each connection is a tuple that describes an input to one of the + subsystems. The entries are of the form: (input-spec, output-spec1, output-spec2, ...) @@ -835,10 +819,15 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], If omitted, the connection map (matrix) can be specified using the :func:`~control.InterconnectedSystem.set_connect_map` method. - inplist : tuple of input specifications, optional + inplist : List of tuple of input specifications, optional List of specifications for how the inputs for the overall system are mapped to the subsystem inputs. The input specification is - the same as the form defined in the connection specification. + similar to the form defined in the connection specification, except + that connections do not specify an input-spec, since these are + the system inputs. The entries are thus of the form: + + (output-spec1, output-spec2, ...) + Each system input is added to the input for the listed subsystem. If omitted, the input map can be specified using the @@ -847,7 +836,7 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], outlist : tuple of output specifications, optional List of specifications for how the outputs for the subsystems are mapped to overall system outputs. The output specification is the - same as the form defined in the connection specification + same as the form defined in the inplist specification (including the optional gain term). Numbered outputs must be chosen from the list of subsystem outputs, but named outputs can also be contained in the list of subsystem inputs. @@ -855,6 +844,23 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], If omitted, the output map can be specified using the `set_output_map` method. + inputs : int, list of str or None, optional + Description of the system inputs. This can be given as an integer + count or as a list of strings that name the individual signals. + If an integer count is specified, the names of the signal will be + of the form `s[i]` (where `s` is one of `u`, `y`, or `x`). If + this parameter is not given or given as `None`, the relevant + quantity will be determined when possible based on other + information provided to functions using the system. + + outputs : int, list of str or None, optional + Description of the system outputs. Same format as `inputs`. + + states : int, list of str, or None, optional + Description of the system states. Same format as `inputs`, except + the state names will be of the form '.', + for each subsys in syslist and each state_name of each subsys. + params : dict, optional Parameter values for the systems. Passed to the evaluation functions for the system as default values, overriding internal @@ -871,7 +877,8 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], * dt = True Discrete time with unspecified sampling time name : string, optional - System name (used for specifying signals). + System name (used for specifying signals). If unspecified, a generic + name is generated with a unique integer id. """ # Convert input and output names to lists if they aren't already @@ -885,8 +892,9 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], nstates = 0; self.state_offset = [] ninputs = 0; self.input_offset = [] noutputs = 0; self.output_offset = [] - system_count = 0 - for sys in syslist: + sysobj_name_dct = {} + sysname_count_dct = {} + for sysidx, sys in enumerate(syslist): # Make sure time bases are consistent # TODO: Use lti._find_timebase() instead? if dt is None and sys.dt is not None: @@ -912,36 +920,44 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], ninputs += sys.ninputs noutputs += sys.noutputs - # Store the index to the system for later retrieval - # TODO: look for duplicated system names - self.syslist_index[sys.name] = system_count - system_count += 1 - - # Check for duplicate systems or duplicate names - sysobj_list = [] - sysname_list = [] - for sys in syslist: - if sys in sysobj_list: - warn("Duplicate object found in system list: %s" % str(sys)) - elif sys.name is not None and sys.name in sysname_list: - warn("Duplicate name found in system list: %s" % sys.name) - sysobj_list.append(sys) - sysname_list.append(sys.name) + # Check for duplicate systems or duplicate names + # Duplicates are renamed sysname_1, sysname_2, etc. + if sys in sysobj_name_dct: + sys = sys.copy() + warn("Duplicate object found in system list: %s. Making a copy" % str(sys)) + if sys.name is not None and sys.name in sysname_count_dct: + count = sysname_count_dct[sys.name] + sysname_count_dct[sys.name] += 1 + sysname = sys.name + "_" + str(count) + sysobj_name_dct[sys] = sysname + self.syslist_index[sysname] = sysidx + warn("Duplicate name found in system list. Renamed to {}".format(sysname)) + else: + sysname_count_dct[sys.name] = 1 + sysobj_name_dct[sys] = sys.name + self.syslist_index[sys.name] = sysidx + + if states is None: + states = [] + for sys, sysname in sysobj_name_dct.items(): + states += [sysname + '.' + statename for statename in sys.state_index.keys()] # Create the I/O system super(InterconnectedSystem, self).__init__( inputs=len(inplist), outputs=len(outlist), - states=nstates, params=params, dt=dt) + states=states, params=params, dt=dt, name=name) # If input or output list was specified, update it - nsignals, self.input_index = \ - self._process_signal_list(inputs, prefix='u') - if nsignals is not None and len(inplist) != nsignals: - raise ValueError("Wrong number/type of inputs given.") - nsignals, self.output_index = \ - self._process_signal_list(outputs, prefix='y') - if nsignals is not None and len(outlist) != nsignals: - raise ValueError("Wrong number/type of outputs given.") + if inputs is not None: + nsignals, self.input_index = \ + self._process_signal_list(inputs, prefix='u') + if nsignals is not None and len(inplist) != nsignals: + raise ValueError("Wrong number/type of inputs given.") + if outputs is not None: + nsignals, self.output_index = \ + self._process_signal_list(outputs, prefix='y') + if nsignals is not None and len(outlist) != nsignals: + raise ValueError("Wrong number/type of outputs given.") # Convert the list of interconnections to a connection map (matrix) self.connect_map = np.zeros((ninputs, noutputs)) @@ -960,9 +976,11 @@ def __init__(self, syslist, connections=[], inplist=[], outlist=[], # Convert the output list to a matrix: maps subsystems to system self.output_map = np.zeros((self.noutputs, noutputs + ninputs)) - for index in range(len(outlist)): - ylist_index, gain = self._parse_output_spec(outlist[index]) - self.output_map[index, ylist_index] = gain + for index, outspec in enumerate(outlist): + if isinstance(outspec, (int, str, tuple)): outspec = [outspec] + for spec in outspec: + ylist_index, gain = self._parse_output_spec(spec) + self.output_map[index, ylist_index] = gain # Save the parameters for the system self.params = params.copy() diff --git a/control/tests/iosys_test.py b/control/tests/iosys_test.py index 0738e8b18..9bc03ca20 100644 --- a/control/tests/iosys_test.py +++ b/control/tests/iosys_test.py @@ -763,17 +763,19 @@ def test_named_signals(self): ios_mul = sys1 * sys2 ss_series = self.mimo_linsys1 * self.mimo_linsys2 lin_series = ct.linearize(ios_mul, 0, 0) - for M, N in ((ss_series.A, lin_series.A), (ss_series.B, lin_series.B), - (ss_series.C, lin_series.C), (ss_series.D, lin_series.D)): - np.testing.assert_array_almost_equal(M, N) + np.testing.assert_array_almost_equal(ss_series.A, lin_series.A) + np.testing.assert_array_almost_equal(ss_series.B, lin_series.B) + np.testing.assert_array_almost_equal(ss_series.C, lin_series.C) + np.testing.assert_array_almost_equal(ss_series.D, lin_series.D) # Series interconnection (sys1 * sys2) using series ios_series = ct.series(sys2, sys1) ss_series = ct.series(self.mimo_linsys2, self.mimo_linsys1) lin_series = ct.linearize(ios_series, 0, 0) - for M, N in ((ss_series.A, lin_series.A), (ss_series.B, lin_series.B), - (ss_series.C, lin_series.C), (ss_series.D, lin_series.D)): - np.testing.assert_array_almost_equal(M, N) + np.testing.assert_array_almost_equal(ss_series.A, lin_series.A) + np.testing.assert_array_almost_equal(ss_series.B, lin_series.B) + np.testing.assert_array_almost_equal(ss_series.C, lin_series.C) + np.testing.assert_array_almost_equal(ss_series.D, lin_series.D) # Series interconnection (sys1 * sys2) using named + mixed signals ios_connect = ios.InterconnectedSystem( @@ -786,9 +788,10 @@ def test_named_signals(self): outlist=((1, 'y[0]'), 'sys1.y[1]') ) lin_series = ct.linearize(ios_connect, 0, 0) - for M, N in ((ss_series.A, lin_series.A), (ss_series.B, lin_series.B), - (ss_series.C, lin_series.C), (ss_series.D, lin_series.D)): - np.testing.assert_array_almost_equal(M, N) + np.testing.assert_array_almost_equal(ss_series.A, lin_series.A) + np.testing.assert_array_almost_equal(ss_series.B, lin_series.B) + np.testing.assert_array_almost_equal(ss_series.C, lin_series.C) + np.testing.assert_array_almost_equal(ss_series.D, lin_series.D) # Make sure that we can use input signal names as system outputs ios_connect = ios.InterconnectedSystem( @@ -807,6 +810,123 @@ def test_named_signals(self): np.testing.assert_array_almost_equal(ss_feedback.C, lin_feedback.C) np.testing.assert_array_almost_equal(ss_feedback.D, lin_feedback.D) + def test_sys_naming_convention(self): + """Enforce generic system names 'sys[i]' to be present when systems are created + without explicit names.""" + + ct.InputOutputSystem.idCounter = 0 + sys = ct.LinearIOSystem(self.mimo_linsys1) + self.assertEquals(sys.name, "sys[0]") + self.assertEquals(sys.copy().name, "copy of sys[0]") + + namedsys = ios.NonlinearIOSystem( + updfcn = lambda t, x, u, params: x, + outfcn = lambda t, x, u, params: u, + inputs = ('u[0]', 'u[1]'), + outputs = ('y[0]', 'y[1]'), + states = self.mimo_linsys1.states, + name = 'namedsys') + unnamedsys1 = ct.NonlinearIOSystem( + lambda t,x,u,params: x, inputs=2, outputs=2, states=2 + ) + unnamedsys2 = ct.NonlinearIOSystem( + None, lambda t,x,u,params: u, inputs=2, outputs=2 + ) + self.assertEquals(unnamedsys2.name, "sys[2]") + + # Unnamed/unnamed connections + uu_series = unnamedsys1 * unnamedsys2 + uu_parallel = unnamedsys1 + unnamedsys2 + u_neg = - unnamedsys1 + uu_feedback = unnamedsys2.feedback(unnamedsys1) + uu_dup = unnamedsys1 * unnamedsys1.copy() + uu_hierarchical = uu_series*unnamedsys1 + + self.assertEquals(uu_series.name, "sys[3]") + self.assertEquals(uu_parallel.name, "sys[4]") + self.assertEquals(u_neg.name, "sys[5]") + self.assertEquals(uu_feedback.name, "sys[6]") + self.assertEquals(uu_dup.name, "sys[7]") + self.assertEquals(uu_hierarchical.name, "sys[8]") + + # Unnamed/named connections + un_series = unnamedsys1 * namedsys + un_parallel = unnamedsys1 + namedsys + un_feedback = unnamedsys2.feedback(namedsys) + un_dup = unnamedsys1 * namedsys.copy() + un_hierarchical = uu_series*unnamedsys1 + + self.assertEquals(un_series.name, "sys[9]") + self.assertEquals(un_parallel.name, "sys[10]") + self.assertEquals(un_feedback.name, "sys[11]") + self.assertEquals(un_dup.name, "sys[12]") + self.assertEquals(un_hierarchical.name, "sys[13]") + + # Same system conflict + with warnings.catch_warnings(record=True) as warnval: + unnamedsys1 * unnamedsys1 + self.assertEqual(len(warnval), 1) + + def test_signals_naming_convention(self): + """Enforce generic names to be present when systems are created + without explicit signal names: + input: 'u[i]' + state: 'x[i]' + output: 'y[i]' + """ + ct.InputOutputSystem.idCounter = 0 + sys = ct.LinearIOSystem(self.mimo_linsys1) + for statename in ["x[0]", "x[1]"]: + self.assertTrue(statename in sys.state_index) + for inputname in ["u[0]", "u[1]"]: + self.assertTrue(inputname in sys.input_index) + for outputname in ["y[0]", "y[1]"]: + self.assertTrue(outputname in sys.output_index) + self.assertEqual(len(sys.state_index), sys.nstates) + self.assertEqual(len(sys.input_index), sys.ninputs) + self.assertEqual(len(sys.output_index), sys.noutputs) + + namedsys = ios.NonlinearIOSystem( + updfcn = lambda t, x, u, params: x, + outfcn = lambda t, x, u, params: u, + inputs = ('u0'), + outputs = ('y0'), + states = ('x0'), + name = 'namedsys') + unnamedsys = ct.NonlinearIOSystem( + lambda t,x,u,params: x, inputs=1, outputs=1, states=1 + ) + self.assertTrue('u0' in namedsys.input_index) + self.assertTrue('y0' in namedsys.output_index) + self.assertTrue('x0' in namedsys.state_index) + + # Unnamed/named connections + un_series = unnamedsys * namedsys + un_parallel = unnamedsys + namedsys + un_feedback = unnamedsys.feedback(namedsys) + un_dup = unnamedsys * namedsys.copy() + un_hierarchical = un_series*unnamedsys + u_neg = - unnamedsys + + self.assertTrue("sys[1].x[0]" in un_series.state_index) + self.assertTrue("namedsys.x0" in un_series.state_index) + self.assertTrue("sys[1].x[0]" in un_parallel.state_index) + self.assertTrue("namedsys.x0" in un_series.state_index) + self.assertTrue("sys[1].x[0]" in un_feedback.state_index) + self.assertTrue("namedsys.x0" in un_feedback.state_index) + self.assertTrue("sys[1].x[0]" in un_dup.state_index) + self.assertTrue("copy of namedsys.x0" in un_dup.state_index) + self.assertTrue("sys[1].x[0]" in un_hierarchical.state_index) + self.assertTrue("sys[2].sys[1].x[0]" in un_hierarchical.state_index) + self.assertTrue("sys[1].x[0]" in u_neg.state_index) + + # Same system conflict + with warnings.catch_warnings(record=True) as warnval: + same_name_series = unnamedsys * unnamedsys + self.assertEquals(len(warnval), 1) + self.assertTrue("sys[1].x[0]" in same_name_series.state_index) + self.assertTrue("copy of sys[1].x[0]" in same_name_series.state_index) + def test_named_signals_linearize_inconsistent(self): """Mare sure that providing inputs or outputs not consistent with updfcn or outfcn fail @@ -904,8 +1024,9 @@ def test_lineariosys_statespace(self): np.testing.assert_array_equal(io_feedback.D, ss_feedback.D) def test_duplicates(self): - nlios = ios.NonlinearIOSystem(None, \ - lambda t, x, u, params: u*u, inputs=1, outputs=1) + nlios = ios.NonlinearIOSystem(lambda t,x,u,params: x, \ + lambda t, x, u, params: u*u, \ + inputs=1, outputs=1, states=1, name="sys") # Turn off deprecation warnings warnings.simplefilter("ignore", category=DeprecationWarning) @@ -926,7 +1047,11 @@ def test_duplicates(self): nlios2 = nlios.copy() with warnings.catch_warnings(record=True) as warnval: ios_series = nlios1 * nlios2 - self.assertEqual(len(warnval), 0) + self.assertEquals(len(warnval), 1) + # when subsystems have the same name, duplicates are + # renamed + self.assertTrue("copy of sys_1.x[0]" in ios_series.state_index.keys()) + self.assertTrue("copy of sys.x[0]" in ios_series.state_index.keys()) # Duplicate names iosys_siso = ct.LinearIOSystem(self.siso_linsys) diff --git a/control/timeresp.py b/control/timeresp.py index 8670c180d..0b79a01af 100644 --- a/control/timeresp.py +++ b/control/timeresp.py @@ -408,8 +408,8 @@ def forced_response(sys, T=None, U=0., X0=0., transpose=False, xout = xout[::inc, :] # Transpose the output and state vectors to match local convention - xout = sp.transpose(xout) - yout = sp.transpose(yout) + xout = np.transpose(xout) + yout = np.transpose(yout) # Get rid of unneeded dimensions if squeeze: