fmdtools.define

Package for modelling system structure and behavior.

The define package provides the building blocks to develop a simulation. Simulations are defined in the sub-classes of the Simulable class (in the block and architecture subpackages), as shown below:

Inheritance of simulable fmdtools classes — Structure of simulable fmdtools subclasses used for developing simulations.

Aside from their internal methods defining behavior, events/indicators, and results, Simulations are additionally composed of internal containers (or sub-attributes) of the class which are defined in their own class.

`fmdtools.define.architecture`	Package for defining system Architecture.
`fmdtools.define.block`	Package for defining system behaviors.
`fmdtools.define.flow`	Package for defining Flow connections between behavioral elements.
`fmdtools.define.object`	Package defining various model building blocks.
`fmdtools.define.base`	Common methods used commonly in model definition constructs.
`fmdtools.define.environment`	Module for representing environments with the `Environment` class.

fmdtools.define.base

Common methods used commonly in model definition constructs.

Includes functions:

get_var():Gets the variable value of the object
set_var():Sets variable of the object to a given value
nest_dict(): Nest a dictionary by a certin number of levels.
set_arg_as_type(): Change argument to given type.
is_iter(): Checks whether a data type should be interpreted as an iterable
is_numeric(): Check if a data type is numeric.
is_bool() : check if a data type is boolean.
is_numeric(): Helper function for Result Class, checks if a given value is numeric
unpack_x(): Unpack an array x as a tuple argument.
array_x(): Pack x into an array.
eq_units(): Find conversion factor between rates and times.
t_key():Used to generate keys for a given (float) time that is queryable as an attribute of an object/dict
round_float(): Round a float to a given precision.
nan_to_x(): Helper function for Result Class, returns nan as zero if present, otherwise returns the number

gen_timerange(): Generates timerange from start/endtime

get_code_atrs(): Get code attributes defining a given object or method.

remove_para(): Remove paragraph newlines in a string.

get_obj_name(): Get the name of an object.

get_memory(): Get the memory an object takes.

get_inheritanc(): Find the bases classes an object inherits from.

fmdtools.define.base.array_x(*x): Translate variable-length x into an array input.

fmdtools.define.base.auto_filename(obj, filename='', suffix='.json'): Create filename for saving a class.

fmdtools.define.base.copy_dict_objs(dic, **kwargs): Copy a dict and its objects.

fmdtools.define.base.dict_from_file(filename, delete=False): Load a json as a dictionary.

fmdtools.define.base.dict_to_json(dic, exclude=[]): Convert a dict to json. Keep keys in exclude out of dict.

fmdtools.define.base.eq_units(rateunit, timeunit)

Find conversion factor from rateunit (str) to timeunit (str).

Options for units are: ‘sec’, ‘min’, ‘hr’, ‘day’, ‘wk’, ‘month’, and ‘year’.

fmdtools.define.base.filter_kwargs(obj, **kwargs)

Get keyword arguments just related to the method being called.

May not work with **kwargs.

fmdtools.define.base.gen_timerange(start_time, end_time, dt=1.0, min_r=7): Generate the times in a given interval given the timestep dt.

fmdtools.define.base.get_code_attrs(obj)

Get a dict of code attributes for a given object or method.

Must be run from file other than where the code was originally written.

Parameters:: obj (Object/method) – Class to get code from.
Returns:: code_attrs – Dict of “source”, “code”, and “docs” code attributes.
Return type:: dict

fmdtools.define.base.get_dict_repr(objdict, one_line=True, with_classname=None)

Get repr for a dict of items in a larger object.

Parameters:

objdict (dict) – Dict of objects to get the repr for.
one_line (bool, optional) – Whether the repr string should fit on one line. The default is True.
with_classname (bool, optional) – Whether to include classnames. The default is None, which includes only if one_line is False.

Returns:

dict_rep – String to use for the repr of the dict.

Return type:

str

fmdtools.define.base.get_inheritance(obj)

Get the base class(es) that the object inherits from.

Parameters:: obj (object) – Object to get base of.
Returns:: classes – Tuple of classes that are the base of the object.
Return type:: tuple

fmdtools.define.base.get_memory(role): Get memory of an object.

fmdtools.define.base.get_methods(obj): Get methods from the given object.

fmdtools.define.base.get_obj_name(obj, role='', basename='')

Get the name of an object.

Parameters:

