Python utils 模块，preprocess() 实例源码

def extract_features(tweet, combinedKeywords, pronDict):
    preprocessed = ut.preprocess(tweet)

    # ignore empty preprocessed tweets or retweets
    if len(preprocessed) == 0 or ut.is_retweet(tweet):
        return []
    else:
        sentLength = la.get_sentence_length(preprocessed)
        exclMarks = la.get_exclamation_marks(tweet)
        gradeLvl = get_flesch_grade_level(preprocessed, pronDict)
        keyCount = get_keywords_count(preprocessed, combinedKeywords)
        # ensure same order everytime
        keys = sorted(list(keyCount.keys()))

        # put all features together
        features = [ sentLength, exclMarks, gradeLvl ]
        for key in keys:
            features.append(keyCount.get(key))
        # return array
        return features

def __init__(self, content_path, style_path):

        # 4D representations of the given content and style images
        self.content_image = utils.preprocess(plt.imread(content_path))[np.newaxis]
        self.style_image = utils.preprocess(plt.imread(style_path))[np.newaxis]

        # The session and graph used for evaluating the content and style of the
        # given content and style images
        self.evaluation_g = tf.Graph()
        self.evaluation_sess = tf.Session(graph=self.evaluation_g)

        # The outputs (:0) of the intermediate layers of the VGG16 model used to represent the
        # content and style of an input to the model
        self.content_layer = config["content_layer"]
        self.style_layers = config["style_layers"]

        with self.evaluation_g.as_default():

            # Import the VGG16 ImageNet predictor model graph into the evaluation_g member variable
            tf.import_graph_def(utils.get_vgg_model(), name="vgg")

            # The input to the VGG16 predictor model is the output (:0) of the first operation of the graph
            self.input_tensor = [op.name for op in self.evaluation_g.get_operations()][0] + ":0"

def std(image):
    return utils.preprocess.per_image_standardization(image)

def detect(sess, model, names, image, path):
    preprocess = eval(args.preprocess)
    _, height, width, _ = image.get_shape().as_list()
    _image = read_image(path)
    image_original = np.array(np.uint8(_image))
    if len(image_original.shape) == 2:
        image_original = np.repeat(np.expand_dims(image_original, -1), 3, 2)
    image_height, image_width, _ = image_original.shape
    image_std = preprocess(np.array(np.uint8(_image.resize((width, height)))).astype(np.float32))
    feed_dict = {image: np.expand_dims(image_std, 0)}
    tensors = [model.conf, model.xy_min, model.xy_max]
    conf, xy_min, xy_max = sess.run([tf.check_numerics(t, t.op.name) for t in tensors], feed_dict=feed_dict)
    boxes = utils.postprocess.non_max_suppress(conf[0], xy_min[0], xy_max[0], args.threshold, args.threshold_iou)
    scale = [image_width / model.cell_width, image_height / model.cell_height]
    fig = plt.figure()
    ax = fig.gca()
    ax.imshow(image_original)
    colors = [prop['color'] for _, prop in zip(names, itertools.cycle(plt.rcParams['axes.prop_cycle']))]
    cnt = 0
    for _conf, _xy_min, _xy_max in boxes:
        index = np.argmax(_conf)
        if _conf[index] > args.threshold:
            wh = _xy_max - _xy_min
            _xy_min = _xy_min * scale
            _wh = wh * scale
            linewidth = min(_conf[index] * 10, 3)
            ax.add_patch(patches.Rectangle(_xy_min, _wh[0], _wh[1], linewidth=linewidth, edgecolor=colors[index], facecolor='none'))
            ax.annotate(names[index] + ' (%.1f%%)' % (_conf[index] * 100), _xy_min, color=colors[index])
            cnt += 1
    fig.canvas.set_window_title('%d objects detected' % cnt)
    ax.set_xticks([])
    ax.set_yticks([])
    return fig

def make_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('path', help='input image path')
    parser.add_argument('-c', '--config', nargs='+', default=['config.ini'], help='config file')
    parser.add_argument('-p', '--preprocess', default='std', help='the preprocess function')
    parser.add_argument('-t', '--threshold', type=float, default=0.3)
    parser.add_argument('--threshold_iou', type=float, default=0.4, help='IoU threshold')
    parser.add_argument('-e', '--exts', nargs='+', default=['.jpg', '.png'])
    parser.add_argument('--level', default='info', help='logging level')
    return parser.parse_args()

def telemetry(sid, data):
    if data:
        # The current steering angle of the car
        steering_angle = float(data["steering_angle"])
        # The current throttle of the car, how hard to push peddle
        throttle = float(data["throttle"])
        # The current speed of the car
        speed = float(data["speed"])
        # The current image from the center camera of the car
        image = Image.open(BytesIO(base64.b64decode(data["image"])))
        try:
            image = np.asarray(image)       # from PIL image to numpy array
            image = utils.preprocess(image) # apply the preprocessing
            image = np.array([image])       # the model expects 4D array

            # predict the steering angle for the image
            steering_angle = float(model.predict(image, batch_size=1))
            # lower the throttle as the speed increases
            # if the speed is above the current speed limit, we are on a downhill.
            # make sure we slow down first and then go back to the original max speed.
            global speed_limit
            if speed > speed_limit:
                speed_limit = MIN_SPEED  # slow down
            else:
                speed_limit = MAX_SPEED
            throttle = 1.0 - steering_angle**2 - (speed/speed_limit)**2

            print('{} {} {}'.format(steering_angle, throttle, speed))
            send_control(steering_angle, throttle)
        except Exception as e:
            print(e)

        # save frame
        if args.image_folder != '':
            timestamp = datetime.utcnow().strftime('%Y_%m_%d_%H_%M_%S_%f')[:-3]
            image_filename = os.path.join(args.image_folder, timestamp)
            image.save('{}.jpg'.format(image_filename))
    else:

        sio.emit('manual', data={}, skip_sid=True)

