14
\$\begingroup\$

This is a piece of code that implements an image-processing algorithm I came up with. I call it Dynamic Colour Binning. It's a fairly academic exercise that was more about providing a learning experience than about producing something useful, but here is what it does:

Input: A set of images with a limited colour space. I designed this with (simple) geographical maps in mind, where regions are represented by easily distinguishable colours. However, those colours need not be entirely pure (due to JPEG artefacts, antialiasing and whatnot.) The point of the algorithm is to group close colours together.

Output: For each image, a set of colours, with very similar colours grouped together, and their pixel count.

Since that probably sounds a little abstract, here's an example. One might take all of the maps from the Wikipedia page on the Territorial Evolution of the US and produce from that a plot like this (where pixel count has been translated into surface area):

enter image description here

The Algorithm: Simply put, the algorithm first finds all the colours in the image, and sorts them by pixel count. Then, starting from the most abundant colour, it takes every other colour, and if that colour is within some distance from the reference colour in Lab Colour Space it is deleted from the list of colours and its pixel count is added to that of the reference colour.

Architecture: I created three classes, from top to bottom:

A SetOfMaps, which is essentially just a list of Maps, together with some labels and some functionality for making the final output uniform and storing it in a pandas DataFrame.

A Map object, which is an OpenCV image with its list of colours and an additional method that acts on the image itself (performs a 'cleaning' of the image that makes the colour groups uniform.)

A ColorList, which is meant to always be a member of a Map but which I separated out because it does all the heavy lifting of the algorithm.

