点击上方“小白学视觉”，选择加"星标"或“置顶”

重磅干货，第一时间送达

 1import os
 2from abc import ABCMeta, abstractmethod
 3
 4
 5class DataProcessor(metaclass=ABCMeta):
 6    """Base processor to be used for all preparation."""
 7    def __init__


    
(self, input_directory, output_directory):
 8        self.input_directory = input_directory
 9        self.output_directory = output_directory
10
11    @abstractmethod
12    def read(self):
13        """Read raw data."""
14
15    @abstractmethod
16    def process(self):
17        """Processes raw data. This step should create the raw dataframe with all the required features. Shouldn't implement statistical or text cleaning."""
18
19    @abstractmethod
20    def save(self):
21        """Saves processed data."""
22
23
24class Trainer(metaclass=ABCMeta):
25    """Base trainer to be used for all models."""
26
27    def __init__(self, directory):
28        self.directory = directory
29        self.model_directory = os.path.join(directory, 'models')
30



    
31    @abstractmethod
32    def preprocess(self):
33        """This takes the preprocessed data and returns clean data. This is more about statistical or text cleaning."""
34
35    @abstractmethod
36    def set_model(self):
37        """Define model here."""
38
39    @abstractmethod
40    def fit_model(self):
41        """This takes the vectorised data and returns a trained model."""
42
43    @abstractmethod
44    def generate_metrics(self):
45        """Generates metric with trained model and test data."""
46
47    @abstractmethod
48    def save_model(self, model_name):
49        """This method saves the model in our required format."""
50
51
52class Predict


    
(metaclass=ABCMeta):
53    """Base predictor to be used for all models."""
54
55    def __init__(self, directory):
56        self.directory = directory
57        self.model_directory = os.path.join(directory, 'models')
58
59    @abstractmethod
60    def load_model(self):
61        """Load model here."""
62
63    @abstractmethod
64    def preprocess(self):
65        """This takes the raw data and returns clean data for prediction."""
66
67    @abstractmethod
68    def predict(self):
69        """This is used for prediction."""
70
71
72class BaseDB(metaclass=ABCMeta):
73    """ Base database class to be used for all DB connectors."""
74    @abstractmethod
75


    
    def get_connection(self):
76        """This creates a new DB connection."""
77    @abstractmethod
78    def close_connection(self):
79        """This closes the DB connection."""

1def set_seed(args):
2    random.seed(args.seed)
3    np.random.seed(args.seed)
4    torch.manual_seed(args.seed)
5    if args.n_gpu > 0:
6        torch.cuda.manual_seed_all(args.seed)

 1import time
 2
 3def timing(f):
 4    """Decorator for timing functions
 5    Usage:
 6    @timing
 7    def function(a):
 8        pass
 9    """
10
11
12    @wraps(f)
13    def wrapper(*args, **kwargs):
14        start = time.time()
15        result = f(*args, **kwargs)
16        end = time.time()
17        print('function:%r took: %2.2f sec' % (f.__name__,  end - start))
18        return result
19    return wrapp

 1import os
 2
 3def run_command(cmd):
 4    return os.system(cmd)
 5
 6def shutdown(seconds=0, os='linux'):
 7    """Shutdown system after seconds given. Useful for shutting EC2 to save costs."""
 8    if os == 'linux':
 9        run_command('sudo shutdown -h -t sec %s' % seconds)
10    elif os == 'windows':
11        run_command('shutdown -s -t %s' % seconds)

 1import json
 2import os
 3
 4from sklearn.metrics import (accuracy_score, classification_report,
 5                             confusion_matrix, f1_score, fbeta_score)
 6
 7def get_metrics(y, y_pred, beta=2, average_method='macro', y_encoder=None):
 8    if y_encoder:
 9        y = y_encoder.inverse_transform(y)
10        y_pred = y_encoder.inverse_transform(y_pred)
11    return {
12        'accuracy': round(accuracy_score(y, y_pred), 4),
13        'f1_score_macro': round(f1_score(y, y_pred, average=average_method), 4),
14        'fbeta_score_macro': round(fbeta_score(y, y_pred, beta, average=average_method), 4),
15        'report': classification_report(y, y_pred, output_dict=True),
16        'report_csv': classification_report(y, y_pred, output_dict=False).replace('\n','\r\n')
17    }



    
18
19
20def save_metrics(metrics: dict, model_directory, file_name):
21    path = os.path.join(model_directory, file_name + '_report.txt')
22    classification_report_to_csv(metrics['report_csv'], path)
23    metrics.pop('report_csv')
24    path = os.path.join(model_directory, file_name + '_metrics.json')
25    json.dump(metrics, open(path, 'w'), indent=4)

1pip install fastapi uvicorn gunicorn
2gunicorn -w


    
 4 -k uvicorn.workers.UvicornH11Worker main:app