NWPC消息平台：验证 ecFlow 命令消息

import pandas as pd
import numpy as np
from tqdm.notebook import tqdm

grep " /globalchartos/00" \
    /g1/u/nwp_pd/ecfworks/ecflow/login_b01.31071.ecf.log > "nwpc_pd.log"

from nwpc_workflow_log_model.log_record.ecflow import (
    EcflowLogParser, 
    EventType,
)

total_lines = 0
with open("nwpc_pd.log") as f:
    for line in f:
        total_lines += 1
        
progress = tqdm(total=total_lines)

records = []
with open("nwpc_pd.log") as f:
    for line in f:
        progress.update(1)
        parser = EcflowLogParser(
            options={
                EventType.Status: {
                    "debug": False,
                    "enable": False,
                },
                EventType.Client: {
                    "debug": False,
                    "enable": False,
                },
                EventType.Child: {
                    "debug": False,
                    "enable": True,
                },
                EventType.Server: {
                    "debug": False,
                    "enable": False,
                },
                EventType.Unknown: {
                    "debug": False,
                    "enable": False,
                }
            }
        )
        record = parser.parse(line)
        if record is None:
            continue
        records.append(record)

len(records)

log_times = [record.time for record in records]
log_dates = [record.date for record in records]
log_events = [record.event for record in records]
node_paths = [record.node_path for record in records]
log_df = pd.DataFrame({
    "date": log_dates,
    "time": log_times,
    "event": log_events,
    "node_path": node_paths,
})
log_df.head()

node_path = "/globalchartos/00/deterministic/base/066/wmc_vor_250_sep_066"
query_date = pd.to_datetime("2020-10-10")

from elasticsearch import Elasticsearch

elastic_server = "localhost:9200"
client = Elasticsearch(hosts=elastic_server)

results = client.search(
    body={
        "query": {
            "bool": {
                "filter": {
                    "term": {
                      "data.ecf_name.keyword": node_path
                    }
                }
            }
        }
    },
    index="ecflow-client-2020-10-08,ecflow-client-2020-10-09,ecflow-client-2020-10-10,ecflow-client-2020-10-11",
    size=30
)

data_list = []
for message in results["hits"]["hits"]:
    message_time = pd.to_datetime(message["_source"]["time"])
    data_list.append({
        "date": pd.to_datetime(message["_source"]["data"]["ecf_date"]),
        "time": message_time.round("s").time(),
        "event": message["_source"]["data"]["command"],
    })
message_df = pd.DataFrame(data_list)
message_df.head()

node_path_df = log_df[log_df.node_path==node_path]
day_df = node_path_df[node_path_df.date==query_date]
day_df[["date", "time", "event"]]

message_df[message_df.date==query_date]

from nwpc_workflow_model.ecflow import Bunch
from nwpc_workflow_model.visitor import (
    NodeVisitor,
    pre_order_travel,
)

node_paths = log_df.node_path.unique()
node_paths[:10]

bunch = Bunch()
for n in node_paths:
    bunch.add_node(n)

class TaskVisitor(NodeVisitor):
    def __init__(self):
        self.tasks = []
        
    def visit(self, node):
        if node.is_leaf():
            self.tasks.append(node)

visitor = TaskVisitor()
pre_order_travel(bunch, visitor)

len(visitor.tasks)

def get_from_log(
    log_df,
    node_path,
    date: pd.Timestamp,
):
    node_path_df = log_df[log_df.node_path==node_path]
    day_df = node_path_df[node_path_df.date==date]
    return day_df[["date", "time", "event"]]

import nwpc_message_tool.source.ecflow_client
from nwpc_message_tool.storage import EsMessageStorage

client = EsMessageStorage(
    hosts=[elastic_server],
    debug=False,
    show_progress=False
)

def get_from_message(
    node_path,
    date: pd.Timestamp,
):
    results = client.get_ecflow_client_messages(
        engine=nwpc_message_tool.source.ecflow_client,
        node_name=node_path,
        ecflow_host="XXXXX",
        ecflow_port="XXXXX",
        ecf_date=date,
        size=20,
    )
    data_list = []
    for message in results:
        message_time = message.time
        data_list.append({
            "date": message.ecf_date,
            "time": message_time.round("s").time(),
            "event": message.command,
        })
    message_df = pd.DataFrame(data_list)
    return message_df

node_path = "/globalchartos/00/deterministic/base/066/wmc_vor_250_sep_066"
query_date = pd.to_datetime("2020-10-07")

node_log_df = get_from_log(log_df, node_path, query_date)
node_log_df

node_message_df = get_from_message(node_path, query_date)
node_message_df

np.array_equal(
    node_log_df.event.values, 
    node_message_df.event.values
)

for node in tqdm(visitor.tasks):
    node_path = node.get_node_path()
    node_log_df = get_from_log_record(log_df, node_path, query_date)
    node_message_df = get_from_message(node_path, query_date)
    if not np.array_equal(node_log_df.event.values, node_message_df.event.values):
        print(node_path)

rmses = []
for node in tqdm(visitor.tasks):
    node_path = node.get_node_path()
    node_log_df = get_from_log_record(log_df, node_path, query_date)
    node_message_df = get_from_message(node_path, query_date)
    rmse = np.sqrt(
        np.mean(
            np.power(
                (
                    node_message_df.time.apply(lambda x: pd.to_timedelta(x.isoformat())) 
                    - node_log_df.time.apply(lambda x: pd.to_timedelta(x.isoformat())).reset_index(drop=True)
                ) / pd.Timedelta(seconds=1), 
                2
            )
        )
    )
    rmses.append(rmse)

np.max(rmses)