Questions: I am of course happy to receive any kind of feedback, but my main concerns are the following:

  1. This was my first foray into OOP. I gave the class structure quite some thought, but I felt like I was forcing an OOP-approach onto something not perfectly suited to it, namely a fairly abstract numerical calculation. (Or perhaps what was not suited was just me. In any case I'm not confident in my design.)

  2. I have no serious training in programming and software development. I'm a physicist and though I've done a lot of computational work, computer code to me has always been just a means to an end and never an end in itself. On top of that, until now my code has only rarely been seen or used by anyone else, so I've never been required to learn how to code neatly, readably and maintainably. I expect there is much that could be improved in that respect.

  3. This was also my first "serious" project in Python. I mostly have a background in C and, while I feel I know enough about Python to work with it, I am probably not always using it in the way it was intended, or in the most optimal way.

  4. I've attempted to make the code as easy to follow as possible, by making the code as self-explanatory as possible and by adding copious amounts of comments. In many cases, the latter compensates for a lack of the former, because some pieces of code (such as the actual numerical calculation being performed, or the regex search) don't seem to lend themselves to self-descriptive code. I wonder if this is just failure on my part, or if there is really no way to make such code more readable.

  5. There is obviously not much in my code in the way of error handling. It's probably trivial to crash. To what extent should that worry me? This obviously doesn't bother me in my day-to-day, where I am happy enough if my code runs on my input, without worrying about what would happen if someone for some reason fed it nonsense.

As a side note, some parts of the code have been optimized for speed, possibly at the expense of other code qualities. I'm interested in any and all comments, but I would most likely not implement any that slow down the program significantly.

Code: The code is separated into two files:

dynamiccolorbinning.py contains the class definitions and backbone of the calculations. analyze_map.py is a script that makes use of this class.

dynamiccolorbinning.py:

"""
dynamiccolorbinning.py: a module that provides functionality
associated with the Dynamic Color Binning algorithm.
Copyright: Marco Tompitak 2016
"""

import cv2
import numpy as np
import pandas as pd
import sys
import os

from ast import literal_eval

from colormath.color_objects import sRGBColor, LabColor
from colormath.color_conversions import convert_color
from colormath.color_diff import delta_e_cie2000

from matplotlib.colors import hex2color, rgb2hex


class ColorList:  
  """
  This class represents a list of colors with their pixel counts as found
  in an associated image. It can be constructed from an OpenCV color histogram
  (which should be a numpy array). The constructor will convert such a
  histogram into the right shape. A ColorList can also be constructed
  from a list using the .from_list() method.
  The actual data structure represented by a ColorList object is an N-by-4
  numpy array. In each row, the first three values are the R, G and B values
  that represent the color. The fourth value is the pixel count.
  """
  def __init__(self, hist=None):
    if ( hist != None ):
      self.colorlist = np.zeros([hist.size,4],'d',order='F')

      NI, NJ, NK = hist.shape

      # build columns for (i,j,k) tuples using repeat and tile
      self.colorlist[:,0] = np.repeat(range(NI),NJ*NK)
      self.colorlist[:,1] = np.tile(np.repeat(range(NJ),NK), NI)
      self.colorlist[:,2] = np.tile(range(NK), NI*NJ)
      self.colorlist[:,3] = hist.flatten()
      reduced_colorlist = self.colorlist[self.colorlist[:,3]>0.0]
      self.colorlist = reduced_colorlist[reduced_colorlist[:,3].argsort()[::-1]].astype(int).tolist()

  @classmethod
  def from_list(cls, src):
    """
    Class method to construct a ColorList object from a list.
    """
    clist = cls()
    clist.colorlist = src
    return clist

  def __getitem__(self, key):
    return self.colorlist[key]

  def dynamic_binning(self, radius):
    """
    This function applies the dynamic color binning algorithm to the
    ColorList. This algorithm sorts the colors by pixel count. Selecting
    the most prominent color, it searches the rest of the list for similar
    (i.e. within a distance <radius> in Lab color space) colors and adds
    the pixel counts of those colors to that of the prominent color, thus
    binning together similar colors. In this way it goes down the list
    until all colors present have been binned.
    The function returns a new ColorList object, as well as a dictionary
    that tells the user which colors have been binned together. This
    dictionary is of the form {major_color: [list, of, minor, colors]}.
    """
    colorlist = self.colorlist
    clustered_colorlist = []
    synonymous_colors = {}

    for color in colorlist:
      color_copy = color
      synonymous_colors[tuple(color[0:2])] = []

      # Store color as Lab-color
      color_rgb = sRGBColor(color[0], color[1], color[2], is_upscaled=True)
      color_lab = convert_color(color_rgb, LabColor)

      # Loop through all the colors that are less prominent than the current color
      for color_compare in colorlist[colorlist.index(color)+1:]:

        # Store color as Lab-color
        color_compare_rgb = sRGBColor(color_compare[0], color_compare[1], color_compare[2], is_upscaled=True)
        color_compare_lab = convert_color(color_compare_rgb, LabColor)

        # Calculate the distance in color space
        delta = delta_e_cie2000(color_lab, color_compare_lab)

        # If distance is smaller than threshold, label as similar
        if ( delta < radius ):

          # Add up pixel counts
          color_copy[3] += color_compare[3]

          # Remove color from the list we are looping over
          colorlist.remove(color_compare)

          synonymous_colors[tuple(color[0:2])].append(color_compare[0:2])

      # Add color with updated pixel count to new list
      clustered_colorlist.append(color_copy)

    clustered_colorlist.sort(key=lambda tup: tup[3], reverse=True)

    BinnedColorList = ColorList.from_list(clustered_colorlist)
    return BinnedColorList, synonymous_colors

  def colors(self):
    """
    Returns a numpy array similar to a ColorList, but without pixel counts.
    """
    colorlist_copy = self.colorlist
    for color in colorlist_copy:
      del color[3]
    return colorlist_copy

  def to_dataframe(self):
    """
    Converts the ColorList to a DataFrame (with a single row.)
    """
    colordict = {'('+str(x[0])+','+str(x[1])+','+str(x[2])+')':x[3] for x in self.colorlist}
    df = pd.DataFrame(colordict, index=[0])
    return df

  def palette(self, barwidth=500, barheight=100):
    """
    Generate a palette image with horizontal bands in the colors found in
    the ColorList. The RGB hex code is overlayed as well.
    """
    paletteimg = np.empty((0,3))
    for color in self.colorlist:
      pixels = np.empty((barwidth*barheight,3))
      pixels[...] = color[0:2]
      paletteimg = np.append(paletteimg,pixels)

    paletteimg = np.reshape(paletteimg,(paletteimg.size/(3*barwidth),barwidth,3))

    counter = 0
    for color in self.colorlist:
      y = barheight/2 + barheight*counter
      cv2.putText(paletteimg,str(rgb2hex([x/255.0 for x in color[0:2:-1]])), (10,y), cv2.FONT_HERSHEY_TRIPLEX, 1.5, uniquecolor)
      counter += 1

    return paletteimg


class Map:
  """
  This class represents a map object, which is essentially an image,
  with the following additional properties:
   - An OpenCV histogram of colors found in the image
   - A ColorList representation of this histogram
  When Dynamic Color Binning is run on the Map, additional properties
  are created:
   - A binned version of the ColorList
   - A dictionary mapping major colors to a list of synonymous minor colors
  """
  def __init__(self, image):
    self.img = image
    self.histogram = cv2.calcHist([self.img], [0,1,2], None, [256,256,256], [0,256,0,256,0,256])
    self.colorlist = ColorList(self.histogram)
    self.binned_colorlist = None
    self.synonymous_colors = None

  def run_dynamic_binning(self, radius=10):
    """
    This function simply runs DCB on the Map's ColorList
    """
    self.binned_colorlist, self.synonymous_colors = self.colorlist.dynamic_binning(radius)

  def dataframe(self):
    """
    This function returns the DataFrame representation of
    the Map's ColorList
    """
    return self.colorlist.to_dataframe()

  def binned_dataframe(self):
    """
    Like dataframe, but returns the binned version
    """
    if ( self.synonymous_colors == None ):
      self.run_dynamic_binning()
    return self.binned_colorlist.to_dataframe()

  def clean(self):
    """
    This function takes the Map's image and replaces all minor
    colors with their major synonym.
    """
    if ( self.synonymous_colors == None ):
      self.run_dynamic_binning()
    cleaned_img = self.img.copy()
    for major_color in self.binned_colorlist.colors():
      for minor_color in self.synonymous_colors[major_color]:
        cleaned_img[np.where((cleaned_img == minor_color).all(axis=2))] = major_color
    return cleaned_img


class SetOfMaps:
  """
  This class represents a set of Map objects. It is essentially
  nothing but a list of such objects, but each identified with a
  given label. This label is necessary for the main functionality
  of the class, which is extract color data from the Map objects
  into a pandas DataFrame. This is the final form of the data that
  we are looking for.
  """
  def __init__(self, list_of_maps=[], list_of_labels=[]):
    self.maps = list_of_maps
    self.labels = list_of_labels
    self.mapping = dict(zip(self.labels, self.maps))
    if ( len(self.maps) > 0 ):
      dfs = []
      # Loop through Maps. For each, get the binned dataframe and attach
      # the label. Then concatenate them all into one big dataframe.
      for label in self.labels:
        df = self.mapping[label].binned_dataframe().rename(index={0:label})
        dfs.append(df)
      self.dataframe = pd.concat(dfs)

  def add_map(self, new_map, new_label):
    """
    This function adds a new Map, with the supplied label, to the
    SetOfMaps object. Note that if one wants to work with the dataframe,
    one should call self.update_dataframe() after adding new Maps.
    """
    self.maps.append(new_map)
    self.labels.append(new_label)
    self.mapping[new_label] = new_map

  def update_dataframe(self):
    """
    This function recreates the dataframe, in the same way that it is done
    in the constructor. This is useful if more Maps are added to the Set.
    """
    dfs = []
    for label in self.labels:
      df = self.mapping[label].binned_dataframe().rename(index={0:label})
      dfs.append(df)
    self.dataframe = pd.concat(dfs)

  def bin_dataframe(self, radius):
    """
    This function looks at the Set's dataframe and checks whether there are
    columns that are closer together than _radius_ in colorspace. Such columns
    are then merged. 

    The algorithm is similar to the DCB algorithm itself, which is heavily commented
    in the ColorList class.
    """
    cols = list(self.dataframe)

    # Perform checking
    for col in cols:
      colbgr = literal_eval(col)
      color = sRGBColor(colbgr[0], colbgr[1], colbgr[2], is_upscaled=True)
      color_lab = convert_color(color, LabColor)

      for compcol in cols[cols.index(col)+1:]:
        compcolbgr = literal_eval(compcol)
        compcolor = sRGBColor(compcolbgr[0], compcolbgr[1], compcolbgr[2], is_upscaled=True)
        compcolor_lab = convert_color(compcolor, LabColor)
        delta = delta_e_cie2000(color_lab, compcolor_lab)
        if ( delta < radius ):
          self.dataframe[col].fillna(self.dataframe[compcol], inplace=True)
          del self.dataframe[compcol]
          cols.remove(compcol)

    # Clean up dataframe (sorting columns, setting NaN to 0)
    #self.dataframe.sort_index(inplace=True)
    self.dataframe.fillna(0, inplace=True)
    self.dataframe = self.dataframe.reindex_axis(sorted(self.dataframe.columns, key=lambda x: self.dataframe[x].sum(), reverse=True), axis=1)

  def filter_dataframe(self, thresh):
    """
    This function removes any columns from the dataframe whose largest
    pixel count is smaller than thresh.
    """
    self.dataframe = self.dataframe.loc[:, self.dataframe.max()>thresh]

  def hexheader_dataframe(self):
    """
    This function returns a copy of the object's dataframe, but with the
    BGR headers replaced by hex color codes.
    """
    dataframe_copy = self.dataframe.copy()
    cols = list(self.dataframe)
    hexcols = [str(rgb2hex([y/255.0 for y in list(literal_eval(x))[::-1]])) for x in cols]
    dataframe_copy.columns = hexcols
    return dataframe_copy

  def palette(self, uniquecolor, barwidth=500, barheight=100):
    """
    Generate a palette image with horizontal bands in the colors found in
    the set's dataframe. The RGB hex code is overlayed as well.
    """
    # Create empty image
    paletteimg = np.empty((0,3))

    # Get colors from dataframe column headers
    cols = list(self.dataframe)
    for col in cols:
      color = list(literal_eval(col))

      # Create a new bar filled with just the color and append to palette
      pixels = np.empty((barwidth*barheight,3))
      pixels[...] = color
      paletteimg = np.append(paletteimg,pixels)

    # Create a 2D image array from the flat list
    paletteimg = np.reshape(paletteimg,(paletteimg.size/(3*barwidth),barwidth,3))

    # Overlay hex color codes as text
    counter = 0
    for col in cols:
      color = list(literal_eval(col))
      y = barheight/2 + barheight*counter
      cv2.putText(paletteimg,str(rgb2hex([x/255.0 for x in color[::-1]])), (10,y), cv2.FONT_HERSHEY_TRIPLEX, 1.5, uniquecolor)
      counter += 1

    return paletteimg

analyze_map.py:

"""
analyze_map.py: Analyze a set of maps for dominant colors
and return their pixel counts as a function of time (assuming
different images represent different times).
Copyright: Marco Tompitak 2016
Usage: 
python analyze_map.py <colorspace binning radius> <folder>
          [unique color]
Example:
python analyze_map.py 10 Test '#00489C'
Description: 
This script runs the Dynamic Color Binning algorithm on a set
of images that represent maps. (It can be run on any set of images
but is designed for images with a small number of colors.) 
The algorithm requires a radius in L*a*b colorspace to do its
binning. Heuristically, colors that are closer together than
this radius in L*a*b colorspace are considered identical.
The script takes a directory as input and assumes that all files 
in this directory are images to be analyzed. The user should take
care to set this up properly.
Optionally one can provide a "unique color", where unique means
a color that is not close to any that is represented in the images.
This color will be used for the text overlays in the color palette.
"""

import dynamiccolorbinning as dcb

import cv2
import numpy as np
import pandas as pd
import sys
import os
import re

from ast import literal_eval

from colormath.color_objects import sRGBColor, LabColor
from colormath.color_conversions import convert_color
from colormath.color_diff import delta_e_cie2000

from matplotlib.colors import hex2color, rgb2hex


filterstrength = 0


# SCRIPT START
# Read in command line arguments.

# Bin size in Lab color space.
r = int(sys.argv[1])

# Folder to analyze.
folder = sys.argv[2]

# Color for text overlay on palette image.
if( len(sys.argv) > 3 ):
  uniquehex = sys.argv[3]
  uniquecolor = [int(255*i) for i in hex2color(uniquehex)]
else:
  uniquecolor = [0,0,255]


# Create new SetOfMaps object.
Maps = dcb.SetOfMaps()

# Loop through files in folder
for fn in os.listdir(folder):
  filename = './' + folder + '/' + fn
  if os.path.isfile(filename):
    # Find date in filename. This script assumes there is at least
    # a four-digit year in the name! Otherwise the whole filename
    # is used.
    date = re.search("([0-9]{4}-[0-9]{2}-)", filename)
    if ( date == None ):
      date = re.search("([0-9]{4})", filename).group(0) + "-01"
    else:
      date = date.group(0)[:-1]

    # Load the image with OpenCV.
    print "Loading image... " + filename
    img = cv2.imread(filename)

    # If desired, a denoising filter can be applied.
    if ( filterstrength > 0.0 ):
      img = cv2.fastNlMeansDenoisingColored(img,None,10,10,7,21)

    # Create a Map object from the image, apply the Dynamic Color
    # Binning algorithm and add to our SetOfMaps, labeled with the
    # date extracted from the filename.
    geo_map = dcb.Map(img)
    geo_map.run_dynamic_binning()
    Maps.add_map(geo_map, date)


# Populate the SetOfMaps' dataframe and clean up.
print "Analyzing set of images..."
Maps.update_dataframe()
Maps.bin_dataframe(r)
Maps.filter_dataframe(4000)


# Convert headers to hex color codes and store as csv.
print "Saving output to " + folder + ".csv"
Maps.hexheader_dataframe().to_csv(folder+'.csv')


# Generate and save palette of abundant colors.
print "Saving palette to " + folder + "_palette.png"
paletteimg = Maps.palette(uniquecolor)  
cv2.imwrite(folder+'_palette.png', paletteimg)

print "Done."
\$\endgroup\$
  • \$\begingroup\$ Welcome to Code Review, this looks amazing. Hope you get some good reviews! \$\endgroup\$ – Peilonrayz May 16 '16 at 14:11
  • \$\begingroup\$ @JoeWallis Thanks! Unfortunately no comments on the code yet, despite all the upvotes. Is there something I could do to improve the post? \$\endgroup\$ – Marco Tompitak May 18 '16 at 9:06
  • \$\begingroup\$ Give it a couple more days, a good review takes a while to write. I'd write you one but I've never used numpy or pandas, D: so my input wouldn't be very helpful. \$\endgroup\$ – Peilonrayz May 18 '16 at 12:50
  • \$\begingroup\$ @JoeWallis You're right, I'll be a little less impatient. ;) \$\endgroup\$ – Marco Tompitak May 18 '16 at 14:35
