Python3 wrapper for autoplait

autoplait_wrapper.py

#!/usr/bin/env python3
#-*- coding: utf-8 -*-

"""
Python3 wrapper for Autoplait
=============================

Original:
Yasuko Matsubara, Yasushi Sakurai, Christos Faloutsos,
"AutoPlait: Automatic Mining of Co-evolving Time Sequences",
ACM SIGMOD Conference, pp. 193-204, Snowbird, Utah,
June 22-27, 2014.
http://www.cs.kumamoto-u.ac.jp/~yasuko/software.html

Original C source code by Dr. Matsubara:
http://www.cs.kumamoto-u.ac.jp/~yasuko/SRC/autoplait.zip

Get started with this wrapper script:
$ cd /path/to/somewhere
$ wget http://www.cs.kumamoto-u.ac.jp/~yasuko/SRC/autoplait.zip
$ unzip autoplait.zip
$ cd autoplait
$ make
$ wget <URL of the raw file of this GIST>
$ python3 autoplait.py $AUTOPLAIT/_dat/21_01.amc.4d

or import `autoplait` (and `load_model`) function from your script.
"""

import os
import subprocess
import tempfile

import numpy as np


SCRIPTPATH = os.path.dirname(os.path.abspath(__file__))
AUTOPLAIT = os.path.join(SCRIPTPATH, "autoplait")


def load_model(path):
    model = {}
    with open(path, "r") as f:
        model["k"] = int(f.readline().strip().split("=")[1])
        model["d"] = int(f.readline().strip().split("=")[1])
        f.readline()
        model["pi"] = np.array(list(map(float, f.readline().strip().split())))
        f.readline()
        model["A"] = np.array([list(map(float, f.readline().strip().split()))
                               for _ in range(model["k"])])
        f.readline()
        model["mean"] = np.array([list(map(float, f.readline().strip().split()))
                                  for _ in range(model["k"])])
        f.readline()
        model["var"] = np.array([list(map(float, f.readline().strip().split()))
                                 for _ in range(model["k"])])
    return model


def autoplait(data):
    # data must be numpy.ndarray with shape (n_samples, n_features)

    d = data.shape[1]
    with tempfile.NamedTemporaryFile(mode="w+", delete=False) as infile, \
         tempfile.TemporaryDirectory() as outdirname:
        for row in data:
            print(*row, file=infile)
        infile.flush()
        with tempfile.NamedTemporaryFile(mode="w+") as f:
            print(infile.name, file=f, flush=True)
            command = [AUTOPLAIT, str(d), f.name, outdirname + "/"]
            subprocess.run(command, check=True)

        segpath = os.path.join(outdirname, "segment.")
        label = [ line.split()[0] for line in open(segpath + "labels", "r") ]
        segs = [
            [
                list(map(int, line.strip().split()))
                for line in open(segpath + l, "r")
            ]
            for l in label
        ]

        modelpath = os.path.join(outdirname, "model.")
        models = [ load_model(modelpath + l) for l in label ]
        
    return label, models, segs


def _normalize(data):
    mx, mn = np.max(data), np.min(data)
    return (data - mn) / (mx - mn)


if __name__ == "__main__":
    import sys
    import matplotlib.pyplot as plt
    
    filepath = sys.argv[1]
    data = np.array([ list(map(float, line.strip().split()))
                      for line in open(filepath, "r") ])
    l, m, s = autoplait(data)

    data = _normalize(data)
    c = ["b", "g", "r", "c", "m", "y", "k"]
    for i in range(data.shape[1]):
        plt.plot(data[:,i], "-")
        for st, ed in s[i]:
            plt.fill_between([st, ed], [1, 1], [0, 0],
                             facecolor=c[i%len(c)])
    plt.show()