객체 주입 후 한번에 실행하기
from abc import ABC, abstractmethod
from typing import List, Tuple, Union, Dict
import torch
# Base class for creating ts analytic items
# classmethod는 staticmethod와 유사한데, staticmethod는 class변수에 access
# 안 하는데 비해, classmethod는 cls로 access한다.
class ItemBase(ABC):
@classmethod # instance 관점이 아닌 class전체 관점에서 변수를 다룰 수 있음
def add_instance(cls, ins): # class의 보편적인 값을 다룬다는 의미에서 class
cls._items.append(ins) # 약자인 cls를 인자로 받음.
print(cls._items, len(cls._items)) # self대신 cls전달 받음
@abstractmethod
def run(self, *args):
pass
# Derivated class for outlier detection
# 생성만 하면 객체 리스트가 자동으로 만들어짐
class OutlierDetector(ItemBase):
_items = [] # ItemBase에 있었다면 모든 자식의 instance를 저장
def __init__(self, *args): # 여기에 있으면, 현 class 인스턴스 저장
print(args)
self.add_instance(self)
def run(self, *args):
print(args)
# Derivated class for forecasting time series
class Forecastor(ItemBase):
_items = []
def __init__(self, *args):
super().__init__()
print(args)
self.add_instance(self)
def run(self, *args):
print(args)
# Collect multiple items and process them at once
class ProcessItems(object):
def __init__(self, items: List = [])->None:
self.items = items
def process(self)->List:
res = []
for item in self.items:
res.append(item.run())
return res
a3 = ProcessItems([OutlierDetector(), Forecastor()])
res = a3.process()
print(res)
응용 예
from abc import ABC, abstractmethod
from typing import List, Tuple, Union, Dict
import torch
import pdb
# Base class for creating ts(Time sereis) analytic items
class ItemBase(ABC):
_modes = ['start', 'stable', 'finish']
@abstractmethod
def run(self, *args):
pass
# cls로 class변수에 access 가능(staticmethod와 차이점)
@classmethod
def _load_weight(cls, items):
cls.models = {}
for key, weight in items.items():
if key not in cls._modes:
assert False, 'Mode Error!!'
# cls.models[key] = cls.dnn_model(weight)
cls.models[key] = None
# Child class for forecasting time series
class Forecastor(ItemBase):
def __init__(self, items):
# self.dnn_model = dnn_model()
self._load_weight(items)
def run(self, *args):
batch, mode = args
# outs = self.models[mode].predict(batch)
print(batch.shape)
return batch.shape
# Child class for outlier detection
class AnomalyDetector(ItemBase):
def __init__(self, items):
# self.dnn_model = dnn_model()
self._load_weight(items)
def run(self, *args):
batch, mode = args
print(batch.shape)
return batch.shape
# Main processor to run all items
class Processor(object):
def __init__(self, items):
self.items = items
def process(self, ts):
res = []
# mode = ts_decision(ts) # 기동,정상,정지부 판정
mode = 'start'
for item in self.items:
outs = item.run(ts, mode)
res.append(outs)
return res
# There are two local processes to handle time series data
forecast = Forecastor({'start':'m1.pt', 'stable': 'm2.pt', 'finish': 'm3.pt'})
anodetec = AnomalyDetector({'start':'m1.pt', 'stable': 'm2.pt', 'finish': 'm3.pt'})
batch = torch.rand(8,32,16)
# If you need to add another process, first define and simply add it to arg list
a1 = Processor([forecast, anodetec])
a1.process(batch)
torch.Size([8, 32, 16])
torch.Size([8, 32, 16])
Out[7]:
[torch.Size([8, 32, 16]), torch.Size([8, 32, 16])]
from abc import ABC, abstractmethod
from typing import List, Tuple, Union, Dict
import numpy as np
import torch
import pdb
def _gt(x,v): return True if x >= v else False # _gt = lambda x, v: True if x >= v else False
def _lt(x,v): return True if x <= v else False # _lt = lambda x, v: True if x <= v else False
def _eq(x,v): return True if x == v else False # _eq = lambda x, v: True if x == v else False
def _in(x,v1,v2): return True if (x >= v1) & (x<= v2) else False # _in = lambda x, v1, v2: True if (x >= v1) & (x<= v2) else False
op1={}
op1['s1'] = _gt
_gt(2,1), _lt(2,1), _in(2,3,4), op1['s1'](2,1)
(True, False, False, True)
# Base class for creating ts(Time sereis) analytic items
class ItemBase(ABC):
@classmethod
def add(cls, ins):
cls._items[ins._name] = ins
@classmethod
def get_items(cls):
return list(cls._items.keys())
@classmethod
def get_conditions(cls):
items = []
for k, v in cls._items.items():
iks = [(ik, str(iv[0]).split()[1]) for ik,iv in v.terms.items()]
items.append({k: iks})
return items
def _save_items(self, items):
if type(items) != dict:
return None
for k,v in items.items():
try:
self.terms[k] = v
except Exception as e:
return None
def refresh(self, ins):
keys = self.terms.keys()
satisfied = {}
for k, v in ins.items():
if k in keys:
try:
check = self.terms[k][0](*v)
except:
satisfied[k] = -1
else:
self.terms[k][1] = check
satisfied[k] = check
return satisfied
class Derivated(ItemBase):
class NoneDict(Dict):
def __getitem__(self, key):
return dict.get(self, key)
_items = NoneDict()
def __init__(self, _name, items):
self.terms = {}
self._name = _name
self._save_items(items)
Derivated.add(Derivated('NOx', {'op1':[_gt, False], 'op2': [_lt, False], 'op3': [_in, False]}))
Derivated.add(Derivated('O2', {'op4':[_lt, False], 'op5': [_in, False]}))
if Derivated._items['NOx']:
print(Derivated._items['NOx'].refresh({'op1': (1,2), 'op2': (3,4), 'op3': (4,1,5)}))
if Derivated._items['O2']:
print(Derivated._items['O2'].refresh({'op4': (1,2), 'op2': (3,4), 'op3': (4,1,5)}))
{'op1': False, 'op2': True, 'op3': True}
{'op4': True}
Derivated.add(Derivated('NOxIn', {'op1':[_gt, False], 'op2': [_lt, False], 'op3': [_in, False]}))
Derivated.add(Derivated('O2out', {'op4':[_lt, False], 'op5': [_in, False]}))
Derivated.get_items()
['NOx', 'O2', 'NOxIn', 'O2out']
Derivated.get_conditions()
[{'NOx': [('op1', '_gt'), ('op2', '_lt'), ('op3', '_in')]},
{'O2': [('op4', '_lt'), ('op5', '_in')]},
{'NOxIn': [('op1', '_gt'), ('op2', '_lt'), ('op3', '_in')]},
{'O2out': [('op4', '_lt'), ('op5', '_in')]}]
[References]
1. D:\2022\Pattern_Test
