2023년 5월 30일 화요일

yolov8 학습 및 테스트

# yolov8 설치 후 학습과 테스트의 예


# 경로- g:\2023
> yolo task=detect mode=train model=yolov8m.pt imgsz=1280 data=fire2023_1.yaml
epochs=50 batch=16 name=yolov8m_v8_50e
> yolo predict model=best.pt source=parking_lot5.mp4  # test

2023년 5월 9일 화요일

pip가 깨졌을 때

 (1) pip가 깨어졌을 때, base에서

conda install --force-reinstall pip


(2) 아래 오류

ERROR: Could not install packages due to an OSError: [Errno 2] No such file or directory: 

> pip3 install --upgrade --user pip

2023년 5월 2일 화요일

python 패턴

 객체 주입 후 한번에 실행하기

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