In [2]:
import numpy as np
import matplotlib.pyplot as plt
In [29]:
def quantized(inp):
n = 10
n_batch, length, _ = inp.shape
out = np.zeros(shape=(n_batch, length, n))
for i_batch in range(n_batch):
for i_element in range(length):
out[i_batch,i_element,:], _ = np.histogram(inp[i_batch, i_element, 0], [-1,-.8,-.6,-.4,-.2,0.0,.2,.4,.6,.8,1])
return out - 0.5
In [30]:
def gen_single_appliance(length, power, on_duration, min_off_duration=20, fdiff=True):
if fdiff:
length += 1
appliance_power = np.zeros(shape=(length))
i = 0
while i < length:
if np.random.binomial(n=1, p=0.2):
end = min(i + on_duration, length)
appliance_power[i:end] = power
i += on_duration + min_off_duration
else:
i += 1
return np.diff(appliance_power) if fdiff else appliance_power
def gen_batches_of_single_appliance(length, n_batch, *args, **kwargs):
batches = np.zeros(shape=(n_batch, length, 1))
for i in range(n_batch):
batches[i, :, :] = gen_single_appliance(length, *args, **kwargs).reshape(length, 1)
return batches
def gen_data(length, n_batch, n_appliances=2, appliance_powers=[10,20], appliance_on_durations=[5,2]):
'''Generate a simple energy disaggregation data.
:parameters:
- length : int
Length of sequences to generate
- n_batch : int
Number of training sequences per batch
:returns:
- X : np.ndarray, shape=(n_batch, length, 1)
Input sequence
- y : np.ndarray, shape=(n_batch, length, 1)
Target sequence, appliance 1
'''
y = gen_batches_of_single_appliance(length, n_batch,
power=appliance_powers[0],
on_duration=appliance_on_durations[0])
X = y.copy()
for power, on_duration in zip(appliance_powers, appliance_on_durations)[1:]:
X += gen_batches_of_single_appliance(length, n_batch, power=power, on_duration=on_duration)
max_power = np.sum(appliance_powers)
return quantized(X / max_power), quantized(y / max_power)
In [31]:
# gen_batches_of_single_appliance(4, 20, 5, 3)
# gen_single_appliance(20, 5, 3).reshape(20, 1).shape
X, y = gen_data(100, 1)
ax = plt.gca()
ax.plot(X[0,:,0])
ax.plot(y[0,:,0])
plt.show()
In [32]:
X[0,:,0]
Out[32]:
In [ ]: