fuzzy-logic-demo/洗衣机.py

from enum import Enum
from functools import reduce
from scipy import integrate
import itertools

DELIMETER = ","


class DirtType(Enum):
    Greasy = 0
    Medium = 1
    NotGreasy = 2

    def membership_grade(self, dirtness: float) -> float:
        if self is DirtType.NotGreasy:
            return max(0, 1 - 1 / 50 * dirtness)
        elif self is DirtType.Medium:
            if dirtness < 50:
                return 1 / 50 * dirtness
            else:
                return 2 - 1 / 50 * dirtness
        else:
            return max(0, 1 / 50 * (dirtness - 50))

    @classmethod
    def membership_grades(cls, dirness: float) -> dict["DirtType", float]:
        ans = {}
        for v in cls:
            ans[v] = v.membership_grade(dirness)
        return ans


class DirtQuality(Enum):
    Small = 0
    Medium = 1
    Large = 2

    def membership_grade(self, type_of_dirt: float) -> float:
        if self is DirtQuality.Small:
            return max(0, 1 - 1 / 50 * type_of_dirt)
        elif self is DirtQuality.Medium:
            if type_of_dirt < 50:
                return 1 / 50 * type_of_dirt
            else:
                return 2 - 1 / 50 * type_of_dirt
        else:
            return max(0, 1 / 50 * (type_of_dirt - 50))

    @classmethod
    def membership_grades(cls, dirness: float) -> dict["DirtQuality", float]:
        ans = {}
        for v in cls:
            ans[v] = v.membership_grade(dirness)
        return ans


class WashTime(Enum):
    VS = 0
    S = 1
    M = 2
    L = 3
    VL = 4

    def reverse_membership_grade(self) -> float:
        v = [10, 20, 30, 40, 50]
        return v[self.value]

    def membership_grade(self, wash_time: float) -> float:
        fns = {
            WashTime.VS: lambda t: min(1 / 10 * t, -1 / 10 * (t - 20)),
            WashTime.S: lambda t: min(1 / 10 * (t - 10), -1 / 10 * (t - 30)),
            WashTime.M: lambda t: min(1 / 10 * (t - 20), -1 / 10 * (t - 40)),
            WashTime.L: lambda t: min(1 / 10 * (t - 30), -1 / 10 * (t - 50)),
            WashTime.VL: lambda t: min(1 / 10 * (t - 40), -1 / 10 * (t - 60)),
        }
        return max(0, fns[self](wash_time))


class Rule:
    def __init__(
        self, dirtness: DirtQuality, type_of_dirt: DirtType, wash_time: WashTime
    ) -> None:
        self.dirtness = dirtness
        self.type_of_dirt = type_of_dirt
        self.wash_time = wash_time

    def rule_support(self, dirtness: float, type_of_dirt: float) -> float:
        # 计算隶属度
        dirtness_grade = self.dirtness.membership_grade(dirtness)
        type_of_dirt_grade = self.type_of_dirt.membership_grade(type_of_dirt)
        # 计算t范数 min
        return min(dirtness_grade, type_of_dirt_grade)


class InferenceEngine:
    def __init__(self, rules: list[Rule]) -> None:
        self.rules = rules

    def output(
        self, dirtness: float, type_of_dirt: float
    ) -> list[tuple[float, WashTime]]:
        # 使用规则
        ans = map(
            lambda rule: (rule.rule_support(dirtness, type_of_dirt), rule.wash_time),
            self.rules,
        )
        ans = list(ans)
        return ans


# 聚合模糊输出，使用离散的质心计算方式，数学积
def aggregate_discret_AP(pairs: list[tuple[float, WashTime]]) -> float:
    return sum(
        map(lambda pair: pair[0] * pair[1].reverse_membership_grade(), pairs)
    ) / sum(map(lambda pair: pair[0], pairs))


# 使用连续的质心计算方式，数学积和代数和
def aggregate_continue_AP(pairs: list[tuple[float, WashTime]]) -> float:
    def f(time: float) -> float:
        return S_AS(map(lambda pair: pair[0] * pair[1].membership_grade(time), pairs))

    base = integrate.quad(f, 0, 60)[0]
    return integrate.quad(lambda v: v * f(v), 0, 60)[0] / base


def S_AS(iter) -> float:
    "代数和，输入数组"
    ans = 0
    iter = tuple(iter)
    for i in range(len(iter)):
        n = i + 1
        if n & 1:
            flag = 1
        else:
            flag = -1
        # 选择n个数，相乘再相加
        for collection in itertools.combinations(iter, n):
            ans += flag * reduce(lambda x, y: x * y, collection)
    return ans


# 标准交
def aggregate_continue_SI(pairs: list[tuple[float, WashTime]]) -> float:
    """lambda: fi=min(alpha, mu), f=max(fi)"""

    def f(time: float) -> float:
        return max(
            map(lambda pair: min(pair[0], pair[1].membership_grade(time)), pairs)
        )

    return integrate.quad(lambda v: v * f(v), 0, 60)[0] / integrate.quad(f, 0, 60)[0]


rules = [
    Rule(DirtQuality.Small, DirtType.Greasy, WashTime.L),
    Rule(DirtQuality.Medium, DirtType.Greasy, WashTime.L),
    Rule(DirtQuality.Large, DirtType.Greasy, WashTime.VL),
    Rule(DirtQuality.Small, DirtType.Medium, WashTime.M),
    Rule(DirtQuality.Medium, DirtType.Medium, WashTime.M),
    Rule(DirtQuality.Large, DirtType.Medium, WashTime.L),
    Rule(DirtQuality.Small, DirtType.NotGreasy, WashTime.VS),
    Rule(DirtQuality.Medium, DirtType.NotGreasy, WashTime.S),
    Rule(DirtQuality.Large, DirtType.NotGreasy, WashTime.S),
]

inputs = [
    [0, 0],
    [10, 10],
    [50, 10],
    [10, 50],
    [50, 50],
    [10, 70],
    [50, 70],
    [70, 10],
    [70, 50],
    [70, 70],
    [100, 100],
]


def main(inputs, rules):
    print("脏的数量0-100，越大越脏；脏的类型0-100，越大油脂越多")
    print(DELIMETER.join(["脏的数量", "脏的类型", "清洗时间/min"]))
    engine = InferenceEngine(rules)
    for input in inputs:
        # print("脏的程度", input[0], "脏的类型", input[1])
        outputs = engine.output(input[0], input[1])
        output = aggregate_discret_AP(outputs)
        output1 = aggregate_continue_AP(outputs)
        output2 = aggregate_continue_SI(outputs)
        print(
            DELIMETER.join(["{:.2f}"] * 5).format(
                input[0], input[1], output, output1, output2
            )
        )


if __name__ == "__main__":
    main(inputs, rules)