update

.gitattributes

@@ -0,0 +1,6 @@
+model/190703-214543/loss.log filter=lfs diff=lfs merge=lfs -text
+model/190703-214543/model_95.ckpt.data-00000-of-00001 filter=lfs diff=lfs merge=lfs -text
+model/190703-214543/model_95.ckpt.index filter=lfs diff=lfs merge=lfs -text
+model/190703-214543/model_95.ckpt.meta filter=lfs diff=lfs merge=lfs -text
+model/190703-214543/checkpoint filter=lfs diff=lfs merge=lfs -text
+model/190703-214543/config.log filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -1,28 +1,20 @@
-FROM tensorflow/tensorflow
+FROM python:3.9
 
-# Create the environment:
-# COPY env.yml /app
-# RUN conda env create --name cs329s --file=env.yml
-# Make RUN commands use the new environment:
-# SHELL ["conda", "run", "-n", "cs329s", "/bin/bash", "-c"]
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
 
-RUN pip install tqdm obspy pandas 
-RUN pip install uvicorn fastapi kafka-python
+WORKDIR /app
 
-WORKDIR /opt
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
 
-# RUN wget https://github.com/AI4EPS/models/releases/download/phasenet-2018/model.tar && tar -xvf model.tar && rm model.tar
+# COPY phasenet /app/phasenet
+# COPY model /app/model
+# RUN wget https://github.com/AI4EPS/models/releases/download/PhaseNet-2018/model.tar && tar -xvf model.tar && rm model.tar
 
-# Copy files
-COPY phasenet /opt/phasenet
-# COPY model /opt/model
-RUN wget https://github.com/AI4EPS/models/releases/download/PhaseNet-2018/model.tar && tar -xvf model.tar && rm model.tar
+# EXPOSE 80
 
-# Expose API port
-EXPOSE 8000
-
-ENV PYTHONUNBUFFERED=1
-
-# Start API server
-#ENTRYPOINT ["conda", "run", "--no-capture-output", "-n", "cs329s", "uvicorn", "--app-dir", "phasenet", "app:app", "--reload", "--port", "8000", "--host", "0.0.0.0"]
-ENTRYPOINT ["uvicorn", "--app-dir", "phasenet", "app:app", "--reload", "--port", "7860", "--host", "0.0.0.0"]
+RUN git lfs pull
+COPY --chown=user . /app
+ENTRYPOINT ["uvicorn", "--app-dir", "phasenet", "app:app", "--reload","--host", "0.0.0.0",  "--port", "7860"]

phasenet/app.py

@@ -1,23 +1,21 @@
 import os
-from collections import defaultdict, namedtuple
+from collections import defaultdict
 from datetime import datetime, timedelta
-from json import dumps
-from typing import Any, AnyStr, Dict, List, NamedTuple, Union, Optional
+from typing import Any, AnyStr, Dict, List, NamedTuple, Optional, Union
 
 import numpy as np
-import requests
 import tensorflow as tf
 from fastapi import FastAPI, WebSocket
-from kafka import KafkaProducer
+from postprocess import extract_picks
 from pydantic import BaseModel
 from scipy.interpolate import interp1d
 
-from model import ModelConfig, UNet
-from postprocess import extract_picks
+from model import UNet
+
+PROJECT_ROOT = os.path.realpath(os.path.join(os.path.dirname(__file__), ".."))
 
 tf.compat.v1.disable_eager_execution()
 tf.compat.v1.logging.set_verbosity(tf.compat.v1.logging.ERROR)
-PROJECT_ROOT = os.path.realpath(os.path.join(os.path.dirname(__file__), ".."))
 JSONObject = Dict[AnyStr, Any]
 JSONArray = List[Any]
 JSONStructure = Union[JSONArray, JSONObject]
