admin, Author at Robotics with ROS

To develop a software based on “Test Driven Development” one should follow the following steps:

1)Write enough failing Test Code (include compile time and run time failures)
2)Write Production code to pass the failing tests.
3)Refactor the production code and verify the same with existing test.
4)Go to step 1

1)Write enough failing Test Code (include compile time and run time failures)

2)Write Production code to pass the failing tests.

3)Refactor the production code and verify the same with existing test.

4)Go to step 1

so let say we want to develop a binary search tree, here I’m using Google Test. Step 1, Write enough failing Test Code so the first step would be something like this:

class testingTree: public ::testing::Test
{
protected:
    testingTree(){}
    ~testingTree(){}
    void SetUp(){}
    void TearDown(){}
};

TEST_F(testingTree,makeTree)
{
    std::shared_ptr<Tree> tree=makeTree();
    EXPECT_NE(tree.get(),nullptr);
}


int main(int argc, char ** argv)
{
    testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}

class testingTree: public ::testing::Test

{

protected:

testingTree(){}

~testingTree(){}

void SetUp(){}

void TearDown(){}

};

TEST_F(testingTree,makeTree)

{

std::shared_ptr<Tree> tree=makeTree();

EXPECT_NE(tree.get(),nullptr);

}

int main(int argc, char ** argv)

{

testing::InitGoogleTest(&argc, argv);

return RUN_ALL_TESTS();

}

Step 2, Write Production code to

Example of Test Driven Development with C++ and Google Test Read More »

Dijkstra’s algorithm with C++

1 Comment / C++ / admin

Dijkstra’s algorithm is an algorithm for finding the shortest path in a graph. First drawing the graph:

import networkx as nx
import matplotlib.pyplot as plt
from networkx.drawing.nx_agraph import write_dot, graphviz_layout


def coloring_tree(G):

    visited_list=[]
    for i in G.nodes():
        visited_list.append( g.nodes[i]['visited'] )
    
    graph_color_map=[]
    for i  in visited_list:
        graph_color_map.append('red')
        
    pos =graphviz_layout(G, prog='dot')
    labels = nx.get_edge_attributes(G,'weight')
    nx.draw(G,pos,with_labels=True, arrows=True, node_color =  graph_color_map)
    nx.draw_networkx_edge_labels(G, pos,edge_labels=labels)
    plt.show()
    
    return

g=nx.DiGraph()
g.add_node('a',visited=False)
g.add_node('b',visited=False)
g.add_node('c',visited=False)
g.add_node('d',visited=False)
g.add_node('e',visited=False)
g.add_node('f',visited=False)
g.add_node('g',visited=False)
g.add_node('h',visited=False)
g.add_node('i',visited=False)


g.add_edge('a','b',weight=4)
g.add_edge('a','h',weight=8)


g.add_edge('b','a',weight=4)
g.add_edge('b','c',weight=8)
g.add_edge('b','h',weight=11)


g.add_edge('c','b',weight=8)
g.add_edge('c','d',weight=7)
g.add_edge('c','f',weight=4)
g.add_edge('c','i',weight=2)

g.add_edge('d','d',weight=7)
g.add_edge('d','e',weight=9)
g.add_edge('d','f',weight=14)

g.add_edge('e','d',weight=9)
g.add_edge('e','f',weight=10)

g.add_edge('f','c',weight=4)
g.add_edge('f','d',weight=14)
g.add_edge('f','e',weight=10)
g.add_edge('f','g',weight=2)

g.add_edge('g','f',weight=2)
g.add_edge('g','h',weight=1)
g.add_edge('g','i',weight=6)


g.add_edge('h','a',weight=8)
g.add_edge('g','b',weight=11)
g.add_edge('h','g',weight=1)
g.add_edge('h','i',weight=7)

g.add_edge('i','c',weight=2)
g.add_edge('i','g',weight=6)
g.add_edge('i','h',weight=7)

coloring_tree(g)

import networkx as nx

import matplotlib.pyplot as plt

from networkx.drawing.nx_agraph import write_dot, graphviz_layout

def coloring_tree(G):

visited_list=[]

for i in G.nodes():

visited_list.append( g.nodes[i]['visited'] )

graph_color_map=[]

for i in visited_list:

graph_color_map.append('red')

pos =graphviz_layout(G, prog='dot')

labels = nx.get_edge_attributes(G,'weight')

nx.draw(G,pos,with_labels=True, arrows=True, node_color = graph_color_map)

nx.draw_networkx_edge_labels(G, pos,edge_labels=labels)

plt.show()

return

g=nx.DiGraph()

g.add_node('a',visited=False)

g.add_node('b',visited=False)

g.add_node('c',visited=False)

g.add_node('d',visited=False)

g.add_node('e',visited=False)

g.add_node('f',visited=False)

g.add_node('g',visited=False)

g.add_node('h',visited=False)

g.add_node('i',visited=False)

g.add_edge('a','b',weight=4)

g.add_edge('a','h',weight=8)

g.add_edge('b','a',weight=4)

g.add_edge('b','c',weight=8)

g.add_edge('b','h',weight=11)

g.add_edge('c','b',weight=8)

g.add_edge('c','d',weight=7)

g.add_edge('c','f',weight=4)

g.add_edge('c','i',weight=2)

g.add_edge('d','d',weight=7)

g.add_edge('d','e',weight=9)

g.add_edge('d','f',weight=14)

g.add_edge('e','d',weight=9)

g.add_edge('e','f',weight=10)

g.add_edge('f','c',weight=4)

g.add_edge('f','d',weight=14)

g.add_edge('f','e',weight=10)

g.add_edge('f','g',weight=2)

g.add_edge('g','f',weight=2)

g.add_edge('g','h',weight=1)

g.add_edge('g','i',weight=6)

g.add_edge('h','a',weight=8)

g.add_edge('g','b',weight=11)

g.add_edge('h','g',weight=1)

g.add_edge('h','i',weight=7)

g.add_edge('i','c',weight=2)

g.add_edge('i','g',weight=6)

g.add_edge('i','h',weight=7)

coloring_tree(g)

#include <iostream>
#include <vector>
#include <string>
#include <list>
 
#include <limits> // for numeric_limits
 
#include <set>
#include <utility> // for pair
#include <algorithm>
#include <iterator>
 
 
typedef int vertex_t;
typedef double weight_t;
 
const weight_t max_weight = std::numeric_limits<double>::infinity();
 
struct neighbor {
    vertex_t target;
    weight_t weight;
    neighbor(vertex_t arg_target, weight_t arg_weight)
        : target(arg_target), weight(arg_weight) { }
};
 