def split_data(X, y, split_ratio=0.6):
    """
    Split data into training and testing.

    :param X: X
    :param y: y
    :param split_ratio: split ratio for train and test data
    """
    #split = int(X.shape[0] * split_ratio)
    #the input dimensions can be control through modifing the second axis.
    X_train= X[:, 1:4, :, :]
    y_train=y
    """
    Load validation data from .npy files.
    """
    X_test_R = np.load('data/X_validation_ALL.npy')
    X_test_R = X_test_R.astype(np.float32)
    X_test_R /= 255
    X_test_R=preprocess(X_test_R)
    X_test = X_test_R[:, 1:4, :, :]
    #y_test = int(y_test_R*0.2)

    y_test_R = np.load('data/y_validation.npy')
    y_test=y_test_R

    return X_train, y_train, X_test, y_test

def telemetry(sid, data):
    if data:
        # The current steering angle of the car
        steering_angle = float(data["steering_angle"])
        # The current throttle of the car
        throttle = float(data["throttle"])
        # The current speed of the car
        speed = float(data["speed"])
        # The current image from the center camera of the car
        image = Image.open(BytesIO(base64.b64decode(data["image"])))
        try:
            image = np.asarray(image)       # from PIL image to numpy array
            image = utils.preprocess(image) # apply the preprocessing
            image = np.array([image])       # the model expects 4D array

            # predict the steering angle for the image
            steering_angle = float(model.predict(image, batch_size=1))
            # lower the throttle as the speed increases
            # if the speed is above the current speed limit, we are on a downhill.
            # make sure we slow down first and then go back to the original max speed.
            global speed_limit
            if speed > speed_limit:
                speed_limit = MIN_SPEED  # slow down
            else:
                speed_limit = MAX_SPEED
            throttle = 1.0 - steering_angle**2 - (speed/speed_limit)**2

            print('{} {} {}'.format(steering_angle, throttle, speed))
            send_control(steering_angle, throttle)
        except Exception as e:
            print(e)

        # save frame
        if args.image_folder != '':
            timestamp = datetime.utcnow().strftime('%Y_%m_%d_%H_%M_%S_%f')[:-3]
            image_filename = os.path.join(args.image_folder, timestamp)
            image.save('{}.jpg'.format(image_filename))
    else:
        # NOTE: DON'T EDIT THIS.
        sio.emit('manual', data={}, skip_sid=True)

def test_preprocessing(self):
        s = (
            "???? >:( xd <3 :'D http://t.co/rlqo5xfbul www.google.com e-mail"
            " three-level-building I'm wouldn't @trump #bad"
            " 1.2 Hi, my name is: Jon!? Next sentence."
        )
        self.assertEqual(
            to_string(ut.preprocess(s)),
            (
                "e mail three level building i m wouldn t"
                " trump bad 12 hi my name is jon next sentence"
            )
        )

def get_linguistic_analysis(user, fromFile):
    tweets = []
    if fromFile:
        tweets = get_tweets_from_file(user)
    else:
        tweets = get_max_amount_tweets(user)
    tweets = utils.remove_retweets(tweets)
    norm = [
        utils.preprocess(tweet)
        for tweet in tweets if len(utils.preprocess(tweet))
        if not utils.is_retweet(tweet)
    ]
    print("\nLinguistic Analysis of ", user, "'s tweets\n")
    print(
        "Average word length: ",
        get_average_word_characters(norm),
        " characters"
    )
    print("Average syllables per word: ", get_average_word_syllables(norm))
    print(
        "Average sentence length: ",
        get_average_sentence_length(norm),
        " words"
    )
    print("Average tweet length: ", get_average_tweet_length(norm), " words")
    print(
        "Average question marks per tweet: ",
        get_average_question_marks(tweets)
    )
    print(
        "Average exclamation marks per tweet: ",
        get_average_exclamation_marks(tweets)
    )
    print("Average flesch grade level: ", get_average_flesch_grade_level(norm))
    print("\nMost frequent 25 keywords:")
    for tag,count in get_most_frequent_keywords(norm):
        print("{}: {}".format(tag, count))

def get_combined_keywords(tweets1, tweets2):
        # preprocess tweets
        preprocTweets1 = [
            ut.preprocess(tweet) for tweet in tweets1
            if len(ut.preprocess(tweet))
        ]
        preprocTweets2 = [
            ut.preprocess(tweet) for tweet in tweets2
            if len(ut.preprocess(tweet))
        ]

        # get combined list of top 25 most used keywords
        keywords1 = la.get_most_frequent_keywords(preprocTweets1)
        keywords2 = la.get_most_frequent_keywords(preprocTweets2)

        # get rid of tuples
        for i,tuple in enumerate(keywords1):
            keywords1[i] = tuple[0]
        for i,tuple in enumerate(keywords2):
            keywords2[i] = tuple[0]

        keywords1 = set(keywords1)
        keywords2 = set(keywords2)

        # combined keywords
        combinedKeywords = keywords1.union(keywords2)
        # return dictionary for counting keywords
        return {keyword:0 for keyword in combinedKeywords}