etm.py

"""
etm.py
******

:author: David Griffin <david.griffin@york.ac.uk>
:coauthor: Susan Stepney <susan.stepney@york.ac.uk>
:copyright: 2024
:license: MIT
:publication: Entropy Transformation Measures for Computational Capacity

The reference implementation for the RUM and CGM Entropy Transformation Measures

Copyright (c) 2024, David Griffin

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""

import random
import math
import scipy
from collections import Counter

def sample(epsilon, tau, input_seq):
    """Sampler function for CGM. Given the similarity and time dependency parameters,
    epislon and tau, calculate a list of indicies for similar inputs of input_seq"""
    sums = {}
    for x in range(tau, len(input_seq)):
        sums[x] = sum(input_seq[x-tau:x])
    similar = []
    for x in range(tau, len(input_seq)):
        valid = []
        for y in range(tau, len(input_seq)):
            if abs(sums[x]-sums[y]) <= epsilon:
                valid.append(y)
        similar.append(random.choice(valid))
    return similar

def uniform_discretization_generator(mn, mx, k):
    """Return a function that categorises a value between mn and mx into one
    of k bins""" 
    def f(x):
        return math.ceil(k * (x - mn) / (mx - mn))
    f.alphabet = list(range(1, k+1))
    return f

def exponential_discretization_generator(d, k):
    """Return a function that categorises a value between 0 and d into one of k
    exponentially growing bins"""
    def f(x):
        if x == 0: return 1
        return max(math.ceil(math.log(x / d, 2) + k), 1)
    f.alphabet = list(range(1, k+1))
    return f
    
def shannon(events_sequence, alphabet):
    """Calculate the Shannon Entropy of a sequence of events, given the alphabet of
    possible events"""
    events = Counter({a: 0 for a in alphabet})
    total = 0
    if isinstance(events_sequence[0], int):
        events.update(events_sequence)
        total = len(events_sequence)
    else:
        for seq in events_sequence:
            events.update(seq)
            total += len(seq)
    pdist = []
    for v in events.values():
        pdist.append(v / total)
    return scipy.stats.entropy(pdist)

def rum(inp, out, udi, udo):
    """Relative utilization measure: given a sequence of inputs and outputs to/from a
    reservoir (inp,out), and uniform discretization functions (udi,udo; generated by
    uniform_discretization_generator) for those sequences"""
    events_in = [udi(x) for x in inp]
    events_out = [udo(x) for x in out]
    in_shannon = shannon(events_in, udi.alphabet)
    out_shannon = shannon(events_out, udo.alphbet)
    return out_shannon / in_shannon
    
def cgm(inp, out, edi, edo, epsilon, tau):
    """Comparative genealization measure: given a sequence of inputs and outputs
    to/from a reservoir (inp,out), and expoential discretization functions (udi,udo;
    generated by exponential_discretization_generator) for those sequences, and the
    similarity and time history parameters epsilon and tau, calculates the comparative
    generalization measure"""
    similar = sample(epsilon, tau, inp)
    delta_inp = [abs(inp[x] - inp[similar[x-tau]]) for x in range(tau, len(inp))]
    delta_out = [abs(out[x] - out[similar[x-tau]]) for x in range(tau, len(inp))]
    events_in = [edi(x) for x in delta_inp]
    events_out = [edo(x) for x in delta_out]
    in_shannon = shannon(events_in, edi.alphabet)
    out_shannon = shannon(events_out, edi.alphabet)
    return out_shannon / in_shannon