callback used to stream events to a server requires the requests library events are sent to root publishepochend by default calls are http post with a data argument which is a jsonencoded dictionary of event data if sendasjsontrue the content type of the request will be applicationjson otherwise the serialized json will be sent within a form args root string root url of the target server path string path relative to root to which the events will be sent field string json field under which the data will be stored the field is used only if the payload is sent within a form i e sendasjson is set to false headers dictionary optional custom http headers sendasjson boolean whether the request should be sent as applicationjson np ndarray and np generic are not scalar types therefore we must unwrap their scalar values and pass to the jsonserializable dict send callback used to stream events to a server requires the requests library events are sent to root publish epoch end by default calls are http post with a data argument which is a json encoded dictionary of event data if send_as_json true the content type of the request will be application json otherwise the serialized json will be sent within a form args root string root url of the target server path string path relative to root to which the events will be sent field string json field under which the data will be stored the field is used only if the payload is sent within a form i e send_as_json is set to false headers dictionary optional custom http headers send_as_json boolean whether the request should be sent as application json np ndarray and np generic are not scalar types therefore we must unwrap their scalar values and pass to the json serializable dict send
import json
import warnings

import numpy as np

from keras.api_export import keras_export
from keras.callbacks.callback import Callback

try:
    import requests
except ImportError:
    requests = None


@keras_export("keras.callbacks.RemoteMonitor")
class RemoteMonitor(Callback):
    

    def __init__(
        self,
        root="http://localhost:9000",
        path="/publish/epoch/end/",
        field="data",
        headers=None,
        send_as_json=False,
    ):
        super().__init__()

        self.root = root
        self.path = path
        self.field = field
        self.headers = headers
        self.send_as_json = send_as_json

    def on_epoch_end(self, epoch, logs=None):
        if requests is None:
            raise ImportError("RemoteMonitor requires the `requests` library.")
        logs = logs or {}
        send = {}
        send["epoch"] = epoch
        for k, v in logs.items():
            
            
            
            if isinstance(v, (np.ndarray, np.generic)):
                send[k] = v.item()
            else:
                send[k] = v
        try:
            if self.send_as_json:
                requests.post(
                    self.root + self.path, json=send, headers=self.headers
                )
            else:
                requests.post(
                    self.root + self.path,
                    {self.field: json.dumps(send)},
                    headers=self.headers,
                )
        except requests.exceptions.RequestException:
            warnings.warn(
                f"Could not reach RemoteMonitor root server at {self.root}",
                stacklevel=2,
            )