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)