typedef std::vector<std::vector<neighbor> > adjacency_list_t;
 
 
void DijkstraComputePaths(vertex_t source,
                          const adjacency_list_t &adjacency_list,
                          std::vector<weight_t> &min_distance,
                          std::vector<vertex_t> &previous)
{
    int n = adjacency_list.size();
    min_distance.clear();
    min_distance.resize(n, max_weight);
    min_distance[source] = 0;
    previous.clear();
    previous.resize(n, -1);
    std::set<std::pair<weight_t, vertex_t> > vertex_queue;
    vertex_queue.insert(std::make_pair(min_distance[source], source));
 
    while (!vertex_queue.empty()) 
    {
        weight_t dist = vertex_queue.begin()->first;
        vertex_t u = vertex_queue.begin()->second;
        vertex_queue.erase(vertex_queue.begin());
 
        // Visit each edge exiting u
	const std::vector<neighbor> &neighbors = adjacency_list[u];
        for (std::vector<neighbor>::const_iterator neighbor_iter = neighbors.begin();
             neighbor_iter != neighbors.end();
             neighbor_iter++)
        {
            vertex_t v = neighbor_iter->target;
            weight_t weight = neighbor_iter->weight;
            weight_t distance_through_u = dist + weight;
	    if (distance_through_u < min_distance[v]) {
	        vertex_queue.erase(std::make_pair(min_distance[v], v));
 
	        min_distance[v] = distance_through_u;
	        previous[v] = u;
	        vertex_queue.insert(std::make_pair(min_distance[v], v));
 
	    }
 
        }
    }
}
 
 
std::list<vertex_t> DijkstraGetShortestPathTo(
    vertex_t vertex, const std::vector<vertex_t> &previous)
{
    std::list<vertex_t> path;
    for ( ; vertex != -1; vertex = previous[vertex])
        path.push_front(vertex);
    return path;
}
 
 
int main()
{
    adjacency_list_t adjacency_list(9);
    adjacency_list[0].push_back(neighbor(1, 4));
    adjacency_list[0].push_back(neighbor(7, 8));

    adjacency_list[1].push_back(neighbor(0, 4));
    adjacency_list[1].push_back(neighbor(2, 8));
    adjacency_list[1].push_back(neighbor(7, 11));

    adjacency_list[2].push_back(neighbor(1, 8));
    adjacency_list[2].push_back(neighbor(3, 7));
    adjacency_list[2].push_back(neighbor(5, 4));
    adjacency_list[2].push_back(neighbor(8, 2));

    adjacency_list[3].push_back(neighbor(3, 7));
    adjacency_list[3].push_back(neighbor(4, 9));
    adjacency_list[3].push_back(neighbor(5, 14));


    adjacency_list[4].push_back(neighbor(3, 9));
    adjacency_list[4].push_back(neighbor(5, 10));
    
    adjacency_list[5].push_back(neighbor(2, 4));
    adjacency_list[5].push_back(neighbor(3, 14));
    adjacency_list[5].push_back(neighbor(4, 10));
    adjacency_list[5].push_back(neighbor(6, 2));

    adjacency_list[6].push_back(neighbor(5, 2));
    adjacency_list[6].push_back(neighbor(7, 1)); 
    adjacency_list[6].push_back(neighbor(8, 6)); 

    
    adjacency_list[7].push_back(neighbor(0, 9));
    adjacency_list[7].push_back(neighbor(1, 9));    
    adjacency_list[7].push_back(neighbor(6, 9));    
    adjacency_list[7].push_back(neighbor(8, 9));    
    

    adjacency_list[8].push_back(neighbor(2, 2));
    adjacency_list[8].push_back(neighbor(6, 6));
    adjacency_list[8].push_back(neighbor(7, 7));

    std::vector<weight_t> min_distance;
    std::vector<vertex_t> previous;
    DijkstraComputePaths(0, adjacency_list, min_distance, previous);
    std::cout << "Distance from 0 to 5: " << min_distance[5] << std::endl;
    std::list<vertex_t> path = DijkstraGetShortestPathTo(5, previous);
    std::cout << "Path : ";
    std::copy(path.begin(), path.end(), std::ostream_iterator<vertex_t>(std::cout, " "));
    std::cout << std::endl;
    return 0;
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

#include <iostream>

#include <vector>

#include <string>

#include <list>

#include <limits> // for numeric_limits

#include <set>

#include <utility> // for pair

#include <algorithm>

#include <iterator>

typedef int vertex_t;

typedef double weight_t;

const weight_t max_weight = std::numeric_limits<double>::infinity();

struct neighbor {

vertex_t target;

weight_t weight;

neighbor(vertex_t arg_target, weight_t arg_weight)

: target(arg_target), weight(arg_weight) { }

};

typedef std::vector<std::vector<neighbor> > adjacency_list_t;

void DijkstraComputePaths(vertex_t source,

const adjacency_list_t &adjacency_list,

std::vector<weight_t> &min_distance,

std::vector<vertex_t> &previous)

{

int n = adjacency_list.size();

min_distance.clear();

min_distance.resize(n, max_weight);

min_distance[source] = 0;

previous.clear();

previous.resize(n, -1);

std::set<std::pair<weight_t, vertex_t> > vertex_queue;

vertex_queue.insert(std::make_pair(min_distance[source], source));

while (!vertex_queue.empty())

{

weight_t dist = vertex_queue.begin()->first;

vertex_t u = vertex_queue.begin()->second;

vertex_queue.erase(vertex_queue.begin());

// Visit each edge exiting u

const std::vector<neighbor> &neighbors = adjacency_list[u];

for (std::vector<neighbor>::const_iterator neighbor_iter = neighbors.begin();

neighbor_iter != neighbors.end();

neighbor_iter++)

{

vertex_t v = neighbor_iter->target;

weight_t weight = neighbor_iter->weight;

weight_t distance_through_u = dist + weight;

if (distance_through_u < min_distance[v]) {

vertex_queue.erase(std::make_pair(min_distance[v], v));

min_distance[v] = distance_through_u;

previous[v] = u;

vertex_queue.insert(std::make_pair(min_distance[v], v));

}

std::list<vertex_t> DijkstraGetShortestPathTo(

vertex_t vertex, const std::vector<vertex_t> &previous)

{

std::list<vertex_t> path;

for ( ; vertex != -1; vertex = previous[vertex])

path.push_front(vertex);

return path;

}

int main()

{

adjacency_list_t adjacency_list(9);

adjacency_list[0].push_back(neighbor(1, 4));

adjacency_list[0].push_back(neighbor(7, 8));

adjacency_list[1].push_back(neighbor(0, 4));

adjacency_list[1].push_back(neighbor(2, 8));

adjacency_list[1].push_back(neighbor(7, 11));

adjacency_list[2].push_back(neighbor(1, 8));

adjacency_list[2].push_back(neighbor(3, 7));

adjacency_list[2].push_back(neighbor(5, 4));

adjacency_list[2].push_back(neighbor(8, 2));

adjacency_list[3].push_back(neighbor(3, 7));

adjacency_list[3].push_back(neighbor(4, 9));

adjacency_list[3].push_back(neighbor(5, 14));

adjacency_list[4].push_back(neighbor(3, 9));

adjacency_list[4].push_back(neighbor(5, 10));

adjacency_list[5].push_back(neighbor(2, 4));

adjacency_list[5].push_back(neighbor(3, 14));

adjacency_list[5].push_back(neighbor(4, 10));

adjacency_list[5].push_back(neighbor(6, 2));

adjacency_list[6].push_back(neighbor(5, 2));

adjacency_list[6].push_back(neighbor(7, 1));

adjacency_list[6].push_back(neighbor(8, 6));

adjacency_list[7].push_back(neighbor(0, 9));

adjacency_list[7].push_back(neighbor(1, 9));

adjacency_list[7].push_back(neighbor(6, 9));

adjacency_list[7].push_back(neighbor(8, 9));

adjacency_list[8].push_back(neighbor(2, 2));

adjacency_list[8].push_back(neighbor(6, 6));

adjacency_list[8].push_back(neighbor(7, 7));

std::vector<weight_t> min_distance;

std::vector<vertex_t> previous;

DijkstraComputePaths(0, adjacency_list, min_distance, previous);

std::cout << "Distance from 0 to 5: " << min_distance[5] << std::endl;

std::list<vertex_t> path = DijkstraGetShortestPathTo(5, previous);

std::cout << "Path : ";

std::copy(path.begin(), path.end(), std::ostream_iterator<vertex_t>(std::cout, " "));

std::cout << std::endl;

return 0;

}

Distance from 0 to 5: 16
Path : 0 1 2 5

1 2	Distance from 0 to 5: 16 Path : 0 1 2 5

ref [1]

Dijkstra’s algorithm with C++ Read More »

Standard Exception Handler in C++ and Custom Exception Handler with Examples

Leave a Comment / C++, Tutorials / admin

Exception handling

try
{
    some code
}
catch (Exception e)
{
    throw e;
}
catch (...)//... will catch any exception
{
    throw;
}

try

{

some code

}

catch (Exception e)

{

throw e;

}

catch (...)//... will catch any exception

{

throw;

}

C++ Standard Exceptions

1)std::bad_alloc
2)std::bad_cast
3)std::bad_exception
4)std::bad_typeid
5)std::logic_error
	5-1)std::domain_error: exception thrown when a mathematically invalid domain is used.
	5-2)std::invalid_argument
	5-3)std::length_error
	5-4)std::out_of_range
6)std::runtime_error: An exception that theoretically cannot be detected by reading the code.
	6-1)std::overflow_error :
    The only standard library components that
    throw std::overflow_error are std::bitset::to_ulong and std::bitset::to_ullong.
	6-2)std::underflow_error
	6-3)std::range_error

1)std::bad_alloc

2)std::bad_cast

3)std::bad_exception

4)std::bad_typeid

5)std::logic_error

5-1)std::domain_error: exception thrown when a mathematically invalid domain is used.

5-2)std::invalid_argument

5-3)std::length_error

5-4)std::out_of_range

6)std::runtime_error: An exception that theoretically cannot be detected by reading the code.

6-1)std::overflow_error :

The only standard library components that

throw std::overflow_error are std::bitset::to_ulong and std::bitset::to_ullong.

6-2)std::underflow_error

6-3)std::range_error

List of Example

struct Foo { virtual ~Foo() {} };
struct Bar { virtual ~Bar() {} };


