AOS-anomaly-detection

aos_write.py

import sys

from opensearchpy import OpenSearch, helpers

host = 'vpc-xxxx-cyz5i7xdfs3dffsdgerwerw.ap-northeast-1.es.amazonaws.com'
port = 443
auth = ('admin', 'password?') # For testing only. Don't store credentials in code.

# Create the client with SSL/TLS enabled, but hostname verification disabled.
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True, # enables gzip compression for request bodies
    http_auth = auth,
    use_ssl = True,
    verify_certs = True,
    ssl_assert_hostname = False,
    ssl_show_warn = False,
)


def create_index(client, index_name):
    if client.indices.exists(index_name):
        return
    index_body = {
        'settings': {
            'index': {
                'number_of_shards': 4
            }
        },
        "mappings": {
            "properties": {
                "db": {
                    "type": "float"
                },
                "device_id": {
                    "type": "text"
                },
                "event_time": {
                    "type": "date",
                    "format": "yyyy-MM-dd HH:mm:ss",
                    "ignore_malformed": True
                },
                "pm10": {
                    "type": "float"
                },
                "pm25": {
                    "type": "float"
                },
                "rh": {
                    "type": "float"
                },
                "temp": {
                    "type": "float"
                },
                "tsp": {
                    "type": "float"
                },
            }
        },
    }

    response = client.indices.create(index_name, body=index_body)
    return response

def bulk_ops(client, docs):
    response = helpers.bulk(client, docs, max_retries=3)
    print(response)


def main():
    f_path = sys.argv[1]

    create_index(client, 'iot_data')
    docs = []
    with open(f_path) as f:
        lines = f.readlines()
        for line in lines:
            vals = line.split(",")
            docs.append(
                {
                    "_index": "iot_data",
                    "device_id": vals[0],
                    "event_time": vals[1],
                    "pm2-5": vals[2],
                    "pm10": vals[3],
                    "tsp": vals[4],
                    "temp": vals[5],
                    "rh": vals[6],
                    "db": vals[7],
                }
            )
    bulk_ops(client, docs[1:])

if __name__ == "__main__":
    main()

sampledata.csv