2
\$\begingroup\$

I don't know enough about pandas and numpy, but I decided to take a look at some of the code anyway.

def bin_dataframe(self, radius):
    """
    This function looks at the Set's dataframe and checks whether there are
    columns that are closer together than _radius_ in colorspace. Such columns
    are then merged. 

    The algorithm is similar to the DCB algorithm itself, which is heavily commented
    in the ColorList class.
    """
    cols = list(self.dataframe)

    # Perform checking
    for col in cols:
      colbgr = literal_eval(col)
      color = sRGBColor(colbgr[0], colbgr[1], colbgr[2], is_upscaled=True)
      color_lab = convert_color(color, LabColor)

      for compcol in cols[cols.index(col)+1:]:
        compcolbgr = literal_eval(compcol)
        compcolor = sRGBColor(compcolbgr[0], compcolbgr[1], compcolbgr[2], is_upscaled=True)
        compcolor_lab = convert_color(compcolor, LabColor)
        delta = delta_e_cie2000(color_lab, compcolor_lab)
        if ( delta < radius ):
          self.dataframe[col].fillna(self.dataframe[compcol], inplace=True)
          del self.dataframe[compcol]
          cols.remove(compcol)

    # Clean up dataframe (sorting columns, setting NaN to 0)
    #self.dataframe.sort_index(inplace=True)
    self.dataframe.fillna(0, inplace=True)
    self.dataframe = self.dataframe.reindex_axis(sorted(self.dataframe.columns, key=lambda x: self.dataframe[x].sum(), reverse=True), axis=1)