void bad_allocExample()
{
    try
    {
        while (true)
        {
            new int[100000000ul];
        }
    } catch (const std::bad_alloc& e)
    {
        std::cout << "Allocation failed: " << e.what() << '\n';
    }
}


void bad_castExample()
{
    Bar b;
    try
    {
        Foo& f = dynamic_cast<Foo&>(b);
    } catch(const std::bad_cast& e)
    {
        std::cout << e.what() << '\n';
    }
}


void bad_typeidExample()
{
    Foo* p = nullptr;
    try
    {
        std::cout << typeid(*p).name() << '\n';
    } catch(const std::bad_typeid& e) {
        std::cout << e.what() << '\n';
    }
}


void logical_ErrorExample()
{
    int amount, available;
    amount=10;
    available=9;
    if(amount>available)
    {
        throw std::logic_error("Error");
    }
    catch ( std::exception &e )
    {
      std::cerr << "Caught: " << e.what( ) << std::endl;
      std::cerr << "Type: " << typeid( e ).name( ) << std::endl;
    };
}

void domain_errorExample()
{
    try
    {
        const double x = std::acos(2.0);
        std::cout << x << '\n';
    }
    catch (std::domain_error& e)
    {
        std::cout << e.what() << '\n';
    }
    catch (...)
    {
        std::cout << "Something unexpected happened" << '\n';
    }

}

void invalid_argumentExample()
{
    try
    {
    // binary wrongly represented by char X
        std::bitset<32> bitset(std::string("0101001X01010110000"));
    }
    catch (std::exception &err)
    {
        std::cerr<<"Caught "<<err.what()<<std::endl;
        std::cerr<<"Type "<<typeid(err).name()<<std::endl;
    }
}

void length_errorExample()
{
    try
    {
        // vector throws a length_error if resized above max_size
        std::vector<int> myvector;
        myvector.resize(myvector.max_size()+1);
    }
    catch (const std::length_error& le)
    {
        std::cerr << "Length error: " << le.what() << '\n';
    }
}

void out_of_rangeExample()
{
    std::vector<int> myvector(10);
    try
    {
        myvector.at(20)=100;      // vector::at throws an out-of-range
    }
    catch (const std::out_of_range& oor)
    {
        std::cerr << "Out of Range error: " << oor.what() << '\n';
    }
}

void overflow_errorExample()
{
    try
    {
        std::bitset<100> bitset;
        bitset[99] = 1;
        bitset[0] = 1;
        // to_ulong(), converts a bitset object to the integer that would generate the sequence of bits
        unsigned long Test = bitset.to_ulong();
    }
    catch(std::exception &err)
    {
        std::cerr<<"Caught "<<err.what()<<std::endl;
        std::cerr<<"Type "<<typeid(err).name()<<std::endl;
    }
}


void range_errorExample()
{
    try
    {
       throw std::range_error( "The range is in error!" );
    }
    catch (std::range_error &e)
    {
       std::cerr << "Caught: " << e.what( ) << std::endl;
       std::cerr << "Type: " << typeid( e ).name( ) << std::endl;
    }
}


//Defining your exceptions

struct CustomException : public std::exception
{
   const char * what () const throw ()
   {
      return "CustomException happened";
   }
};

