53 KiB
53 KiB
In [ ]:
# Install the required libraries (run once)
!pip install numpy matplotlib scikit-fuzzy torch scikit-learnIn [ ]:
# Import all necessary libraries
import numpy as np
import matplotlib.pyplot as plt
import skfuzzy as fuzz
from skfuzzy import control as ctrl
import torch
import torch.nn as nn
import pandas as pd
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error
import warnings
warnings.filterwarnings('ignore')
# Fix the random seeds so results are reproducible
np.random.seed(42)
torch.manual_seed(42)In [ ]:
# ============================================================
# STEP 1.1 – Define the universes of discourse
# ============================================================
# Universes of the input variables
u_temp = np.arange(15, 35, 0.1) # temperature: 15 - 35°C
u_hum = np.arange(20, 90, 0.5) # humidity: 20 - 90%
u_occ = np.arange(0, 100, 1) # occupancy: 0 - 100 people
u_co2 = np.arange(400, 2000, 5) # CO2: 400 - 2000 ppm
u_time = np.arange(0, 24, 0.25) # time of day: 0 - 24h
# Universe of the output variable
u_power = np.arange(0, 100, 1) # power: 0 - 100%In [ ]:
# ============================================================
# STEP 1.2 – Membership functions
# ============================================================
# --- TEMPERATURE (°C) ---
mf_temp_low = fuzz.trapmf(u_temp, [15, 15, 18, 23]) # L-shape
mf_temp_medium = fuzz.trapmf(u_temp, [18, 23, 25, 30]) # trapezoid
mf_temp_high = fuzz.trapmf(u_temp, [25, 30, 35, 35]) # Γ-shape
# --- HUMIDITY (%) ---
mf_hum_dry = fuzz.trapmf(u_hum, [20, 20, 35, 55]) # L-shape
mf_hum_comfort = fuzz.trimf (u_hum, [35, 55, 75]) # triangle
mf_hum_humid = fuzz.trapmf(u_hum, [55, 75, 90, 90]) # Γ-shape
# --- OCCUPANCY (number of people) ---
mf_occ_low = fuzz.trapmf(u_occ, [0, 0, 15, 40]) # L-shape
mf_occ_medium = fuzz.trapmf(u_occ, [15, 40, 60, 85]) # trapezoid
mf_occ_high = fuzz.trapmf(u_occ, [60, 85, 100, 100]) # Γ-shape
# --- CO2 (ppm) ---
mf_co2_normal = fuzz.trapmf(u_co2, [400, 400, 800, 1000]) # L-shape
mf_co2_elev = fuzz.trimf (u_co2, [800, 1000, 1400]) # triangle
mf_co2_high = fuzz.trapmf(u_co2, [1000, 1400, 2000, 2000]) # Γ-shape
# --- TIME OF DAY (h) ---
mf_time_night = fuzz.trapmf(u_time, [0, 0, 4, 6]) # L-shape
mf_time_day = fuzz.trapmf(u_time, [4, 6, 16, 20]) # trapezoid
mf_time_even = fuzz.trapmf(u_time, [16, 20, 24, 24]) # Γ-shape
# --- FAN POWER (%) — output variable ---
mf_power_off = fuzz.trapmf(u_power, [0, 0, 10, 20]) # L-shape
mf_power_low = fuzz.trimf (u_power, [10, 20, 40]) # triangle
mf_power_medium = fuzz.trapmf(u_power, [20, 40, 60, 80]) # trapezoid
mf_power_high = fuzz.trimf (u_power, [60, 80, 95]) # triangle
mf_power_full = fuzz.trapmf(u_power, [80, 95, 100, 100]) # Γ-shape
print("Universe coverage check (minimum of the summed memberships; for this exercise it should be ≈ 1.0):")
print(f" Temperature — min: {(mf_temp_low+mf_temp_medium+mf_temp_high).min():.3f}")
print(f" Humidity — min: {(mf_hum_dry+mf_hum_comfort+mf_hum_humid).min():.3f}")
print(f" Occupancy — min: {(mf_occ_low+mf_occ_medium+mf_occ_high).min():.3f}")
print(f" CO₂ — min: {(mf_co2_normal+mf_co2_elev+mf_co2_high).min():.3f}")
print(f" Time of day — min: {(mf_time_night+mf_time_day+mf_time_even).min():.3f}")In [ ]:
# ============================================================
# STEP 1.3 – Visualization
# ============================================================
fig, axes = plt.subplots(3, 2, figsize=(14, 12))
fig.suptitle("Membership functions — Air-conditioning controller", fontsize=14, fontweight='bold')
# Temperature
ax = axes[0, 0]
ax.plot(u_temp, mf_temp_low, 'b-', linewidth=2, label='low')
ax.plot(u_temp, mf_temp_medium, 'g-', linewidth=2, label='medium')
ax.plot(u_temp, mf_temp_high, 'r-', linewidth=2, label='high')
ax.set_title('Temperature (°C)'); ax.set_ylabel('Membership μ')
ax.legend(); ax.grid(True, alpha=0.3); ax.set_ylim([-0.05, 1.1])
# Humidity
ax = axes[0, 1]
ax.plot(u_hum, mf_hum_dry, 'b-', linewidth=2, label='dry')
ax.plot(u_hum, mf_hum_comfort, 'g-', linewidth=2, label='comfortable')
ax.plot(u_hum, mf_hum_humid, 'r-', linewidth=2, label='humid')
ax.set_title('Humidity (%)'); ax.legend(); ax.grid(True, alpha=0.3); ax.set_ylim([-0.05, 1.1])
# Occupancy
ax = axes[1, 0]
ax.plot(u_occ, mf_occ_low, 'b-', linewidth=2, label='few')
ax.plot(u_occ, mf_occ_medium, 'g-', linewidth=2, label='medium')
ax.plot(u_occ, mf_occ_high, 'r-', linewidth=2, label='many')
ax.set_title('Number of people'); ax.set_ylabel('Membership μ')
ax.legend(); ax.grid(True, alpha=0.3); ax.set_ylim([-0.05, 1.1])
# CO2
ax = axes[1, 1]
ax.plot(u_co2, mf_co2_normal, 'b-', linewidth=2, label='normal')
ax.plot(u_co2, mf_co2_elev, 'g-', linewidth=2, label='elevated')
ax.plot(u_co2, mf_co2_high, 'r-', linewidth=2, label='high')
ax.set_title('CO₂ (ppm)'); ax.legend(); ax.grid(True, alpha=0.3); ax.set_ylim([-0.05, 1.1])
# Time of day
ax = axes[2, 0]
ax.plot(u_time, mf_time_night, 'b-', linewidth=2, label='night')
ax.plot(u_time, mf_time_day, 'g-', linewidth=2, label='day')
ax.plot(u_time, mf_time_even, 'r-', linewidth=2, label='evening')
ax.set_title('Time of day (h)'); ax.set_ylabel('Membership μ'); ax.set_xlabel('Hour')
ax.legend(); ax.grid(True, alpha=0.3); ax.set_ylim([-0.05, 1.1])
# Cooling power
ax = axes[2, 1]
ax.plot(u_power, mf_power_off, 'navy', linewidth=2, label='off')
ax.plot(u_power, mf_power_low, 'b-', linewidth=2, label='low')
ax.plot(u_power, mf_power_medium,'g-', linewidth=2, label='medium')
ax.plot(u_power, mf_power_high, 'orange', linewidth=2, label='high')
ax.plot(u_power, mf_power_full, 'r-', linewidth=2, label='full')
ax.set_title('Cooling power (%)'); ax.set_xlabel('Power [%]')
ax.legend(); ax.grid(True, alpha=0.3); ax.set_ylim([-0.05, 1.1])
plt.tight_layout()
plt.savefig('membership_functions.png', dpi=120, bbox_inches='tight')
plt.show()In [ ]:
# ============================================================
# STEP 1.4 – Universe coverage verification
# ============================================================
fig, axes = plt.subplots(2, 3, figsize=(15, 8))
fig.suptitle("Universe coverage check — sum of membership functions",
fontsize=13, fontweight='bold')
variables = [
("Temperature (°C)", u_temp, [mf_temp_low, mf_temp_medium, mf_temp_high],
['low','medium','high'], ['b','g','r']),
("Humidity (%)", u_hum, [mf_hum_dry, mf_hum_comfort, mf_hum_humid],
['dry','comfort','humid'], ['b','g','r']),
("Number of people", u_occ, [mf_occ_low, mf_occ_medium, mf_occ_high],
['few','medium','many'], ['b','g','r']),
("CO₂ (ppm)", u_co2, [mf_co2_normal, mf_co2_elev, mf_co2_high],
['normal','elevated','high'], ['b','g','r']),
("Time of day (h)", u_time, [mf_time_night, mf_time_day, mf_time_even],
['night','day','evening'], ['b','g','r']),
]
for i, (name, u, mfs, labels, colors) in enumerate(variables):
row, col = divmod(i, 3)
ax = axes[row, col]
# Draw the individual membership functions
for mf, lab, col in zip(mfs, labels, colors):
ax.plot(u, mf, color=col, linewidth=1.5, alpha=0.6, label=lab)
# Summed membership
total = sum(mfs)
ax.plot(u, total, 'k--', linewidth=2, label='sum')
ax.fill_between(u, total, alpha=0.1, color='gray')
ax.axhline(y=1.0, color='gray', linewidth=0.8, linestyle=':')
# Mark dead zones (sum < 0.5) — with L/Γ functions there should be none
dead = u[total < 0.5]
if len(dead) > 0:
ax.scatter(dead, np.zeros_like(dead), color='red', s=5, zorder=5,
label='dead zone!')
ax.set_title(name); ax.set_ylim([-0.05, 2.0])
ax.legend(fontsize=7); ax.grid(True, alpha=0.3)
# Remove the empty sixth subplot
axes[1, 2].axis('off')
plt.tight_layout()
plt.savefig('coverage_check.png', dpi=120, bbox_inches='tight')
plt.show()
# Text report
print("UNIVERSE COVERAGE")
print("=" * 55)
print(f" {'Variable':<22} {'min sum':>9} {'Status'}")
print("-" * 55)
for name, u, mfs, *_ in variables:
total = sum(mfs)
min_total = total.min()
n_dead = (total < 0.5).sum()
status = "- full coverage" if n_dead == 0 else f"✗ DEAD ZONE ({n_dead} pts)"
print(f" {name:<22} {min_total:>9.3f} {status}")In [ ]:
example = {
'temp': 27.0, # °C — warm, but not extreme
'hum': 70.0, # % — humid
'occ': 60.0, # — many people
'co2': 1100.0, # ppm — elevated/high CO₂
'time': 20.0 # h — evening
}In [ ]:
# ============================================================
# STEP 1.4b – Fuzzification of a single case
# ============================================================
# Compute membership degrees
mu_temp_low = fuzz.interp_membership(u_temp, mf_temp_low, example['temp'])
mu_temp_medium = fuzz.interp_membership(u_temp, mf_temp_medium, example['temp'])
mu_temp_high = fuzz.interp_membership(u_temp, mf_temp_high, example['temp'])
mu_hum_dry = fuzz.interp_membership(u_hum, mf_hum_dry, example['hum'])
mu_hum_comfort = fuzz.interp_membership(u_hum, mf_hum_comfort, example['hum'])
mu_hum_humid = fuzz.interp_membership(u_hum, mf_hum_humid, example['hum'])
mu_occ_low = fuzz.interp_membership(u_occ, mf_occ_low, example['occ'])
mu_occ_medium = fuzz.interp_membership(u_occ, mf_occ_medium, example['occ'])
mu_occ_high = fuzz.interp_membership(u_occ, mf_occ_high, example['occ'])
mu_co2_normal = fuzz.interp_membership(u_co2, mf_co2_normal, example['co2'])
mu_co2_elev = fuzz.interp_membership(u_co2, mf_co2_elev, example['co2'])
mu_co2_high = fuzz.interp_membership(u_co2, mf_co2_high, example['co2'])
mu_time_night = fuzz.interp_membership(u_time, mf_time_night, example['time'])
mu_time_day = fuzz.interp_membership(u_time, mf_time_day, example['time'])
mu_time_even = fuzz.interp_membership(u_time, mf_time_even, example['time'])
# Print the fuzzification results
print("=" * 55)
print("FUZZIFICATION — membership degrees")
print("=" * 55)
print(f"Temperature = {example['temp']}°C")
print(f" μ(low) = {mu_temp_low:.4f}")
print(f" μ(medium) = {mu_temp_medium:.4f}")
print(f" μ(high) = {mu_temp_high:.4f}")
print()
print(f"Humidity = {example['hum']}%")
print(f" μ(dry) = {mu_hum_dry:.4f}")
print(f" μ(comfort) = {mu_hum_comfort:.4f}")
print(f" μ(humid) = {mu_hum_humid:.4f}")
print()
print(f"People = {example['occ']}")
print(f" μ(few) = {mu_occ_low:.4f}")
print(f" μ(medium) = {mu_occ_medium:.4f}")
print(f" μ(many) = {mu_occ_high:.4f}")
print()
print(f"CO₂ = {example['co2']} ppm")
print(f" μ(normal) = {mu_co2_normal:.4f}")
print(f" μ(elevated) = {mu_co2_elev:.4f}")
print(f" μ(high) = {mu_co2_high:.4f}")
print()
print(f"Time of day = {example['time']}h")
print(f" μ(night) = {mu_time_night:.4f}")
print(f" μ(day) = {mu_time_day:.4f}")
print(f" μ(evening) = {mu_time_even:.4f}")
print("=" * 55)
In [ ]:
# ============================================================
# STEP 1.5 – Rule base (Mamdani) — table-based definition
# ============================================================
RULE_BASE = [
# temp hum occ co2 time → power
{'temp': 'high', 'hum': 'humid', 'occ': '-', 'co2': '-', 'time': '-', 'power': 'full'}, # R01
{'temp': 'high', 'hum': '-', 'occ': 'high', 'co2': '-', 'time': '-', 'power': 'full'}, # R02
{'temp': 'high', 'hum': '-', 'occ': '-', 'co2': '-', 'time': '-', 'power': 'high'}, # R03
{'temp': 'medium', 'hum': 'humid', 'occ': '-', 'co2': '-', 'time': '-', 'power': 'high'}, # R04
{'temp': 'medium', 'hum': '-', 'occ': 'high', 'co2': '-', 'time': '-', 'power': 'medium'}, # R05
{'temp': 'medium', 'hum': '-', 'occ': '-', 'co2': '-', 'time': '-', 'power': 'medium'}, # R06
{'temp': 'low', 'hum': '-', 'occ': 'low', 'co2': '-', 'time': '-', 'power': 'off'}, # R07
{'temp': 'low', 'hum': '-', 'occ': '-', 'co2': '-', 'time': '-', 'power': 'low'}, # R08
{'temp': '-', 'hum': '-', 'occ': '-', 'co2': 'high', 'time': '-', 'power': 'high'}, # R09
{'temp': '-', 'hum': '-', 'occ': 'high', 'co2': 'elev', 'time': '-', 'power': 'medium'}, # R10
{'temp': 'medium', 'hum': '-', 'occ': '-', 'co2': '-', 'time': 'night', 'power': 'low'}, # R11
{'temp': 'low', 'hum': '-', 'occ': '-', 'co2': '-', 'time': 'night', 'power': 'off'}, # R12
{'temp': 'high', 'hum': '-', 'occ': 'high', 'co2': '-', 'time': 'even', 'power': 'full'}, # R13
{'temp': '-', 'hum': '-', 'occ': '-', 'co2': 'high', 'time': 'even', 'power': 'high'}, # R14
{'temp': 'medium', 'hum': '-', 'occ': '-', 'co2': 'normal', 'time': 'day', 'power': 'medium'}, # R15
]
# ============================================================
# Dictionary of fuzzy sets — maps (variable, label) to data
# ============================================================
FUZZY_SETS = {
# TEMPERATURE
('temp', 'low'): (u_temp, mf_temp_low),
('temp', 'medium'): (u_temp, mf_temp_medium),
('temp', 'high'): (u_temp, mf_temp_high),
# HUMIDITY
('hum', 'dry'): (u_hum, mf_hum_dry),
('hum', 'comfort'): (u_hum, mf_hum_comfort),
('hum', 'humid'): (u_hum, mf_hum_humid),
# OCCUPANCY
('occ', 'low'): (u_occ, mf_occ_low),
('occ', 'medium'): (u_occ, mf_occ_medium),
('occ', 'high'): (u_occ, mf_occ_high),
# CO₂
('co2', 'normal'): (u_co2, mf_co2_normal),
('co2', 'elev'): (u_co2, mf_co2_elev),
('co2', 'high'): (u_co2, mf_co2_high),
# TIME OF DAY
('time', 'night'): (u_time, mf_time_night),
('time', 'day'): (u_time, mf_time_day),
('time', 'even'): (u_time, mf_time_even),
}
INPUT_VARIABLES = ['temp', 'hum', 'occ', 'co2', 'time']
# ============================================================
# Inference engine — generic, interprets any rule table
# ============================================================
def evaluate_rules(rule_base, values):
active = []
for rule in rule_base:
strengths = []
for variable in INPUT_VARIABLES:
label = rule[variable]
if label == '-':
continue # this variable does not take part in the rule
u, mf = FUZZY_SETS[(variable, label)]
mu = fuzz.interp_membership(u, mf, values[variable])
strengths.append(mu)
strength = min(strengths) if strengths else 0.0
active.append((strength, rule['power']))
return active
# --- Print the rule table ---
print(f"Rule base defined ({len(RULE_BASE)} rules).")
print()
print(f" {'Nr':<4} {'temp':<10} {'hum':<12} {'occ':<8} {'co2':<10} {'time':<10} │ {'power':<8}")
print(" " + "-" * 68)
for i, r in enumerate(RULE_BASE, 1):
print(f" R{i:02d} {r['temp']:<10} {r['hum']:<12} {r['occ']:<8} "
f"{r['co2']:<10} {r['time']:<10} │ {r['power']:<8}")In [ ]:
# ============================================================
# STEP 1.6 – Aggregation and defuzzification (centroid method)
# ============================================================
OUTPUT_SETS = {
'off': mf_power_off,
'low': mf_power_low,
'medium': mf_power_medium,
'high': mf_power_high,
'full': mf_power_full,
}
def defuzzify(active_rules):
# Aggregation (max) + defuzzification (centroid), Mamdani style.
# Always returns a tuple (numeric_result, aggregated_set).
aggregate = np.zeros_like(u_power, dtype=float)
for strength, label in active_rules:
if strength > 0:
# Clip the output membership function at the rule's firing strength
clipped = np.fmin(strength, OUTPUT_SETS[label])
# Aggregation: max over all rules
aggregate = np.fmax(aggregate, clipped)
# Centroid defuzzification
if np.sum(aggregate) == 0:
return 0.0, aggregate
result = fuzz.defuzz(u_power, aggregate, 'centroid')
return result, aggregate
def fls_controller(temp, hum, occ, co2, time, base=None):
# Complete FLS controller: fuzzification → rules → defuzzification.
#
# The `base` parameter allows passing a different rule table
# (e.g. with a fallback rule). Defaults to RULE_BASE.
if base is None:
base = RULE_BASE
values = {'temp': temp, 'hum': hum, 'occ': occ, 'co2': co2, 'time': time}
active = evaluate_rules(base, values)
result, aggregate = defuzzify(active)
return result, aggregate
# Test the controller on the example case
result_fls, aggregate_fls = fls_controller(
temp=example['temp'],
hum=example['hum'],
occ=example['occ'],
co2=example['co2'],
time=example['time']
)
print(f"Input: temp={example['temp']}°C, hum={example['hum']}%,",
f"people={example['occ']}, CO₂={example['co2']} ppm, time={example['time']}h")
print(f"FLS output: cooling power = {result_fls:.2f}%")
# Visualize the aggregate and the defuzzification result
fig, ax = plt.subplots(figsize=(8, 4))
ax.fill_between(u_power, aggregate_fls, alpha=0.4, color='steelblue', label='aggregated set')
ax.axvline(x=result_fls, color='red', linewidth=2, linestyle='--',
label=f'centroid = {result_fls:.1f}%')
ax.set_title("Defuzzification — FLS output passed to the actuator")
ax.set_xlabel("Fan power [%]")
ax.set_ylabel("Membership μ")
ax.legend(); ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()In [ ]:
# ============================================================
# STEP 2.1 – Generate training data from the FLS
# ============================================================
N_SAMPLES_FLS = 10000
rng_fls = np.random.default_rng(42)
data_X_fls = []
data_y_fls = []
print(f"Generating {N_SAMPLES_FLS} samples from the FLS...")
for i in range(N_SAMPLES_FLS):
t = rng_fls.uniform(15, 35)
w = rng_fls.uniform(20, 90)
o = rng_fls.uniform(0, 100)
c = rng_fls.uniform(400, 2000)
p = rng_fls.uniform(0, 24)
try:
power, _ = fls_controller(t, w, o, c, p)
except Exception:
continue
data_X_fls.append([t, w, o, c, p])
data_y_fls.append(power)
data_X_fls = np.array(data_X_fls)
data_y_fls = np.array(data_y_fls).reshape(-1, 1)
print(f"Generated {len(data_X_fls)} samples.")
print(f"FLS power range: {data_y_fls.min():.1f}% - {data_y_fls.max():.1f}%")In [ ]:
# ============================================================
# STEP 2.2 – Normalization and splitting
# ============================================================
scaler_X_fls = MinMaxScaler()
scaler_y_fls = MinMaxScaler()
X_fls_norm = scaler_X_fls.fit_transform(data_X_fls)
y_fls_norm = scaler_y_fls.fit_transform(data_y_fls)
X_tr_f, X_te_f, y_tr_f, y_te_f = train_test_split(
X_fls_norm, y_fls_norm, test_size=0.15, random_state=42)
X_tr_f, X_val_f, y_tr_f, y_val_f = train_test_split(
X_tr_f, y_tr_f, test_size=0.176, random_state=42)
print(f"Train: {len(X_tr_f)}, Validation: {len(X_val_f)}, Test: {len(X_te_f)}")
# Tensors
X_tr_ft = torch.FloatTensor(X_tr_f)
y_tr_ft = torch.FloatTensor(y_tr_f)
X_val_ft = torch.FloatTensor(X_val_f)
y_val_ft = torch.FloatTensor(y_val_f)
X_te_ft = torch.FloatTensor(X_te_f)
y_te_ft = torch.FloatTensor(y_te_f)
train_ds_fls = TensorDataset(X_tr_ft, y_tr_ft)
train_ldr_fls = DataLoader(train_ds_fls, batch_size=64, shuffle=True)In [ ]:
# ============================================================
# STEP 2.3 – Architecture and training of the ANN-imitator
# ============================================================
class FLSImitator(nn.Module):
# Simple neural network imitating the FLS controller.
# Input (5): [temp, hum, occ, co2, time]
# Output (1): fan power [0–1] (normalized)
def __init__(self):
super(FLSImitator, self).__init__()
self.net = nn.Sequential(
nn.Linear(5, 32),
nn.ReLU(),
nn.Linear(32, 16),
nn.ReLU(),
nn.Linear(16, 1),
nn.Sigmoid()
)
def forward(self, x):
return self.net(x)
imitator = FLSImitator()
criterion_fls = nn.MSELoss()
optimizer_fls = torch.optim.Adam(imitator.parameters(), lr=0.001)
N_EPOCHS_FLS = 80
history_tr_fls = []
history_val_fls = []
print(f"Training the ANN-imitator for {N_EPOCHS_FLS} epochs...")
print("-" * 50)
for epoch in range(1, N_EPOCHS_FLS + 1):
imitator.train()
loss_tr = 0.0
for Xb, yb in train_ldr_fls:
optimizer_fls.zero_grad()
pred = imitator(Xb)
loss = criterion_fls(pred, yb)
loss.backward()
optimizer_fls.step()
loss_tr += loss.item() * len(Xb)
loss_tr /= len(X_tr_f)
imitator.eval()
with torch.no_grad():
loss_val = criterion_fls(imitator(X_val_ft), y_val_ft).item()
history_tr_fls.append(loss_tr)
history_val_fls.append(loss_val)
if epoch % 20 == 0:
print(f" Epoch {epoch:3d}/{N_EPOCHS_FLS} | "
f"train: {loss_tr:.6f} | val: {loss_val:.6f}")
print("-" * 50)
print("Training finished.")
# Learning curve
fig, ax = plt.subplots(figsize=(9, 4))
ax.plot(history_tr_fls, label='Train', color='steelblue', linewidth=2)
ax.plot(history_val_fls, label='Validation', color='orange', linewidth=2)
ax.set_title("Learning curve — ANN-imitator of the FLS")
ax.set_xlabel('Epoch'); ax.set_ylabel('MSE')
ax.set_yscale('log'); ax.legend(); ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()In [ ]:
# ============================================================
# STEP 2.4 – FLS vs ANN-imitator comparison
# ============================================================
imitator.eval()
with torch.no_grad():
y_pred_ann = imitator(X_te_ft).numpy()
# De-normalization
y_pred_ann_real = scaler_y_fls.inverse_transform(y_pred_ann)
y_true_real = scaler_y_fls.inverse_transform(y_te_f)
mae_imitator = mean_absolute_error(y_true_real, y_pred_ann_real)
mse_imitator = mean_squared_error(y_true_real, y_pred_ann_real)
print(f"FLS vs ANN-imitator comparison on the test set:")
print(f" MAE = {mae_imitator:.2f}%")
print(f" MSE = {mse_imitator:.2f}")
print(f" RMSE = {np.sqrt(mse_imitator):.2f}%")
# Comparison plot
fig, ax = plt.subplots(figsize=(7, 7))
ax.scatter(y_true_real, y_pred_ann_real, alpha=0.3, s=8, color='steelblue')
ax.plot([0, 100], [0, 100], 'r--', linewidth=2)
ax.set_xlabel("Power according to FLS [%]"); ax.set_ylabel("Power according to ANN [%]")
ax.set_title(f"FLS vs ANN-imitator (MAE={mae_imitator:.2f}%)")
ax.grid(True, alpha=0.3)
ax.set_xlim([0, 100]); ax.set_ylim([0, 100])
plt.tight_layout()
plt.show()In [ ]:
# ============================================================
# STEP 2.5A – FLS with a fallback rule
# ============================================================
# New table = original + one row.
# The '_strength' field enforces a constant activation strength
# (the inference engine checks it before computing the min).
RULE_BASE_FALLBACK = RULE_BASE + [
{'temp': '-', 'hum': '-', 'occ': '-', 'co2': '-', 'time': '-',
'power': 'medium', '_strength': 0.3}, # R00 — FALLBACK
]
def evaluate_rules_v2(rule_base, values):
# Extended inference engine — handles the optional '_strength' field
# that forces a constant rule activation strength (for fallbacks).
active = []
for rule in rule_base:
if '_strength' in rule:
active.append((rule['_strength'], rule['power']))
continue
strengths = []
for variable in INPUT_VARIABLES:
label = rule[variable]
if label == '-':
continue
u, mf = FUZZY_SETS[(variable, label)]
mu = fuzz.interp_membership(u, mf, values[variable])
strengths.append(mu)
strength = min(strengths) if strengths else 0.0
active.append((strength, rule['power']))
return active
def fls_controller_fallback(temp, hum, occ, co2, time):
values = {'temp': temp, 'hum': hum, 'occ': occ, 'co2': co2, 'time': time}
active = evaluate_rules_v2(RULE_BASE_FALLBACK, values)
return defuzzify(active)
# --- Quick test: comparison on several random scenarios ---
rng_diag = np.random.default_rng(99)
print("Comparison: FLS vs FLS+fallback (10 random scenarios):")
print("-" * 72)
print(f" {'temp':>5} {'hum':>5} {'occ':>5} {'co2':>6} {'time':>5} │ "
f"{'FLS':>7} {'FLS+fb':>7} {'Δ':>6} {'max μ':>6}")
print("-" * 72)
for _ in range(10):
t = rng_diag.uniform(15, 35)
w = rng_diag.uniform(20, 90)
o = rng_diag.uniform(0, 100)
c = rng_diag.uniform(400, 2000)
p = rng_diag.uniform(0, 24)
power_orig, agg_orig = fls_controller(t, w, o, c, p)
power_fb, _ = fls_controller_fallback(t, w, o, c, p)
delta = power_fb - power_orig
max_mu = np.max(agg_orig)
print(f" {t:5.1f} {w:5.1f} {o:5.1f} {c:6.0f} {p:5.1f} │ "
f"{power_orig:7.1f} {power_fb:7.1f} {delta:+6.1f} {max_mu:6.3f}")In [ ]:
# ============================================================
# STEP 2.5B – Comparing ANNs trained on FLS vs FLS+fallback data
# ============================================================
def generate_fls_data(controller_fn, n_samples, seed):
rng = np.random.default_rng(seed)
X_list, y_list = [], []
for _ in range(n_samples):
t = rng.uniform(15, 35)
w = rng.uniform(20, 90)
o = rng.uniform(0, 100)
c = rng.uniform(400, 2000)
p = rng.uniform(0, 24)
try:
power, _ = controller_fn(t, w, o, c, p)
except Exception:
continue
X_list.append([t, w, o, c, p])
y_list.append(power)
return np.array(X_list), np.array(y_list).reshape(-1, 1)
def train_imitator(X_raw, y_raw, n_epochs=80, seed=42):
scX = MinMaxScaler()
scY = MinMaxScaler()
Xn = scX.fit_transform(X_raw)
yn = scY.fit_transform(y_raw)
Xtr, Xte, ytr, yte = train_test_split(Xn, yn, test_size=0.15, random_state=seed)
Xtr, Xval, ytr, yval = train_test_split(Xtr, ytr, test_size=0.176, random_state=seed)
ds = TensorDataset(torch.FloatTensor(Xtr), torch.FloatTensor(ytr))
ldr = DataLoader(ds, batch_size=64, shuffle=True)
torch.manual_seed(seed)
model = FLSImitator()
opt = torch.optim.Adam(model.parameters(), lr=0.001)
crit = nn.MSELoss()
hist_tr, hist_val = [], []
for ep in range(1, n_epochs + 1):
model.train()
s = 0.0
for Xb, yb in ldr:
opt.zero_grad()
loss = crit(model(Xb), yb)
loss.backward()
opt.step()
s += loss.item() * len(Xb)
hist_tr.append(s / len(Xtr))
model.eval()
with torch.no_grad():
hist_val.append(crit(model(torch.FloatTensor(Xval)),
torch.FloatTensor(yval)).item())
# Evaluation on the test set
model.eval()
with torch.no_grad():
pred = model(torch.FloatTensor(Xte)).numpy()
y_pred_real = scY.inverse_transform(pred)
y_true_real = scY.inverse_transform(yte)
return {
'model': model, 'scX': scX, 'scY': scY,
'y_pred': y_pred_real, 'y_true': y_true_real,
'hist_tr': hist_tr, 'hist_val': hist_val,
'mae': mean_absolute_error(y_true_real, y_pred_real),
'rmse': np.sqrt(mean_squared_error(y_true_real, y_pred_real)),
}
# --- Generate data and train ---
N_SAMPLES = 10000
print("Generating data from the FLS (15 rules)...")
X_orig, y_orig = generate_fls_data(fls_controller, N_SAMPLES, seed=42)
print("Generating data from the FLS+fallback (16 rules)...")
X_fb, y_fb = generate_fls_data(fls_controller_fallback, N_SAMPLES, seed=42)
print("\nTraining the ANN-imitator on FLS data...")
results_orig = train_imitator(X_orig, y_orig)
print("Training the ANN-imitator on FLS+fallback data...")
results_fb = train_imitator(X_fb, y_fb)
# --- Report ---
print()
print("=" * 55)
print("COMPARISON: ANN trained on FLS vs FLS+fallback data")
print("=" * 55)
print(f" {'Data source':<28} {'MAE':>8} {'RMSE':>8}")
print("-" * 55)
print(f" {'FLS (15 rules)':<28} {results_orig['mae']:8.2f}% {results_orig['rmse']:8.2f}%")
print(f" {'FLS + fallback (16 rules)':<28} {results_fb['mae']:8.2f}% {results_fb['rmse']:8.2f}%")
print("=" * 55)In [ ]:
# --- Comparative visualization ---
fig, axes = plt.subplots(1, 2, figsize=(14, 6))
for i, (name, result, color) in enumerate([
('ANN trained on FLS (15 rules)', results_orig, 'steelblue'),
('ANN trained on FLS+fallback (16)', results_fb, 'darkorange'),
]):
ax = axes[i]
ax.scatter(result['y_true'], result['y_pred'], alpha=0.3, s=8, color=color)
ax.plot([0, 100], [0, 100], 'r--', linewidth=2)
ax.set_xlabel("Power according to FLS [%]"); ax.set_ylabel("Power according to ANN [%]")
ax.set_title(f"{name}\nMAE = {result['mae']:.2f}%", fontsize=11)
ax.grid(True, alpha=0.3)
ax.set_xlim([0, 100]); ax.set_ylim([0, 100])
plt.suptitle("Impact of FLS data quality on the ANN-imitator", fontsize=13, fontweight='bold')
plt.tight_layout()
plt.savefig('comparison_fls_vs_fallback_ann.png', dpi=120, bbox_inches='tight')
plt.show()
# Learning curves side by side
fig, axes = plt.subplots(1, 2, figsize=(14, 4))
for i, (name, result, color) in enumerate([
('FLS (15 rules)', results_orig, 'steelblue'),
('FLS+fallback (16)', results_fb, 'darkorange'),
]):
ax = axes[i]
ax.plot(result['hist_tr'], label='Train', color=color, linewidth=2)
ax.plot(result['hist_val'], label='Validation', color='gray', linewidth=2, linestyle='--')
ax.set_title(f"Learning curve — {name}")
ax.set_xlabel('Epoch'); ax.set_ylabel('MSE')
ax.set_yscale('log'); ax.legend(); ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()