obj (object) – Object to be graphed (BaseObject, BaseContainer, or other).
role (str) – Role the object plays in the larger system. Determines the name of Containers.

Returns:

name – Name of the object.

Return type:

str

fmdtools.define.base.get_repr(obj, name, with_classname=True, with_name=False, one_line=True)

Get the appropriate repr for the object from a larger object.

Parameters:

obj (object) – Object to get repr from.
name (str) – Name of the object in the larger object.
with_classname (bool, optional) – Whether to include the classname of the object. The default is True.
with_name (bool, optional) – Whether to include the name of the object. The default is False.
one_line (bool, optional) – Whether the repr should fit on one line. The default is True.

Returns:

objrep – String to use for the repr of the object.

Return type:

str

fmdtools.define.base.get_var(obj, var)

Get the variable value of the object.

Parameters:

obj (Object) – Object to get the value from.
var (str/list) – list specifying the attribute (or sub-attribute of the object)

Returns:

var_value – value of the variable

Return type:

any

fmdtools.define.base.is_bool(val)

Check if the value is a boolean.

Examples

>>> is_bool(True)
True
>>> is_bool(1.0)
False
>>> is_bool(np.array([True])[0])
True
>>> is_bool(np.array([1.0])[0])
False

fmdtools.define.base.is_iter(data)

Check whether a data type should be interpreted as an iterable or not.

Returned as a single value or tuple/array.

fmdtools.define.base.is_known_immutable(val): Check if value is known immutable.

fmdtools.define.base.is_known_mutable(val): Check if value is a known mutable.

fmdtools.define.base.is_numeric(val)

Check if a given value is a number.

Examples

>>> is_numeric(1.0)
True
>>> is_numeric("hi")
False
>>> is_numeric(np.array([1.0])[0])
True
>>> is_numeric(np.array(["hi"])[0])
False

fmdtools.define.base.map_obj_fields(obj, *fields, **mapping)

Create a dictionary mapping aspects of a given object obj to known fields.

Parameters:

obj (object) – Object with parameters.
*fields (str) – Names of parameters to get from object
**mapping (kwargs) – Mapping of fields to another name, e.g. x=’y’ if ‘y’ should be the name of x in the returned dict.

Returns:

fielddict – Dictionary of fields with values gotten from the object.

Return type:

dict

fmdtools.define.base.nan_to_x(metric, x=0.0)

Return nan as zero if present, otherwise return the number.

Examples

>>> nan_to_x(1.0)
1.0
>>> nan_to_x(np.nan, 10.0)
10.0

fmdtools.define.base.nest_dict(dic, levels=inf, separator='.', skip=0)

Nest a dictionary a certain number of levels by separator.

Parameters:

dict (dict) – Dictionary to nest. e.g. {‘a.b’: 1.0}
levels (int, optional) – Levels to nest over. The default is float(‘inf’).
separator (str) – Seperator to nest by. The default is “.”
skip (str) – Levels to skip. The default is 0.

Returns:

newhist – Nested dictionary. e.g. {‘a’: {‘b’: 1.0}}

Return type:

dict

fmdtools.define.base.remove_para(source): Remove paragraph newlines in a string (e.g., of code).

fmdtools.define.base.round_float(number, res=1.0, min_r=7): Round floats to a given resolution (avoiding fp errors).

fmdtools.define.base.set_arg_as_type(true_type, new_arg)

Set a given argument as the type true_type.

Parameters:

true_type (class/type) – Class/type to set to
new_arg (value) – Value to set as.

Returns:

new_arg – Value with correct type (if possible).

Return type:

value

fmdtools.define.base.set_var(obj, var, val)

Set variable of the object to a given value.

Parameters:

var (list/tuple of strings) – list of nested attributes
val (attr) – attribute to set the value to

Returns:

flowdict – dict of flows indexed by flownames

Return type:

dict

fmdtools.define.base.t_key(time)

Generate keys for a given (float) time in a queryable format.

e.g. endresults.t10p0, the result at time t=10.0

fmdtools.define.base.unpack_x(*x): Unpack arrays/lists sent from libraries into tuples.

fmdtools.define.base.value_to_jsonable(value, exclude=[]): Make a given value json-able.

fmdtools.define.environment

Module for representing environments with the Environment class.

class fmdtools.define.environment.Environment(name='', root='', glob=[], p={}, s={}, r={}, sp={}, c={}, ga={}, track='default', **kwargs)

Bases: CommsFlow

Class for representing environments (in development).

Environments are CommsFlows in order to readily enable perception as well as sending and recieving of information. In addition to having normal flow properties, they also contain the roles:

TODO: Properly expand create_local, update, send, receive, etc to use ga and coords.

Roles

c: Coords: Representation of gridworld properties
r: Rand: Representation of random variables/rng
ga: GeomArch: Representaion of shapes/forms

Examples

>>> class ExampleEnvironment(Environment):
...    coords_c = ExampleCoords
...    arch_ga = ExGeomArch
>>> env = ExampleEnvironment('env')
>>> env
env ExampleEnvironment
- r=Rand(seed=42)
- c=ExampleCoords()
- ga=ExGeomArch()
>>> env.create_hist([1.0])
c.r.probdens:                   array(1)
c.st:                           array(1)
ga.geoms.ex_point.s.occupied:   array(1)
ga.geoms.ex_point.s.buffer_around: array(1)
ga.geoms.ex_line.s.occupied:    array(1)
ga.geoms.ex_line.s.buffer_around: array(1)
ga.geoms.ex_poly.s.occupied:    array(1)
ga.geoms.ex_poly.s.buffer_around: array(1)

Copies should be identical but independent after copying, e.g.:

>>> e = ExampleEnvironment("env")
>>> e.ga.geoms['ex_point'].s.occupied = True
>>> e.c.st[0, 0] = 1

Given these changes, the copy should have the same states (and not default):

>>> d = e.copy()
>>> d.ga.geoms['ex_point'].s.occupied
True
>>> d.c.st[0, 0]
np.float64(1.0)

It should also be independent, meaning changes don’t effect the original:

>>> d.c.st[0, 1] = 1.0
>>> e.c.st[0, 1]
np.float64(0.0)
>>> d.ga.geoms['ex_line'].s.occupied = True
>>> e.ga.geoms['ex_line'].s.occupied
False

This should also be the case for contained local versions:

>>> e = ExampleEnvironment("env")
>>> hi = e.create_local("hi")
>>> e.hi.ga.geoms['ex_point'].s.occupied=True
>>> d = e.copy()
>>> d.hi.ga.geoms['ex_point'].s.occupied
True

arch_ga: alias of GeomArchitecture

base_type(): Return fmdtools type of the model class.

container_r: alias of Rand

container_sp: alias of SimParam

coords_c: alias of Coords

create_repr(rolenames=['s', 'c', 'ga'], **kwargs): Add details to repr.

reset(): Reset the CommsFlow (and all subflows).

fmdtools.define.pathplan

Module for representing path planners with the PathPlannerBase class.

class fmdtools.define.pathplan.ExampleCoordsParam(x_size: int = 10, y_size: int = 10, blocksize: float = 1.0)

Bases: CoordsParam

Create class ExampleCoordsParam instance

blocksize: float

x_size: int

y_size: int

class fmdtools.define.pathplan.ExampleGeomArch: Bases: GeomArchitecture

class fmdtools.define.pathplan.ExampleGeomPoint(*args, s={}, p={}, track='default', **kwargs)

Bases: GeomPoint

container_p: alias of ExampleObstacleParam

container_s: alias of ExampleObstacleState

class fmdtools.define.pathplan.ExampleGrid(track='default', **kwargs)

Bases: Coords

container_p: alias of ExampleCoordsParam

init_properties(**kwargs): Initialize arrays with non-default values.

class fmdtools.define.pathplan.ExampleObstacleParam(coordinates: tuple = (8.0, 8.0), buffer_around: float = 0.5)

Bases: GeomParameter

Create class ExampleObstacleParam instance

coordinates: tuple

class fmdtools.define.pathplan.ExampleObstacleState(cost: float = 2.0, traversable: bool = False, goal_allowed: bool = False)

Bases: State

Create class ExampleObstacleState instance

class fmdtools.define.pathplan.PathPlannerBase(cost_function=None, **kwargs)

Bases: BaseObject

Universal base path planner for both Coords (grid) and Geom/GeomArchitecture (continuous) environments.

Users can supply a custom cost function via:: planner.cost_function = fn
where fn is one of:: fn(path, planner) -> float fn(path) -> float fn(segment_costs) -> float

check_collision(x, y, geom=None)