@@ -39,41 +37,6 @@ latest_check_point = tf.train.latest_checkpoint(f"{PROJECT_ROOT}/model/190703-21
 print(f"restoring model {latest_check_point}")
 saver.restore(sess, latest_check_point)
 
-# GAMMA API Endpoint
-GAMMA_API_URL = "http://gamma-api:8001"
-# GAMMA_API_URL = 'http://localhost:8001'
-# GAMMA_API_URL = "http://gamma.quakeflow.com"
-# GAMMA_API_URL = "http://127.0.0.1:8001"
-
-# Kafak producer
-use_kafka = False
-
-try:
-    print("Connecting to k8s kafka")
-    BROKER_URL = "quakeflow-kafka-headless:9092"
-    # BROKER_URL = "34.83.137.139:9094"
-    producer = KafkaProducer(
-        bootstrap_servers=[BROKER_URL],
-        key_serializer=lambda x: dumps(x).encode("utf-8"),
-        value_serializer=lambda x: dumps(x).encode("utf-8"),
-    )
-    use_kafka = True
-    print("k8s kafka connection success!")
-except BaseException:
-    print("k8s Kafka connection error")
-    try:
-        print("Connecting to local kafka")
-        producer = KafkaProducer(
-            bootstrap_servers=["localhost:9092"],
-            key_serializer=lambda x: dumps(x).encode("utf-8"),
-            value_serializer=lambda x: dumps(x).encode("utf-8"),
-        )
-        use_kafka = True
-        print("local kafka connection success!")
-    except BaseException:
-        print("local Kafka connection error")
-print(f"Kafka status: {use_kafka}")
-
 
 def normalize_batch(data, window=3000):
     """
@@ -208,10 +171,7 @@ def get_prediction(data, return_preds=False):
 
 
 class Data(BaseModel):
-    # id: Union[List[str], str]
-    # timestamp: Union[List[str], str]
-    # vec: Union[List[List[List[float]]], List[List[float]]]
-    id: List[str]
+    id: List[List[str]]
     timestamp: List[Union[str, float, datetime]]
     vec: Union[List[List[List[float]]], List[List[float]]]
 
@@ -239,6 +199,13 @@ def predict(data: Data):
     return picks
 
 
+@app.post("/predict_prob")
+def predict(data: Data):
+    picks, preds = get_prediction(data, True)
+
+    return picks, preds.tolist()
+
+
 @app.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket):
     await websocket.accept()
@@ -251,95 +218,11 @@ async def websocket_endpoint(websocket: WebSocket):
         print("PhaseNet Updating...")
 
 
-@app.post("/predict_prob")
-def predict(data: Data):
-    picks, preds = get_prediction(data, True)
-
-    return picks, preds.tolist()
-
-
-@app.post("/predict_phasenet2gamma")
-def predict(data: Data):
-    picks = get_prediction(data)
-
-    # if use_kafka:
-    #     print("Push picks to kafka...")
-    #     for pick in picks:
-    #         producer.send("phasenet_picks", key=pick["id"], value=pick)
-    try:
-        catalog = requests.post(
-            f"{GAMMA_API_URL}/predict", json={"picks": picks, "stations": data.stations, "config": data.config}
-        )
-        print(catalog.json()["catalog"])
-        return catalog.json()
-    except Exception as error:
-        print(error)
-
-    return {}
-
-
-@app.post("/predict_phasenet2gamma2ui")
-def predict(data: Data):
-    picks = get_prediction(data)
-
-    try:
-        catalog = requests.post(
-            f"{GAMMA_API_URL}/predict", json={"picks": picks, "stations": data.stations, "config": data.config}
-        )
-        print(catalog.json()["catalog"])
-        return catalog.json()
-    except Exception as error:
-        print(error)
-
-    if use_kafka:
-        print("Push picks to kafka...")
-        for pick in picks:
-            producer.send("phasenet_picks", key=pick["id"], value=pick)
-        print("Push waveform to kafka...")
-        for id, timestamp, vec in zip(data.id, data.timestamp, data.vec):
-            producer.send("waveform_phasenet", key=id, value={"timestamp": timestamp, "vec": vec, "dt": data.dt})
-
-    return {}
-
-
-@app.post("/predict_stream_phasenet2gamma")
-def predict(data: Data):
-    data = format_data(data)
-    # for i in range(len(data.id)):
-    #     plt.clf()
-    #     plt.subplot(311)
-    #     plt.plot(np.array(data.vec)[i, :, 0])
-    #     plt.subplot(312)
-    #     plt.plot(np.array(data.vec)[i, :, 1])
-    #     plt.subplot(313)
-    #     plt.plot(np.array(data.vec)[i, :, 2])
-    #     plt.savefig(f"{data.id[i]}.png")
-
-    picks = get_prediction(data)
-
-    return_value = {}
-    try:
-        catalog = requests.post(f"{GAMMA_API_URL}/predict_stream", json={"picks": picks})
-        print("GMMA:", catalog.json()["catalog"])
-        return_value = catalog.json()
-    except Exception as error:
-        print(error)
-
-    if use_kafka:
-        print("Push picks to kafka...")
-        for pick in picks:
-            producer.send("phasenet_picks", key=pick["id"], value=pick)
-        print("Push waveform to kafka...")
-        for id, timestamp, vec in zip(data.id, data.timestamp, data.vec):
-            producer.send("waveform_phasenet", key=id, value={"timestamp": timestamp, "vec": vec, "dt": data.dt})
-
-    return return_value
-
-
 @app.get("/healthz")
 def healthz():
     return {"status": "ok"}
 
+
 @app.get("/")
-def read_root():
-    return {"Hello": "PhaseNet!"}
+def greet_json():
+    return {"Hello": "PhaseNet!"}