<commit_before>import json
import os
import sys
import time
from typing import Tuple

import datetime

from plenum.common.constants import OP_FIELD_NAME, PREPREPARE, BATCH, \
    KeyValueStorageType, CLIENT_STACK_SUFFIX
from plenum.common.types import f
from plenum.common.util import get_utc_epoch
from plenum.recorder.combined_recorder import CombinedRecorder
from plenum.recorder.recorder import Recorder
from storage.helper import initKeyValueStorageIntKeys


def to_bytes(v):
    if not isinstance(v, bytes):
        return v.encode()


def get_recorders_from_node_data_dir(node_data_dir, node_name) -> Tuple[Recorder, Recorder]:
    rec_path = os.path.join(node_data_dir, node_name, 'recorder')
    client_stack_name = node_name + CLIENT_STACK_SUFFIX
    client_rec_kv_store = initKeyValueStorageIntKeys(KeyValueStorageType.Rocksdb,
                                                     rec_path, client_stack_name)
    node_rec_kv_store = initKeyValueStorageIntKeys(
        KeyValueStorageType.Rocksdb, rec_path, node_name)

    return Recorder(node_rec_kv_store, skip_metadata_write=True), \
        Recorder(client_rec_kv_store, skip_metadata_write=True)


def patch_sent_prepreapres(replaying_node, node_recorder):
    sent_pps = {}

    def add_preprepare(msg):
        inst_id, v, p = msg[f.INST_ID.nm], msg[f.VIEW_NO.nm], msg[
            f.PP_SEQ_NO.nm]
        if inst_id not in sent_pps:
            sent_pps[inst_id] = {}
        sent_pps[inst_id][v, p] = [msg[f.PP_TIME.nm],
                                   [l for l in
                                    msg[f.REQ_IDR.nm]],
                                   msg[f.DISCARDED.nm],
                                   ]

    msg_count = 0
    min_msg_time = sys.maxsize
    max_msg_time = -1

    for k, v in node_recorder.store.iterator(include_value=True):
        max_msg_time = max(max_msg_time, int(k))
        min_msg_time = min(min_msg_time, int(k))
        parsed = Recorder.get_parsed(v.decode())
        msg_count += len(parsed)

        outgoings = Recorder.filter_outgoing(parsed)
        if not outgoings:
            continue
        for out in outgoings:
            try:
                msg = json.loads(out[0])
                if isinstance(msg, dict) and OP_FIELD_NAME in msg:
                    op_name = msg[OP_FIELD_NAME]
                    if op_name == PREPREPARE:
                        add_preprepare(msg)
                    elif op_name == BATCH:
                        for m in msg['messages']:
                            try:
                                m = json.loads(m)
                                if m[OP_FIELD_NAME] == PREPREPARE:
                                    add_preprepare(m)
                            except json.JSONDecodeError:
                                continue
                    else:
                        continue
            except json.JSONDecodeError:
                continue

    for r in replaying_node.replicas:
        r.sent_pps = sent_pps.pop(r.instId, {})

    replaying_node.sent_pps = sent_pps
    replaying_node.replay_msg_count = msg_count

    run_time = max_msg_time-min_msg_time
    run_time = int(run_time/Recorder.TIME_FACTOR)
    print ("Aprox run time: {}".format(str(datetime.timedelta(seconds=run_time))))


def get_combined_recorder(replaying_node, node_recorder, client_recorder):
    kv_store = initKeyValueStorageIntKeys(KeyValueStorageType.Rocksdb,
                                          replaying_node.dataLocation,
                                          'combined_recorder')
    cr = CombinedRecorder(kv_store)
    # Always add node recorder first and then client recorder
    cr.add_recorders(node_recorder, client_recorder)
    cr.combine_recorders()
    return cr


def prepare_node_for_replay_and_replay(looper, replaying_node,
                                       node_recorder, client_recorder,
                                       start_times):
    cr = get_combined_recorder(replaying_node, node_recorder, client_recorder)
    cr.start_times = start_times
    patch_replaying_node(replaying_node, node_recorder, start_times)
    return replay_patched_node(looper, replaying_node, node_recorder, cr)


def patch_replaying_node(replaying_node, node_recorder, start_times):
    patch_sent_prepreapres(replaying_node, node_recorder)
    patch_replaying_node_for_time(replaying_node, start_times)


def patch_replaying_node_for_time(replaying_node, start_times):
    node_1st_start_time = start_times[0][0]
    replaying_node._time_diff = get_utc_epoch() - node_1st_start_time<commit_msg>fix static validation<commit_after>import json
import os
import sys
import time
from typing import Tuple

import datetime

from plenum.common.constants import OP_FIELD_NAME, PREPREPARE, BATCH, \
    KeyValueStorageType, CLIENT_STACK_SUFFIX
from plenum.common.types import f
from plenum.common.util import get_utc_epoch
from plenum.recorder.combined_recorder import CombinedRecorder
from plenum.recorder.recorder import Recorder
from storage.helper import initKeyValueStorageIntKeys


def to_bytes(v):
    if not isinstance(v, bytes):
        return v.encode()


def get_recorders_from_node_data_dir(node_data_dir, node_name) -> Tuple[Recorder, Recorder]:
    rec_path = os.path.join(node_data_dir, node_name, 'recorder')
    client_stack_name = node_name + CLIENT_STACK_SUFFIX
    client_rec_kv_store = initKeyValueStorageIntKeys(KeyValueStorageType.Rocksdb,
                                                     rec_path, client_stack_name)
    node_rec_kv_store = initKeyValueStorageIntKeys(
        KeyValueStorageType.Rocksdb, rec_path, node_name)

    return Recorder(node_rec_kv_store, skip_metadata_write=True), \
        Recorder(client_rec_kv_store, skip_metadata_write=True)


def patch_sent_prepreapres(replaying_node, node_recorder):
    sent_pps = {}

    def add_preprepare(msg):
        inst_id, v, p = msg[f.INST_ID.nm], msg[f.VIEW_NO.nm], msg[
            f.PP_SEQ_NO.nm]
        if inst_id not in sent_pps:
            sent_pps[inst_id] = {}
        sent_pps[inst_id][v, p] = [msg[f.PP_TIME.nm],
                                   [l for l in
                                    msg[f.REQ_IDR.nm]],
                                   msg[f.DISCARDED.nm],
                                   ]

    msg_count = 0
    min_msg_time = sys.maxsize
    max_msg_time = -1

    for k, v in node_recorder.store.iterator(include_value=True):
        max_msg_time = max(max_msg_time, int(k))
        min_msg_time = min(min_msg_time, int(k))
        parsed = Recorder.get_parsed(v.decode())
        msg_count += len(parsed)

        outgoings = Recorder.filter_outgoing(parsed)
        if not outgoings:
            continue
        for out in outgoings:
            try:
                msg = json.loads(out[0])
                if isinstance(msg, dict) and OP_FIELD_NAME in msg:
                    op_name = msg[OP_FIELD_NAME]
                    if op_name == PREPREPARE:
                        add_preprepare(msg)
                    elif op_name == BATCH:
                        for m in msg['messages']:
                            try:
                                m = json.loads(m)
                                if m[OP_FIELD_NAME] == PREPREPARE:
                                    add_preprepare(m)
                            except json.JSONDecodeError:
                                continue
                    else:
                        continue
            except json.JSONDecodeError:
                continue

    for r in replaying_node.replicas:
        r.sent_pps = sent_pps.pop(r.instId, {})

    replaying_node.sent_pps = sent_pps
    replaying_node.replay_msg_count = msg_count

    run_time = max_msg_time - min_msg_time
    run_time = int(run_time / Recorder.TIME_FACTOR)
    print("Aprox run time: {}".format(str(datetime.timedelta(seconds=run_time))))


def get_combined_recorder(replaying_node, node_recorder, client_recorder):
    kv_store = initKeyValueStorageIntKeys(KeyValueStorageType.Rocksdb,
                                          replaying_node.dataLocation,
                                          'combined_recorder')
    cr = CombinedRecorder(kv_store)
    # Always add node recorder first and then client recorder
    cr.add_recorders(node_recorder, client_recorder)
    cr.combine_recorders()
    return cr


def prepare_node_for_replay_and_replay(looper, replaying_node,
                                       node_recorder, client_recorder,
                                       start_times):
    cr = get_combined_recorder(replaying_node, node_recorder, client_recorder)
    cr.start_times = start_times
    patch_replaying_node(replaying_node, node_recorder, start_times)
    return replay_patched_node(looper, replaying_node, node_recorder, cr)


def patch_replaying_node(replaying_node, node_recorder, start_times):
    patch_sent_prepreapres(replaying_node, node_recorder)
    patch_replaying_node_for_time(replaying_node, start_times)


def patch_replaying_node_for_time(replaying_node, start_times):
    node_1st_start_time = start_times[0][0]
    replaying_node._time_diff = get_utc_epoch() - node_1st_start_time
{'fraction_non_alphanumeric': 0.07095295273441764, 'fraction_numerical': 0.0015282174435105338, 'mean_word_length': 2.205738278516445, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2945399', 'n_tokens_mistral': 2897, 'n_tokens_neox': 2768, 'n_words': 506}