void customExceptionExample()
{
    try
    {
        throw CustomException();
    } catch(CustomException& e)
    {
        std::cout << "CustomException caught" << std::endl;
        std::cout << e.what() << std::endl;
    } catch(std::exception& e)
    {
    //Other errors
    }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

struct Foo { virtual ~Foo() {} };

struct Bar { virtual ~Bar() {} };

void bad_allocExample()

{

try

{

while (true)

{

new int[100000000ul];

}

} catch (const std::bad_alloc& e)

{

std::cout << "Allocation failed: " << e.what() << '\n';

}

void bad_castExample()

{

Bar b;

try

{

Foo& f = dynamic_cast<Foo&>(b);

} catch(const std::bad_cast& e)

{

std::cout << e.what() << '\n';

}

void bad_typeidExample()

{

Foo* p = nullptr;

try

{

std::cout << typeid(*p).name() << '\n';

} catch(const std::bad_typeid& e) {

std::cout << e.what() << '\n';

}

void logical_ErrorExample()

{

int amount, available;

amount=10;

available=9;

if(amount>available)

{

throw std::logic_error("Error");

}

catch ( std::exception &e )

{

std::cerr << "Caught: " << e.what( ) << std::endl;

std::cerr << "Type: " << typeid( e ).name( ) << std::endl;

};

}

void domain_errorExample()

{

try

{

const double x = std::acos(2.0);

std::cout << x << '\n';

}

catch (std::domain_error& e)

{

std::cout << e.what() << '\n';

}

catch (...)

{

std::cout << "Something unexpected happened" << '\n';

}

void invalid_argumentExample()

{

try

{

// binary wrongly represented by char X

std::bitset<32> bitset(std::string("0101001X01010110000"));

}

catch (std::exception &err)

{

std::cerr<<"Caught "<<err.what()<<std::endl;

std::cerr<<"Type "<<typeid(err).name()<<std::endl;

}

void length_errorExample()

{

try

{

// vector throws a length_error if resized above max_size

std::vector<int> myvector;

myvector.resize(myvector.max_size()+1);

}

catch (const std::length_error& le)

{

std::cerr << "Length error: " << le.what() << '\n';

}

void out_of_rangeExample()

{

std::vector<int> myvector(10);

try

{

myvector.at(20)=100; // vector::at throws an out-of-range

}

catch (const std::out_of_range& oor)

{

std::cerr << "Out of Range error: " << oor.what() << '\n';

}

void overflow_errorExample()

{

try

{

std::bitset<100> bitset;

bitset[99] = 1;

bitset[0] = 1;

// to_ulong(), converts a bitset object to the integer that would generate the sequence of bits

unsigned long Test = bitset.to_ulong();

}

catch(std::exception &err)

{

std::cerr<<"Caught "<<err.what()<<std::endl;

std::cerr<<"Type "<<typeid(err).name()<<std::endl;

}

void range_errorExample()

{

try

{

throw std::range_error( "The range is in error!" );

}

catch (std::range_error &e)

{

std::cerr << "Caught: " << e.what( ) << std::endl;

std::cerr << "Type: " << typeid( e ).name( ) << std::endl;

}

//Defining your exceptions

struct CustomException : public std::exception

{

const char * what () const throw ()

{

return "CustomException happened";

}

};

void customExceptionExample()

{

try

{

throw CustomException();

} catch(CustomException& e)

{

std::cout << "CustomException caught" << std::endl;

std::cout << e.what() << std::endl;

} catch(std::exception& e)

{

//Other errors

}

Standard Exception Handler in C++ and Custom Exception Handler with Examples Read More »

Finding Memory leaking, Stack and Heap overflow

Leave a Comment / C++, Tutorials / admin

When you access an array index, C and C++ don’t do bound checking. Segmentation faults only happen when you try to read or write to a page that was not allocated (or try to do something on a page which isn’t permitted, e.g. trying to write to a read-only page), but since pages are usually

Finding Memory leaking, Stack and Heap overflow Read More »

Extended Kalman Filter Explained with Python Code

12 Comments / Machine Learning, Python, Robotic, Tutorials / admin

In the following code, I have implemented an Extended Kalman Filter for modeling the movement of a car with constant turn rate and velocity. The code is mainly based on this work (I did some bug fixing and some adaptation such that the code runs similar to the Kalman filter that I have earlier implemented).

import matplotlib.dates as mdates
import numpy as np
np.set_printoptions(threshold=3)
np.set_printoptions(suppress=True)

from numpy import genfromtxt
import matplotlib.pyplot as plt
from scipy.stats import norm
from sympy import Symbol, symbols, Matrix, sin, cos
from sympy import init_printing
from sympy.utilities.codegen import codegen
init_printing(use_latex=True)

x0 = []
x1 = []
x2 = []

def prediction(X_hat_t_1,P_t_1,Q_t,drivingStraight):
    X_hat_t=X_hat_t_1

    if drivingStraight: # Driving straight
        X_hat_t[0] = X_hat_t_1[0] + X_hat_t_1[3]*dt * np.cos(X_hat_t_1[2])
        X_hat_t[1] = X_hat_t_1[1] + X_hat_t_1[3]*dt * np.sin(X_hat_t_1[2])
        X_hat_t[2] = X_hat_t_1[2]
        X_hat_t[3] = X_hat_t_1[3] + X_hat_t_1[5]*dt
        X_hat_t[4] = 0.0000001 # avoid numerical issues in Jacobians
        X_hat_t[5] = X_hat_t_1[5]
       
    else: # otherwise
        X_hat_t[0] = X_hat_t_1[0] + (X_hat_t_1[3]/X_hat_t_1[4]) * (np.sin(X_hat_t_1[4]*dt+X_hat_t_1[2]) - np.sin(X_hat_t_1[2]))
        X_hat_t[1] = X_hat_t_1[1] + (X_hat_t_1[3]/X_hat_t_1[4]) * (-np.cos(X_hat_t_1[4]*dt+X_hat_t_1[2])+ np.cos(X_hat_t_1[2]))
        X_hat_t[2] = (X_hat_t_1[2] + X_hat_t_1[4]*dt + np.pi) % (2.0*np.pi) - np.pi
        X_hat_t[3] = X_hat_t_1[3] + X_hat_t_1[5]*dt
        X_hat_t[4] = X_hat_t_1[4] # Constant Turn Rate
        X_hat_t[5] = X_hat_t_1[5] # Constant Acceleration
        
    
    # Calculate the Jacobian of the Dynamic Matrix A
    # see "Calculate the Jacobian of the Dynamic Matrix with respect to the state vector"
    a13 = float((X_hat_t[3]/X_hat_t[4]) * (np.cos(X_hat_t[4]*dt+X_hat_t[2]) - np.cos(X_hat_t[2])))
    a14 = float((1.0/X_hat_t[4]) * (np.sin(X_hat_t[4]*dt+X_hat_t[2]) - np.sin(X_hat_t[2])))
    a15 = float((dt*X_hat_t[3]/X_hat_t[4])*np.cos(X_hat_t[4]*dt+X_hat_t[2]) - (X_hat_t[3]/X_hat_t[4]**2)*(np.sin(X_hat_t[4]*dt+X_hat_t[2]) - np.sin(X_hat_t[2])))
    a23 = float((X_hat_t[3]/X_hat_t[4]) * (np.sin(X_hat_t[4]*dt+X_hat_t[2]) - np.sin(X_hat_t[2])))
    a24 = float((1.0/X_hat_t[4]) * (-np.cos(X_hat_t[4]*dt+X_hat_t[2]) + np.cos(X_hat_t[2])))
    a25 = float((dt*X_hat_t[3]/X_hat_t[4])*np.sin(X_hat_t[4]*dt+X_hat_t[2]) - (X_hat_t[3]/X_hat_t[4]**2)*(-np.cos(X_hat_t[4]*dt+X_hat_t[2]) + np.cos(X_hat_t[2])))
    JA = np.matrix([[1.0, 0.0, a13, a14, a15, 0.0],
                    [0.0, 1.0, a23, a24, a25, 0.0],
                    [0.0, 0.0, 1.0, 0.0, dt, 0.0],
                    [0.0, 0.0, 0.0, 1.0, 0.0, dt],
                    [0.0, 0.0, 0.0, 0.0, 1.0, 0.0],
                    [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]])
    
    
    # Project the error covariance ahead
    P_t_1 = JA*P_t_1*JA.T + Q_t
    return X_hat_t,P_t_1

def update(X_hat_t,P_t,Z_t,R_t,GPSAvailable):
    hx = np.matrix([[float(X_hat_t[0])],
                    [float(X_hat_t[1])],
                    [float(X_hat_t[3])],
                    [float(X_hat_t[4])],
                    [float(X_hat_t[5])]])

    if GPSAvailable: # with 10Hz, every 5th step
        JH = np.matrix([[1.0, 0.0, 0.0, 0.0, 0.0, 0.0],
                        [0.0, 1.0, 0.0, 0.0, 0.0, 0.0],
                        [0.0, 0.0, 0.0, 1.0, 0.0, 0.0],
                        [0.0, 0.0, 0.0, 0.0, 1.0, 0.0],
                        [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]])
    else: # every other step
        JH = np.matrix([[0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
                        [0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
                        [0.0, 0.0, 0.0, 1.0, 0.0, 0.0],
                        [0.0, 0.0, 0.0, 0.0, 1.0, 0.0],
                        [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]])        
    
    S = JH*P_t*JH.T + R_t
    K = (P_t*JH.T) * np.linalg.inv(S)
    #print("K:\n",K)
    # Update the estimate via
    Z = Z_t.reshape(JH.shape[0],1)
    #print("Z:\n",Z)
    y = Z - (hx)                         # Innovation or Residual
    X_t = X_hat_t + (K*y)

    # Update the error covariance
    
    
    I = np.eye(X_hat_t.shape[0])
    P_t = (I - (K*JH))*P_t


    x0.append(float(X_t[0]))
    x1.append(float(X_t[1]))
    x2.append(float(X_t[2]))
    return X_t,P_t

numstates=6 # States

dt = 1.0/50.0 # Sample Rate of the Measurements is 50Hz
dtGPS=1.0/10.0 # Sample Rate of GPS is 10Hz

vs, psis, dpsis, dts, xs, ys, lats, lons, axs = symbols('v \psi \dot\psi T x y lat lon a')

gs = Matrix([[xs+(vs/dpsis)*(sin(psis+dpsis*dts)-sin(psis))],
             [ys+(vs/dpsis)*(-cos(psis+dpsis*dts)+cos(psis))],
             [psis+dpsis*dts],
             [axs*dts + vs],
             [dpsis],
             [axs]])
state = Matrix([xs,ys,psis,vs,dpsis,axs])


#Initial State cov
P_t = np.diag([1000.0, 1000.0, 1000.0, 1000.0, 1000.0, 1000.0])

#Measurment cov

varGPS = 5.0 # Standard Deviation of GPS Measurement
varspeed = 3.0 # Variance of the speed measurement
varyaw = 0.1 # Variance of the yawrate measurement
varacc = 1.0 # Variance of the longitudinal Acceleration
R_t = np.diag([varGPS**2, varGPS**2, varspeed**2, varyaw**2, varacc**2])


#Measurment Matrix
hs = Matrix([[xs],
             [ys],
             [vs],
             [dpsis],
             [axs]])

#Process cov
sGPS     = 0.5*8.8*dt**2  # assume 8.8m/s2 as maximum acceleration, forcing the vehicle
sCourse  = 0.1*dt # assume 0.1rad/s as maximum turn rate for the vehicle
sVelocity= 8.8*dt # assume 8.8m/s2 as maximum acceleration, forcing the vehicle
sYaw     = 1.0*dt # assume 1.0rad/s2 as the maximum turn rate acceleration for the vehicle
sAccel   = 0.5

Q_t = np.diag([sGPS**2, sGPS**2, sCourse**2, sVelocity**2, sYaw**2, sAccel**2])

path="data/"
datafile = '2014-03-26-000-Data.csv'

fullPath=path+datafile

def bytespdate2num(fmt, encoding='utf-8'):
    strconverter = mdates.strpdate2num(fmt)
    def bytesconverter(b):
        s = b.decode(encoding)
        return strconverter(s)
    return bytesconverter


date, time, millis, ax, ay, az, rollrate, pitchrate, yawrate, roll, pitch, yaw, speed, course, latitude, longitude, altitude, pdop, hdop, vdop, epe, fix, satellites_view, satellites_used, temp = np.loadtxt(fullPath, delimiter=',', unpack=True, 
                  converters={1:bytespdate2num('%H%M%S%f'),
                              0: bytespdate2num('%y%m%d')},
                  skiprows=1)


# A course of 0 means the Car is traveling north bound
# and 90 means it is traveling east bound.
# In the Calculation following, East is Zero and North is 90
# We need an offset.
course =(-course+90.0)

# ## Approx. Lat/Lon to Meters to check Location

# In[17]:

RadiusEarth = 6378388.0 # m
arc= 2.0*np.pi*(RadiusEarth+altitude)/360.0 # m/

dx = arc * np.cos(latitude*np.pi/180.0) * np.hstack((0.0, np.diff(longitude))) # in m
dy = arc * np.hstack((0.0, np.diff(latitude))) # in m

mx = np.cumsum(dx)
my = np.cumsum(dy)

ds = np.sqrt(dx**2+dy**2)

GPS=(ds!=0.0).astype('bool') # GPS Trigger for Kalman Filter

# ## Initial State
X_hat_t = np.matrix([[mx[0], my[0], course[0]/180.0*np.pi, speed[0]/3.6+0.001, yawrate[0]/180.0*np.pi, ax[0]]]).T

measurements = np.vstack((mx, my, speed/3.6, yawrate/180.0*np.pi, ax))
# Lenth of the measurement
m = measurements.shape[1]

for i in range(measurements.shape[1]):
#for i in range(3):
    
    if  np.abs(yawrate[i])<0.0001:
        drivingStraight=True
    else:
        drivingStraight=False
    
    
    X_hat_t,P_hat_t = prediction(X_hat_t,P_t,Q_t,drivingStraight)
    #print("Prediction:")
    #print("X_hat_t:\n",X_hat_t,"\nP_t:\n",P_hat_t)
   
    Z_t=measurements[:,i]
    if GPS[i]:
        GPSAvailable=True
    else:
        GPSAvailable=False
    
    X_t,P_t=update(X_hat_t,P_hat_t,Z_t,R_t,GPSAvailable)
    #print("Update:")
    #print("X_t:\n",X_t,"\nP_t:\n",P_t)
    X_hat_t=X_t
    P_hat_t=P_t
    
fig = plt.figure(figsize=(16,9))

# EKF State
plt.quiver(x0,x1,np.cos(x2), np.sin(x2), color='#94C600', units='xy', width=0.05, scale=0.5)
plt.plot(x0,x1, label='EKF Position', c='k', lw=5)

# Measurements
plt.scatter(mx[::5],my[::5], s=50, label='GPS Measurements', marker='+')

# Start/Goal
plt.scatter(x0[0],x1[0], s=60, label='Start', c='g')
plt.scatter(x0[-1],x1[-1], s=60, label='Goal', c='r')

plt.xlabel('X [m]')
plt.ylabel('Y [m]')
plt.title('Position')
plt.legend(loc='best')
plt.axis('equal')
plt.show()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

import matplotlib.dates as mdates

import numpy as np

np.set_printoptions(threshold=3)

np.set_printoptions(suppress=True)

from numpy import genfromtxt

import matplotlib.pyplot as plt

from scipy.stats import norm

from sympy import Symbol, symbols, Matrix, sin, cos

from sympy import init_printing

from sympy.utilities.codegen import codegen

init_printing(use_latex=True)

x0 = []

x1 = []

x2 = []

def prediction(X_hat_t_1,P_t_1,Q_t,drivingStraight):

X_hat_t=X_hat_t_1

if drivingStraight: # Driving straight

X_hat_t[0] = X_hat_t_1[0] + X_hat_t_1[3]*dt * np.cos(X_hat_t_1[2])

X_hat_t[1] = X_hat_t_1[1] + X_hat_t_1[3]*dt * np.sin(X_hat_t_1[2])

X_hat_t[2] = X_hat_t_1[2]

X_hat_t[3] = X_hat_t_1[3] + X_hat_t_1[5]*dt

X_hat_t[4] = 0.0000001 # avoid numerical issues in Jacobians

X_hat_t[5] = X_hat_t_1[5]

else: # otherwise

X_hat_t[0] = X_hat_t_1[0] + (X_hat_t_1[3]/X_hat_t_1[4]) * (np.sin(X_hat_t_1[4]*dt+X_hat_t_1[2]) - np.sin(X_hat_t_1[2]))

X_hat_t[1] = X_hat_t_1[1] + (X_hat_t_1[3]/X_hat_t_1[4]) * (-np.cos(X_hat_t_1[4]*dt+X_hat_t_1[2])+ np.cos(X_hat_t_1[2]))

X_hat_t[2] = (X_hat_t_1[2] + X_hat_t_1[4]*dt + np.pi) % (2.0*np.pi) - np.pi

X_hat_t[3] = X_hat_t_1[3] + X_hat_t_1[5]*dt

X_hat_t[4] = X_hat_t_1[4] # Constant Turn Rate

X_hat_t[5] = X_hat_t_1[5] # Constant Acceleration

# Calculate the Jacobian of the Dynamic Matrix A

# see "Calculate the Jacobian of the Dynamic Matrix with respect to the state vector"

a13 = float((X_hat_t[3]/X_hat_t[4]) * (np.cos(X_hat_t[4]*dt+X_hat_t[2]) - np.cos(X_hat_t[2])))

a14 = float((1.0/X_hat_t[4]) * (np.sin(X_hat_t[4]*dt+X_hat_t[2]) - np.sin(X_hat_t[2])))

a15 = float((dt*X_hat_t[3]/X_hat_t[4])*np.cos(X_hat_t[4]*dt+X_hat_t[2]) - (X_hat_t[3]/X_hat_t[4]**2)*(np.sin(X_hat_t[4]*dt+X_hat_t[2]) - np.sin(X_hat_t[2])))

a23 = float((X_hat_t[3]/X_hat_t[4]) * (np.sin(X_hat_t[4]*dt+X_hat_t[2]) - np.sin(X_hat_t[2])))

a24 = float((1.0/X_hat_t[4]) * (-np.cos(X_hat_t[4]*dt+X_hat_t[2]) + np.cos(X_hat_t[2])))

a25 = float((dt*X_hat_t[3]/X_hat_t[4])*np.sin(X_hat_t[4]*dt+X_hat_t[2]) - (X_hat_t[3]/X_hat_t[4]**2)*(-np.cos(X_hat_t[4]*dt+X_hat_t[2]) + np.cos(X_hat_t[2])))

JA = np.matrix([[1.0, 0.0, a13, a14, a15, 0.0],

[0.0, 1.0, a23, a24, a25, 0.0],

[0.0, 0.0, 1.0, 0.0, dt, 0.0],

[0.0, 0.0, 0.0, 1.0, 0.0, dt],

[0.0, 0.0, 0.0, 0.0, 1.0, 0.0],

[0.0, 0.0, 0.0, 0.0, 0.0, 1.0]])

# Project the error covariance ahead

P_t_1 = JA*P_t_1*JA.T + Q_t

return X_hat_t,P_t_1

def update(X_hat_t,P_t,Z_t,R_t,GPSAvailable):

hx = np.matrix([[float(X_hat_t[0])],

[float(X_hat_t[1])],

[float(X_hat_t[3])],

[float(X_hat_t[4])],

[float(X_hat_t[5])]])

if GPSAvailable: # with 10Hz, every 5th step

JH = np.matrix([[1.0, 0.0, 0.0, 0.0, 0.0, 0.0],

[0.0, 1.0, 0.0, 0.0, 0.0, 0.0],

[0.0, 0.0, 0.0, 1.0, 0.0, 0.0],

[0.0, 0.0, 0.0, 0.0, 1.0, 0.0],

[0.0, 0.0, 0.0, 0.0, 0.0, 1.0]])

else: # every other step

JH = np.matrix([[0.0, 0.0, 0.0, 0.0, 0.0, 0.0],

[0.0, 0.0, 0.0, 0.0, 0.0, 0.0],

[0.0, 0.0, 0.0, 1.0, 0.0, 0.0],

[0.0, 0.0, 0.0, 0.0, 1.0, 0.0],

[0.0, 0.0, 0.0, 0.0, 0.0, 1.0]])

S = JH*P_t*JH.T + R_t

K = (P_t*JH.T) * np.linalg.inv(S)

#print("K:\n",K)

# Update the estimate via

Z = Z_t.reshape(JH.shape[0],1)

#print("Z:\n",Z)

y = Z - (hx) # Innovation or Residual

X_t = X_hat_t + (K*y)

# Update the error covariance

I = np.eye(X_hat_t.shape[0])

P_t = (I - (K*JH))*P_t

x0.append(float(X_t[0]))

x1.append(float(X_t[1]))

x2.append(float(X_t[2]))

return X_t,P_t

numstates=6 # States

dt = 1.0/50.0 # Sample Rate of the Measurements is 50Hz

dtGPS=1.0/10.0 # Sample Rate of GPS is 10Hz

vs, psis, dpsis, dts, xs, ys, lats, lons, axs = symbols('v \psi \dot\psi T x y lat lon a')

gs = Matrix([[xs+(vs/dpsis)*(sin(psis+dpsis*dts)-sin(psis))],

[ys+(vs/dpsis)*(-cos(psis+dpsis*dts)+cos(psis))],

[psis+dpsis*dts],

[axs*dts + vs],

[dpsis],

[axs]])

state = Matrix([xs,ys,psis,vs,dpsis,axs])

#Initial State cov

P_t = np.diag([1000.0, 1000.0, 1000.0, 1000.0, 1000.0, 1000.0])

#Measurment cov

varGPS = 5.0 # Standard Deviation of GPS Measurement

varspeed = 3.0 # Variance of the speed measurement

varyaw = 0.1 # Variance of the yawrate measurement

varacc = 1.0 # Variance of the longitudinal Acceleration

R_t = np.diag([varGPS**2, varGPS**2, varspeed**2, varyaw**2, varacc**2])

#Measurment Matrix

hs = Matrix([[xs],

[ys],

[vs],

[dpsis],

[axs]])

#Process cov

sGPS = 0.5*8.8*dt**2 # assume 8.8m/s2 as maximum acceleration, forcing the vehicle

sCourse = 0.1*dt # assume 0.1rad/s as maximum turn rate for the vehicle

sVelocity= 8.8*dt # assume 8.8m/s2 as maximum acceleration, forcing the vehicle

sYaw = 1.0*dt # assume 1.0rad/s2 as the maximum turn rate acceleration for the vehicle

sAccel = 0.5

Q_t = np.diag([sGPS**2, sGPS**2, sCourse**2, sVelocity**2, sYaw**2, sAccel**2])

path="data/"

datafile = '2014-03-26-000-Data.csv'

fullPath=path+datafile

def bytespdate2num(fmt, encoding='utf-8'):

strconverter = mdates.strpdate2num(fmt)

def bytesconverter(b):

s = b.decode(encoding)

return strconverter(s)

return bytesconverter

date, time, millis, ax, ay, az, rollrate, pitchrate, yawrate, roll, pitch, yaw, speed, course, latitude, longitude, altitude, pdop, hdop, vdop, epe, fix, satellites_view, satellites_used, temp = np.loadtxt(fullPath, delimiter=',', unpack=True,

converters={1:bytespdate2num('%H%M%S%f'),

0: bytespdate2num('%y%m%d')},

skiprows=1)

# A course of 0 means the Car is traveling north bound

# and 90 means it is traveling east bound.

# In the Calculation following, East is Zero and North is 90

# We need an offset.

course =(-course+90.0)

# ## Approx. Lat/Lon to Meters to check Location

# In[17]:

RadiusEarth = 6378388.0 # m

arc= 2.0*np.pi*(RadiusEarth+altitude)/360.0 # m/

dx = arc * np.cos(latitude*np.pi/180.0) * np.hstack((0.0, np.diff(longitude))) # in m

dy = arc * np.hstack((0.0, np.diff(latitude))) # in m

mx = np.cumsum(dx)

my = np.cumsum(dy)

ds = np.sqrt(dx**2+dy**2)

GPS=(ds!=0.0).astype('bool') # GPS Trigger for Kalman Filter

# ## Initial State

X_hat_t = np.matrix([[mx[0], my[0], course[0]/180.0*np.pi, speed[0]/3.6+0.001, yawrate[0]/180.0*np.pi, ax[0]]]).T

measurements = np.vstack((mx, my, speed/3.6, yawrate/180.0*np.pi, ax))

# Lenth of the measurement

m = measurements.shape[1]

for i in range(measurements.shape[1]):

#for i in range(3):

if np.abs(yawrate[i])<0.0001:

drivingStraight=True

else:

drivingStraight=False

X_hat_t,P_hat_t = prediction(X_hat_t,P_t,Q_t,drivingStraight)

#print("Prediction:")

#print("X_hat_t:\n",X_hat_t,"\nP_t:\n",P_hat_t)

Z_t=measurements[:,i]

if GPS[i]:

GPSAvailable=True

else:

GPSAvailable=False

X_t,P_t=update(X_hat_t,P_hat_t,Z_t,R_t,GPSAvailable)

#print("Update:")

#print("X_t:\n",X_t,"\nP_t:\n",P_t)

X_hat_t=X_t

P_hat_t=P_t

fig = plt.figure(figsize=(16,9))

# EKF State

plt.quiver(x0,x1,np.cos(x2), np.sin(x2), color='#94C600', units='xy', width=0.05, scale=0.5)

plt.plot(x0,x1, label='EKF Position', c='k', lw=5)

# Measurements

plt.scatter(mx[::5],my[::5], s=50, label='GPS Measurements', marker='+')

# Start/Goal

plt.scatter(x0[0],x1[0], s=60, label='Start', c='g')

plt.scatter(x0[-1],x1[-1], s=60, label='Goal', c='r')

plt.xlabel('X [m]')

plt.ylabel('Y [m]')

plt.title('Position')

plt.legend(loc='best')

plt.axis('equal')

plt.show()

Extended Kalman Filter Explained with Python Code Read More »

Parcticle Filter Explained With Python Code From Scratch

8 Comments / Machine Learning, Python, Robotic, Tutorials / admin

In the following code I have implemented a localization algorithm based on particle filter. I have used conda to run my code, you can run the following for installation of dependencies:

conda create -n Filters python=3
conda activate Filters
conda install -c menpo opencv3
conda install numpy scipy matplotlib sympy

conda create -n Filters python=3

conda activate Filters

conda install -c menpo opencv3

conda install numpy scipy matplotlib sympy

and the code:

import numpy as np
import scipy as scipy
from numpy.random import uniform
import scipy.stats


np.set_printoptions(threshold=3)
np.set_printoptions(suppress=True)
import cv2


def drawLines(img, points, r, g, b):
    cv2.polylines(img, [np.int32(points)], isClosed=False, color=(r, g, b))

def drawCross(img, center, r, g, b):
    d = 5
    t = 2
    LINE_AA = cv2.LINE_AA if cv2.__version__[0] == '3' else cv2.CV_AA
    color = (r, g, b)
    ctrx = center[0,0]
    ctry = center[0,1]
    cv2.line(img, (ctrx - d, ctry - d), (ctrx + d, ctry + d), color, t, LINE_AA)
    cv2.line(img, (ctrx + d, ctry - d), (ctrx - d, ctry + d), color, t, LINE_AA)
    

def mouseCallback(event, x, y, flags,null):
    global center
    global trajectory
    global previous_x
    global previous_y
    global zs
    
    center=np.array([[x,y]])
    trajectory=np.vstack((trajectory,np.array([x,y])))
    #noise=sensorSigma * np.random.randn(1,2) + sensorMu
    
    if previous_x >0:
        heading=np.arctan2(np.array([y-previous_y]), np.array([previous_x-x ]))

        if heading>0:
            heading=-(heading-np.pi)
        else:
            heading=-(np.pi+heading)
            
        distance=np.linalg.norm(np.array([[previous_x,previous_y]])-np.array([[x,y]]) ,axis=1)
        
        std=np.array([2,4])
        u=np.array([heading,distance])
        predict(particles, u, std, dt=1.)
        zs = (np.linalg.norm(landmarks - center, axis=1) + (np.random.randn(NL) * sensor_std_err))
        update(particles, weights, z=zs, R=50, landmarks=landmarks)
        
        indexes = systematic_resample(weights)
        resample_from_index(particles, weights, indexes)

    previous_x=x
    previous_y=y
    


WIDTH=800
HEIGHT=600
WINDOW_NAME="Particle Filter"

#sensorMu=0
#sensorSigma=3

sensor_std_err=5


def create_uniform_particles(x_range, y_range, N):
    particles = np.empty((N, 2))
    particles[:, 0] = uniform(x_range[0], x_range[1], size=N)
    particles[:, 1] = uniform(y_range[0], y_range[1], size=N)
    return particles



def predict(particles, u, std, dt=1.):
    N = len(particles)
    dist = (u[1] * dt) + (np.random.randn(N) * std[1])
    particles[:, 0] += np.cos(u[0]) * dist
    particles[:, 1] += np.sin(u[0]) * dist
   
def update(particles, weights, z, R, landmarks):
    weights.fill(1.)
    for i, landmark in enumerate(landmarks):
       
        distance=np.power((particles[:,0] - landmark[0])**2 +(particles[:,1] - landmark[1])**2,0.5)
        weights *= scipy.stats.norm(distance, R).pdf(z[i])

 
    weights += 1.e-300 # avoid round-off to zero
    weights /= sum(weights)
    
def neff(weights):
    return 1. / np.sum(np.square(weights))

def systematic_resample(weights):
    N = len(weights)
    positions = (np.arange(N) + np.random.random()) / N

    indexes = np.zeros(N, 'i')
    cumulative_sum = np.cumsum(weights)
    i, j = 0, 0
    while i < N and j<N:
        if positions[i] < cumulative_sum[j]:
            indexes[i] = j
            i += 1
        else:
            j += 1
    return indexes
    
def estimate(particles, weights):
    pos = particles[:, 0:1]
    mean = np.average(pos, weights=weights, axis=0)
    var = np.average((pos - mean)**2, weights=weights, axis=0)
    return mean, var

def resample_from_index(particles, weights, indexes):
    particles[:] = particles[indexes]
    weights[:] = weights[indexes]
    weights /= np.sum(weights)

    
x_range=np.array([0,800])
y_range=np.array([0,600])

#Number of partciles
N=400

landmarks=np.array([ [144,73], [410,13], [336,175], [718,159], [178,484], [665,464]  ])
NL = len(landmarks)
particles=create_uniform_particles(x_range, y_range, N)


weights = np.array([1.0]*N)


# Create a black image, a window and bind the function to window
img = np.zeros((HEIGHT,WIDTH,3), np.uint8)
cv2.namedWindow(WINDOW_NAME)
cv2.setMouseCallback(WINDOW_NAME,mouseCallback)

center=np.array([[-10,-10]])

trajectory=np.zeros(shape=(0,2))
robot_pos=np.zeros(shape=(0,2))
previous_x=-1
previous_y=-1
DELAY_MSEC=50

while(1):

    cv2.imshow(WINDOW_NAME,img)
    img = np.zeros((HEIGHT,WIDTH,3), np.uint8)
    drawLines(img, trajectory,   0,   255, 0)
    drawCross(img, center, r=255, g=0, b=0)
    
    #landmarks
    for landmark in landmarks:
        cv2.circle(img,tuple(landmark),10,(255,0,0),-1)
    
    #draw_particles:
    for particle in particles:
        cv2.circle(img,tuple((int(particle[0]),int(particle[1]))),1,(255,255,255),-1)

    if cv2.waitKey(DELAY_MSEC) & 0xFF == 27:
        break
    
    cv2.circle(img,(10,10),10,(255,0,0),-1)
    cv2.circle(img,(10,30),3,(255,255,255),-1)
    cv2.putText(img,"Landmarks",(30,20),1,1.0,(255,0,0))
    cv2.putText(img,"Particles",(30,40),1,1.0,(255,255,255))
    cv2.putText(img,"Robot Trajectory(Ground truth)",(30,60),1,1.0,(0,255,0))

    drawLines(img, np.array([[10,55],[25,55]]), 0, 255, 0)
    


cv2.destroyAllWindows()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

import numpy as np

import scipy as scipy

from numpy.random import uniform

import scipy.stats

np.set_printoptions(threshold=3)

np.set_printoptions(suppress=True)

import cv2

def drawLines(img, points, r, g, b):

cv2.polylines(img, [np.int32(points)], isClosed=False, color=(r, g, b))

def drawCross(img, center, r, g, b):

d = 5

t = 2

LINE_AA = cv2.LINE_AA if cv2.__version__[0] == '3' else cv2.CV_AA

color = (r, g, b)

ctrx = center[0,0]

ctry = center[0,1]

cv2.line(img, (ctrx - d, ctry - d), (ctrx + d, ctry + d), color, t, LINE_AA)

cv2.line(img, (ctrx + d, ctry - d), (ctrx - d, ctry + d), color, t, LINE_AA)

def mouseCallback(event, x, y, flags,null):

global center

global trajectory

global previous_x

global previous_y

global zs

center=np.array([[x,y]])

trajectory=np.vstack((trajectory,np.array([x,y])))

#noise=sensorSigma * np.random.randn(1,2) + sensorMu

if previous_x >0:

heading=np.arctan2(np.array([y-previous_y]), np.array([previous_x-x ]))

if heading>0:

heading=-(heading-np.pi)

else:

heading=-(np.pi+heading)

distance=np.linalg.norm(np.array([[previous_x,previous_y]])-np.array([[x,y]]) ,axis=1)

std=np.array([2,4])

u=np.array([heading,distance])

predict(particles, u, std, dt=1.)

zs = (np.linalg.norm(landmarks - center, axis=1) + (np.random.randn(NL) * sensor_std_err))

update(particles, weights, z=zs, R=50, landmarks=landmarks)

indexes = systematic_resample(weights)

resample_from_index(particles, weights, indexes)

previous_x=x

previous_y=y

WIDTH=800

HEIGHT=600

WINDOW_NAME="Particle Filter"

#sensorMu=0

#sensorSigma=3

sensor_std_err=5

def create_uniform_particles(x_range, y_range, N):

particles = np.empty((N, 2))

particles[:, 0] = uniform(x_range[0], x_range[1], size=N)

particles[:, 1] = uniform(y_range[0], y_range[1], size=N)

return particles

def predict(particles, u, std, dt=1.):

N = len(particles)

dist = (u[1] * dt) + (np.random.randn(N) * std[1])

particles[:, 0] += np.cos(u[0]) * dist

particles[:, 1] += np.sin(u[0]) * dist

def update(particles, weights, z, R, landmarks):

weights.fill(1.)

for i, landmark in enumerate(landmarks):

distance=np.power((particles[:,0] - landmark[0])**2 +(particles[:,1] - landmark[1])**2,0.5)

weights *= scipy.stats.norm(distance, R).pdf(z[i])

weights += 1.e-300 # avoid round-off to zero

weights /= sum(weights)

def neff(weights):

return 1. / np.sum(np.square(weights))

def systematic_resample(weights):

N = len(weights)

positions = (np.arange(N) + np.random.random()) / N

indexes = np.zeros(N, 'i')

cumulative_sum = np.cumsum(weights)

i, j = 0, 0

while i < N and j<N:

if positions[i] < cumulative_sum[j]:

indexes[i] = j

i += 1

else:

j += 1

return indexes

def estimate(particles, weights):

pos = particles[:, 0:1]

mean = np.average(pos, weights=weights, axis=0)

var = np.average((pos - mean)**2, weights=weights, axis=0)

return mean, var

def resample_from_index(particles, weights, indexes):

particles[:] = particles[indexes]

weights[:] = weights[indexes]

weights /= np.sum(weights)

x_range=np.array([0,800])

y_range=np.array([0,600])

#Number of partciles

N=400

landmarks=np.array([ [144,73], [410,13], [336,175], [718,159], [178,484], [665,464] ])

NL = len(landmarks)

particles=create_uniform_particles(x_range, y_range, N)

weights = np.array([1.0]*N)

# Create a black image, a window and bind the function to window

img = np.zeros((HEIGHT,WIDTH,3), np.uint8)

cv2.namedWindow(WINDOW_NAME)

cv2.setMouseCallback(WINDOW_NAME,mouseCallback)

center=np.array([[-10,-10]])

trajectory=np.zeros(shape=(0,2))

robot_pos=np.zeros(shape=(0,2))

previous_x=-1

previous_y=-1

DELAY_MSEC=50

while(1):

cv2.imshow(WINDOW_NAME,img)

img = np.zeros((HEIGHT,WIDTH,3), np.uint8)

drawLines(img, trajectory, 0, 255, 0)

drawCross(img, center, r=255, g=0, b=0)

#landmarks

for landmark in landmarks:

cv2.circle(img,tuple(landmark),10,(255,0,0),-1)

#draw_particles:

for particle in particles:

cv2.circle(img,tuple((int(particle[0]),int(particle[1]))),1,(255,255,255),-1)

if cv2.waitKey(DELAY_MSEC) & 0xFF == 27:

break

cv2.circle(img,(10,10),10,(255,0,0),-1)

cv2.circle(img,(10,30),3,(255,255,255),-1)

cv2.putText(img,"Landmarks",(30,20),1,1.0,(255,0,0))

cv2.putText(img,"Particles",(30,40),1,1.0,(255,255,255))

cv2.putText(img,"Robot Trajectory(Ground truth)",(30,60),1,1.0,(0,255,0))

drawLines(img, np.array([[10,55],[25,55]]), 0, 255, 0)

cv2.destroyAllWindows()

Parcticle Filter Explained With Python Code From Scratch Read More »

Kalman Filter Explained With Python Code From Scratch

23 Comments / C++, Machine Learning, Python, Robotic, Tutorials / admin

This snippet shows tracking mouse cursor with Python code from scratch and comparing the result with OpenCV. The CSV file that has been used are being created with below c++ code. A sample could be downloaded from here 1, 2, 3. Python Kalman Filter

import numpy as np
np.set_printoptions(threshold=3)
np.set_printoptions(suppress=True)
from numpy import genfromtxt


#Notation used coming from: https://www.bzarg.com/p/how-a-kalman-filter-works-in-pictures/


def prediction(X_hat_t_1,P_t_1,F_t,B_t,U_t,Q_t):
    X_hat_t=F_t.dot(X_hat_t_1)+(B_t.dot(U_t).reshape(B_t.shape[0],-1) )
    P_t=np.diag(np.diag(F_t.dot(P_t_1).dot(F_t.transpose())))+Q_t
    return X_hat_t,P_t
    

def update(X_hat_t,P_t,Z_t,R_t,H_t):
    
    K_prime=P_t.dot(H_t.transpose()).dot( np.linalg.inv ( H_t.dot(P_t).dot(H_t.transpose()) +R_t ) )  
    print("K:\n",K_prime)
    
    X_t=X_hat_t+K_prime.dot(Z_t-H_t.dot(X_hat_t))
    P_t=P_t-K_prime.dot(H_t).dot(P_t)
    
    return X_t,P_t


acceleration=0
delta_t=1/20#milisecond

groundTruth = genfromtxt('data/groundTruth.csv', delimiter=',',skip_header=1)

#Observations: position_X, position_Y
measurmens = genfromtxt('data/measurmens.csv', delimiter=',',skip_header=1)


#Checking our result with OpenCV
opencvKalmanOutput = genfromtxt('data/kalmanv.csv', delimiter=',',skip_header=1)

#Transition matrix
F_t=np.array([ [1 ,0,delta_t,0] , [0,1,0,delta_t] , [0,0,1,0] , [0,0,0,1] ])

#Initial State cov
P_t= np.identity(4)*0.2

#Process cov
Q_t= np.identity(4)

#Control matrix
B_t=np.array( [ [0] , [0], [0] , [0] ])

#Control vector
U_t=acceleration

#Measurment Matrix
H_t = np.array([ [1, 0, 0, 0], [ 0, 1, 0, 0]])

#Measurment cov
R_t= np.identity(2)*5

# Initial State
X_hat_t = np.array( [[0],[0],[0],[0]] )
print("X_hat_t",X_hat_t.shape)
print("P_t",P_t.shape)
print("F_t",F_t.shape)
print("B_t",B_t.shape)
print("Q_t",Q_t.shape)
print("R_t",R_t.shape)
print("H_t",H_t.shape)

for i in range(measurmens.shape[0]):
    X_hat_t,P_hat_t = prediction(X_hat_t,P_t,F_t,B_t,U_t,Q_t)
    print("Prediction:")
    print("X_hat_t:\n",X_hat_t,"\nP_t:\n",P_t)
    
    Z_t=measurmens[i].transpose()
    Z_t=Z_t.reshape(Z_t.shape[0],-1)
    
    print(Z_t.shape)
    
    X_t,P_t=update(X_hat_t,P_hat_t,Z_t,R_t,H_t)
    print("Update:")
    print("X_t:\n",X_t,"\nP_t:\n",P_t)
    X_hat_t=X_t
    P_hat_t=P_t
    
    print("=========================================")
    print("Opencv Kalman Output:")
    print("X_t:\n",opencvKalmanOutput[i])

import numpy as np

np.set_printoptions(threshold=3)

np.set_printoptions(suppress=True)

from numpy import genfromtxt

#Notation used coming from: https://www.bzarg.com/p/how-a-kalman-filter-works-in-pictures/

def prediction(X_hat_t_1,P_t_1,F_t,B_t,U_t,Q_t):

X_hat_t=F_t.dot(X_hat_t_1)+(B_t.dot(U_t).reshape(B_t.shape[0],-1) )

P_t=np.diag(np.diag(F_t.dot(P_t_1).dot(F_t.transpose())))+Q_t

return X_hat_t,P_t

def update(X_hat_t,P_t,Z_t,R_t,H_t):

K_prime=P_t.dot(H_t.transpose()).dot( np.linalg.inv ( H_t.dot(P_t).dot(H_t.transpose()) +R_t ) )

print("K:\n",K_prime)

X_t=X_hat_t+K_prime.dot(Z_t-H_t.dot(X_hat_t))

P_t=P_t-K_prime.dot(H_t).dot(P_t)

return X_t,P_t

acceleration=0

delta_t=1/20#milisecond

groundTruth = genfromtxt('data/groundTruth.csv', delimiter=',',skip_header=1)

#Observations: position_X, position_Y

measurmens = genfromtxt('data/measurmens.csv', delimiter=',',skip_header=1)

#Checking our result with OpenCV

opencvKalmanOutput = genfromtxt('data/kalmanv.csv', delimiter=',',skip_header=1)

#Transition matrix

F_t=np.array([ [1 ,0,delta_t,0] , [0,1,0,delta_t] , [0,0,1,0] , [0,0,0,1] ])

#Initial State cov

P_t= np.identity(4)*0.2

#Process cov

Q_t= np.identity(4)

#Control matrix

B_t=np.array( [ [0] , [0], [0] , [0] ])

#Control vector

U_t=acceleration

#Measurment Matrix

H_t = np.array([ [1, 0, 0, 0], [ 0, 1, 0, 0]])

#Measurment cov

R_t= np.identity(2)*5

# Initial State

X_hat_t = np.array( [[0],[0],[0],[0]] )

print("X_hat_t",X_hat_t.shape)

print("P_t",P_t.shape)

print("F_t",F_t.shape)

print("B_t",B_t.shape)

print("Q_t",Q_t.shape)

print("R_t",R_t.shape)

print("H_t",H_t.shape)

for i in range(measurmens.shape[0]):

X_hat_t,P_hat_t = prediction(X_hat_t,P_t,F_t,B_t,U_t,Q_t)

print("Prediction:")

print("X_hat_t:\n",X_hat_t,"\nP_t:\n",P_t)

Z_t=measurmens[i].transpose()

Z_t=Z_t.reshape(Z_t.shape[0],-1)

print(Z_t.shape)

X_t,P_t=update(X_hat_t,P_hat_t,Z_t,R_t,H_t)

print("Update:")

print("X_t:\n",X_t,"\nP_t:\n",P_t)

X_hat_t=X_t

P_hat_t=P_t

print("=========================================")

print("Opencv Kalman Output:")

print("X_t:\n",opencvKalmanOutput[i])

C++ and OpenCV Kalman Filter Rapidcsv has been downloaded from

Kalman Filter Explained With Python Code From Scratch Read More »

Eigen unaryExpr (Function Pointer, Lambda Expression) Example

Leave a Comment / C++, Linear Algebra, Tutorials / admin

double ramp(double x)
{
    if (x > 0)
        return x;
    else
        return 0;
}

void unaryExprExample()
{
    //unaryExpr is a const function so it will not give you the possibility to modify values in place
    Eigen::ArrayXd x = Eigen::ArrayXd::Random(5);
    std::cout<<x <<std::endl;
    x = x.unaryExpr([](double elem) // changed type of parameter
    {
        return elem < 0.0 ? 0.0 : 1.0; // return instead of assignment
    });
    std::cout<<x <<std::endl;
    std::cout << x.unaryExpr(std::ptr_fun(ramp)) << std::endl;

}

double ramp(double x)

{

if (x > 0)

return x;

else

return 0;

}

void unaryExprExample()

{

//unaryExpr is a const function so it will not give you the possibility to modify values in place

Eigen::ArrayXd x = Eigen::ArrayXd::Random(5);

std::cout<<x <<std::endl;

x = x.unaryExpr([](double elem) // changed type of parameter

{

return elem < 0.0 ? 0.0 : 1.0; // return instead of assignment

});

std::cout<<x <<std::endl;

std::cout << x.unaryExpr(std::ptr_fun(ramp)) << std::endl;

}

Eigen unaryExpr (Function Pointer, Lambda Expression) Example Read More »

Cookie	Duration	Description
cookielawinfo-checbox-analytics	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookie is used to store the user consent for the cookies in the category "Analytics".
cookielawinfo-checbox-functional	11 months	The cookie is set by GDPR cookie consent to record the user consent for the cookies in the category "Functional".
cookielawinfo-checbox-others	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookie is used to store the user consent for the cookies in the category "Other.
cookielawinfo-checkbox-necessary	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookies is used to store the user consent for the cookies in the category "Necessary".
cookielawinfo-checkbox-performance	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookie is used to store the user consent for the cookies in the category "Performance".
viewed_cookie_policy	11 months	The cookie is set by the GDPR Cookie Consent plugin and is used to store whether or not user has consented to the use of cookies. It does not store any personal data.

admin

Bézier Curve Equation Explained

Kinematic Equation for Differential Drive Explained

Example of Test Driven Development with C++ and Google Test

Dijkstra’s algorithm with C++

Standard Exception Handler in C++ and Custom Exception Handler with Examples

Finding Memory leaking, Stack and Heap overflow

Extended Kalman Filter Explained with Python Code

Parcticle Filter Explained With Python Code From Scratch

Kalman Filter Explained With Python Code From Scratch

Eigen unaryExpr (Function Pointer, Lambda Expression) Example