These are nested loops, which you can probably not do too much about due to your algorithm.

However, what I do notice is that you're a lot of duplicate work. You're creating slices of the list, which could be memory expensive. You're using cols.index which does a look-up all the time. You're also computing the color_lab all the time (len(cols)**2/2 times) which is expensive.

def bin_dataframe(self, radius):
    """
    This function looks at the Set's dataframe and checks whether there are
    columns that are closer together than _radius_ in colorspace. Such columns
    are then merged. 
    """
    def mklabcolor(color):
        parts = literal_eval(color)
        rgbcolor = sRGBColor(parts[0], parts[1], parts[2], is_upscaled=True)
        return convert_color(rgbcolor, LabColor)

    cols = [(color, mklabcolor(color) for color in self.dataframe]

    for idx, (col, color_lab) in enumerate(cols):
        for compidx, (compcolor, compcolor_lab) in enumerate(cols[idx+1:], idx+1):
            if delta_e_cie2000(color_lab, compcolor_lab)
                self.dataframe[col].fillna(self.dataframe[compcolor], inplace=True)
                del self.dataframe[compcolor]
                del cols[compidx]

    # Clean up dataframe (sorting columns, setting NaN to 0)
    #self.dataframe.sort_index(inplace=True)
    self.dataframe.fillna(0, inplace=True)
    self.dataframe = self.dataframe.reindex_axis(sorted(self.dataframe.columns, key=lambda x: self.dataframe[x].sum(), reverse=True), axis=1)

I left the last part (regarding the cleanup intact), as that's probably numpy specific, and I don't know enough numpy yet.

\$\endgroup\$
  • \$\begingroup\$ Thanks Sjoerd! I'll see if your suggestion speeds up the code when I have a chance. Do you have any opinion on any of the five specific points I mentioned? \$\endgroup\$ – Marco Tompitak May 28 '16 at 18:49

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.