基于机器学习的入侵检测系统python源码+详细注释(课程设计新项目).zip

import glob import logging import os import platform import random import re import shutil import subprocess import time from contextlib import contextmanager from copy import copy from pathlib import Path import cv2 import math import matplotlib import matplotlib.pyplot as plt import numpy as np import torch import torch.nn as nn import yaml from PIL import Image from scipy.cluster.vq import kmeans from scipy.signal import butter, filtfilt from tqdm import tqdm from utils.google_utils import gsutil_getsize from utils.torch_utils import is_parallel, init_torch_seeds # Set printoptions torch.set_printoptions(linewidth=320, precision=5, profile='long') np.set_printoptions(linewidth=320, formatter={'float_kind': '{:11.5g}'.format}) # format short g, %precision=5 matplotlib.rc('font', **{'size': 11}) # Prevent OpenCV from multithreading (to use PyTorch DataLoader) cv2.setNumThreads(0) @contextmanager def torch_distributed_zero_first(local_rank: int): """ Decorator to make all processes in distributed training wait for each local_master to do something. """ if local_rank not in [-1, 0]: torch.distributed.barrier() yield if local_rank == 0: torch.distributed.barrier() def set_logging(rank=-1): logging.basicConfig( format="%(message)s", level=logging.INFO if rank in [-1, 0] else logging.WARN) def init_seeds(seed=0): random.seed(seed) np.random.seed(seed) init_torch_seeds(seed) def get_latest_run(search_dir='./runs'): # Return path to most recent 'last.pt' in /runs (i.e. to --resume from) last_list = glob.glob(f'{search_dir}/**/last*.pt', recursive=True) return max(last_list, key=os.path.getctime) if last_list else '' def check_git_status(): # Suggest 'git pull' if repo is out of date if platform.system() in ['Linux', 'Darwin'] and not os.path.isfile('/.dockerenv'): s = subprocess.check_output('if [ -d .git ]; then git fetch && git status -uno; fi', shell=True).decode('utf-8') if 'Your branch is behind' in s: print(s[s.find('Your branch is behind'):s.find('\n\n')] + '\n') def check_img_size(img_size, s=32): # Verify img_size is a multiple of stride s new_size = make_divisible(img_size, int(s)) # ceil gs-multiple if new_size != img_size: print('WARNING: --img-size %g must be multiple of max stride %g, updating to %g' % (img_size, s, new_size)) return new_size def check_anchors(dataset, model, thr=4.0, imgsz=640): # Check anchor fit to data, recompute if necessary print('\nAnalyzing anchors... ', end='') m = model.module.model[-1] if hasattr(model, 'module') else model.model[-1] # Detect() shapes = imgsz * dataset.shapes / dataset.shapes.max(1, keepdims=True) scale = np.random.uniform(0.9, 1.1, size=(shapes.shape[0], 1)) # augment scale wh = torch.tensor(np.concatenate([l[:, 3:5] * s for s, l in zip(shapes * scale, dataset.labels)])).float() # wh def metric(k): # compute metric r = wh[:, None] / k[None] x = torch.min(r, 1. / r).min(2)[0] # ratio metric best = x.max(1)[0] # best_x aat = (x > 1. / thr).float().sum(1).mean() # anchors above threshold bpr = (best > 1. / thr).float().mean() # best possible recall return bpr, aat bpr, aat = metric(m.anchor_grid.clone().cpu().view(-1, 2)) print('anchors/target = %.2f, Best Possible Recall (BPR) = %.4f' % (aat, bpr), end='') if bpr < 0.98: # threshold to recompute print('. Attempting to generate improved anchors, please wait...' % bpr) na = m.anchor_grid.numel() // 2 # number of anchors new_anchors = kmean_anchors(dataset, n=na, img_size=imgsz, thr=thr, gen=1000, verbose=False) new_bpr = metric(new_anchors.reshape(-1, 2))[0] if new_bpr > bpr: # replace anchors new_anchors = torch.tensor(new_anchors, device=m.anchors.device).type_as(m.anchors) m.anchor_grid[:] = new_anchors.clone().view_as(m.anchor_grid) # for inference m.anchors[:] = new_anchors.clone().view_as(m.anchors) / m.stride.to(m.anchors.device).view(-1, 1, 1) # loss check_anchor_order(m) print('New anchors saved to model. Update model *.yaml to use these anchors in the future.') else: print('Original anchors better than new anchors. Proceeding with original anchors.') print('') # newline def check_anchor_order(m): # Check anchor order against stride order for YOLOv5 Detect() module m, and correct if necessary a = m.anchor_grid.prod(-1).view(-1) # anchor area da = a[-1] - a[0] # delta a ds = m.stride[-1] - m.stride[0] # delta s if da.sign() != ds.sign(): # same order print('Reversing anchor order') m.anchors[:] = m.anchors.flip(0) m.anchor_grid[:] = m.anchor_grid.flip(0) def check_file(file): # Search for file if not found if os.path.isfile(file) or file == '': return file else: files = glob.glob('./**/' + file, recursive=True) # find file assert len(files), 'File Not Found: %s' % file # assert file was found assert len(files) == 1, "Multiple files match '%s', specify exact path: %s" % (file, files) # assert unique return files[0] # return file def check_dataset(dict): # Download dataset if not found val, s = dict.get('val'), dict.get('download') if val and len(val): val = [os.path.abspath(x) for x in (val if isinstance(val, list) else [val])] # val path if not all(os.path.exists(x) for x in val): print('\nWARNING: Dataset not found, nonexistent paths: %s' % [*val]) if s and len(s): # download script print('Downloading %s ...' % s) if s.startswith('http') and s.endswith('.zip'): # URL f = Path(s).name # filename torch.hub.download_url_to_file(s, f) r = os.system('unzip -q %s -d ../ && rm %s' % (f, f)) # unzip else: # bash script r = os.system(s) print('Dataset autodownload %s\n' % ('success' if r == 0 else 'failure')) # analyze return value else: raise Exception('Dataset not found.') def make_divisible(x, divisor): # Returns x evenly divisible by divisor return math.ceil(x / divisor) * divisor def labels_to_class_weights(labels, nc=80): # Get class weights (inverse frequency) from training labels if labels[0] is None: # no labels loaded return torch.Tensor() labels = np.concatenate(labels, 0) # labels.shape = (866643, 5) for COCO classes = labels[:, 0].astype(np.int) # labels = [class xywh] weights = np.bincount(classes, minlength=nc) # occurrences per class # Prepend gridpoint count (for uCE training) # gpi = ((320 / 32 * np.array([1, 2, 4])) ** 2 * 3).sum() # gridpoints per image # weights = np.hstack([gpi * len(labels) - weights.sum() * 9, weights * 9]) ** 0.5 # prepend gridpoints to start weights[weights == 0] = 1 # replace empty bins with 1 weights = 1 / weights # number of targets per class weights /= weights.sum() # normalize return torch.from_numpy(weights) def labels_to_image_weights(labels, nc=80, class_weights=np.ones(80)): # Produces image weights based on class mAPs n = len(labels) class_counts = np.array([np.bincount(labels[i][:, 0].astype(np.int), minlength=nc) for i in range(n)]) image_weights = (class_weights.reshape(1, nc) * class_counts).sum(1) # index = random.choices(range(n), weights=image_weights, k=1) # weight image sample return image_weights def coco80_to_coco91_class(): # converts 80-index (val2014) to 91-index (paper) # https://tech.amikelive.com/node-718/what-object-categories-labels-are-in-coco-dataset/ # a = np.loadtxt('data/coco.names', dtype='str', delimiter='\n') # b = np.loadtxt('data/