Returns True if traversable. - If shape is provided: acts as check_shape_collision.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> pp.check_collision(2,2)
False
>>> pp.check_collision(9,9)
True
>>> pp.init_environment(ExampleGeomArch())
>>> pp.check_collision(8,8)
False
>>> pp.check_collision(1,1)
True

check_goal_feasible(goal, geom=None)

Check whether goal is feasible for a shaped agent.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> # Not traversable (2,2) and not goal_allowed
>>> pp.check_goal_feasible((2,2))['feasible']
False
>>> # Traversable and goal_allowed region!
>>> pp.check_goal_feasible((9,9))['feasible']
True
>>> pp.init_environment(ExampleGeomArch())
>>> # Inside the small circle at (8,8): not traversable and not goal_allowed
>>> pp.check_goal_feasible((8,8))['feasible']
False
>>> # Outside circle: traversable, goal_allowed
>>> pp.check_goal_feasible((1,1))['feasible']
True

check_segment_collision(x1, y1, x2, y2, geom=None, shape_name='shape', resolution=None)

Check if a line segment from (x1, y1) to (x2, y2) is collision-free.

If geom is provided, performs a swept-volume check (shape along segment). If geom is None, performs a point-based check.

Parameters:

x1 (float) – Segment endpoints
y1 (float) – Segment endpoints
x2 (float) – Segment endpoints
y2 (float) – Segment endpoints
geom (GeomPoint/GeomPoly/GeomLine, optional) – The agent shape. If None, treats agent as a point.
shape_name (str) – Name for get_buffered_shape lookup (only used when geom is provided)
resolution (float, optional) – Distance between sample points for shape checks. If None, uses collision_check_resolution.

Returns:

is_free : bool collision_info : list of collision points (point mode) or dict (shape mode)

Return type:

tuple (is_free, collision_info)

check_shape_collision(x, y, geom, shape_name='shape')

Check if shape at (x,y) collides with obstacles.

Parameters:

x (float) – Position to check
y (float) – Position to check
geom (GeomPoint/GeomPoly/GeomLine) – The agent shape (Geom object)

Return type:

tuple (is_traversable, collision_info)

compute_number_path_steps(path)

Computes number of waypoints in a path.

Examples

>>> pp = PathPlannerBase()
>>> pp.compute_number_path_steps([(0,0), (1,1), (2,2)])
3
>>> pp.compute_number_path_steps([])
0

compute_path(start, goal, planner=None, geom=None, **kwargs)

Use external planner to compute a path.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> straight = lambda start, goal, planner, **kw: [start, goal]
>>> path = pp.compute_path((9, 9), (9, 9), planner=straight)
>>> pp.s.planned
True

compute_path_cost(path, geom=None, shape_name='shape', cost_fn=None)

Compute path cost.

If geom is None (point agent): uses the existing trapezoidal-average point-query model (avg endpoint cost × segment length).

If geom is provided (shape agent): sweeps the shape along each segment, accumulating the cost of all cells/obstacles the agent overlaps.

Parameters:

path (list of (x, y) tuples) – The path waypoints.
geom (GeomPoint/GeomPoly/GeomLine, optional) – The agent shape. If None, treat agent as a point.
shape_name (str) – Name for get_buffered_shape lookup.
cost_fn (callable, optional) – User-supplied cost function override.

Returns:

Total path cost. Returns inf if path is infeasible.

Return type:

float

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> pp.compute_path_cost([(1, 1)])
inf

compute_path_length(path)

Compute the total length of a path.

Parameters:: path (sequence of (x, y) tuples) – The path points
Returns:: Total length of the path (sum of segment distances)
Return type:: float

Examples

>>> pp = PathPlannerBase()
>>> round(pp.compute_path_length([(0,0), (3,4)]),2)
5.0
>>> pp.compute_path_length([(0,0)])
0.0
>>> round(pp.compute_path_length([(0,0), (0,5), (5,5)]), 2)
10.0

container_p: alias of PathPlannerParameter

container_s: alias of PathPlannerState

get_buffered_shape(geom_obj, buffer_name=None)

Returns a buffered (footprint) shape from a Geom object for collision checking. Uses specified buffer_name, else first available buffer, else base shape.

Parameters:

geom_obj (Geom) – The geometry object (agent, obstacle, etc.)
buffer_name (str, optional) – The specific buffer name to use (e.g., “on”, “around”, etc). If None, uses the first available buffer.

Returns:

shape – The shapely shape to use for collision checks.

Return type:

shapely.geometry

property get_env

Examples

>>> pp = PathPlannerBase()
>>> pp.get_env
Traceback (most recent call last):
    ...
RuntimeError: Environment not initialized. Call init_environment() first.

init_environment(env)

Initialize with either Coords, Geom, or GeomArchitecture.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(None)
Traceback (most recent call last):
    ...
ValueError: Environment cannot be None.

is_coords()

Returns True if current environment is Coords.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> pp.is_coords()
True
>>> pp.init_environment(ExampleGeomArch())
>>> pp.is_coords()
False
>>> pp.init_environment(ExampleHybrid())
>>> pp.is_coords()
False

is_geom_arch()

Returns True if current environment is GeomArchitecture.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGeomArch())
>>> pp.is_geom_arch()
True
>>> pp.init_environment(ExampleGrid())
>>> pp.is_geom_arch()
False
>>> pp.init_environment(ExampleHybrid())
>>> pp.is_geom_arch()
False

is_hybrid()

Returns True if environment is hybrid (has both grid and geoms).

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleHybrid())
>>> pp.is_hybrid()
True
>>> pp.init_environment(ExampleGrid())
>>> pp.is_hybrid()
False

next_position()

Return the next point in flightplan.

Examples

>>> pp = PathPlannerBase()
>>> pp.s.flightplan = ((0, 0), (1, 1), (2, 2))
>>> pp.s.planned = True
>>> pp.s.pt = 0
>>> pp.next_position()
(0, 0)
>>> pp.next_position()
(1, 1)
>>> pp.next_position()
(2, 2)
>>> pp.next_position() is None
True

plan_and_validate(start, goal, planner=None, geom=None, use_shape_collision=False, **kwargs): Plan, validate, and if needed replan.

query_point(x, y)

Unified hybrid environment point query. Supports: Coords only, Geom only, or BOTH simultaneously

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> # Not traversable, not goal_allowed
>>> pp.query_point(2,2)
{'cost': inf, 'traversable': False, 'goal_allowed': False}
>>> # Traversable region
>>> pp.query_point(9,9)
{'cost': 1.0, 'traversable': True, 'goal_allowed': True}
>>> pp.init_environment(ExampleGeomArch())
>>> # Inside the small circle at (8,8)
>>> pp.query_point(8,8)
{'cost': inf, 'traversable': False, 'goal_allowed': False}
>>> # Outside the circle
>>> pp.query_point(1,1)
{'cost': 0.0, 'traversable': True, 'goal_allowed': True}
>>> pp.init_environment(ExampleHybrid())
>>> # Hybrid at (8,8): both grid and geom allow
>>> pp.query_point(8,8)
{'cost': inf, 'traversable': False, 'goal_allowed': False}
>>> # Only grid (traversable/goal_allowed for grid)
>>> pp.query_point(9,9)
{'cost': 1.0, 'traversable': True, 'goal_allowed': True}
>>> # Only grid at a restricted region
>>> pp.query_point(2,2)
{'cost': inf, 'traversable': False, 'goal_allowed': False}

query_shape(shape)

Query properties of a region defined by a shapely geometry. Useful for checking if a shape-based agent can occupy a position.

Parameters:: shape (shapely.geometry.base.BaseGeometry) – Shape to query (e.g., Point, Polygon, Circle)
Returns:: {“traversable”: bool, “cost”: float, “goal_allowed”: bool, “blocked_cells”: list}
Return type:: dict

validate_path(path, geom=None, use_shape_collision=False)

Validate that all segments are free of collisions.

Examples

>>> pp = PathPlannerBase()
>>> pp.init_environment(ExampleGrid())
>>> pp.validate_path([(1, 1)])
(False, None, {'error': 'Path too short'})

class fmdtools.define.pathplan.PathPlannerParameter(*args, strict_immutability=True, check_type=True, check_pickle=True, set_type=True, check_lim=True, **kwargs)

Bases: Parameter

Static planner parameters (set at initialization).

class fmdtools.define.pathplan.PathPlannerState(*args, check_docs=False, get_fields=True, set_type=True, **kwargs)

Bases: State

Generic planner state: - flightplan: tuple of (x,y) points - planned: whether the plan is valid - pt: current index within plan - last_valid_path: stores last validated path for recovery - replanning_triggered: flag indicating if replanning was needed