rename csquares based detection

parent 1453837a
......@@ -9,17 +9,16 @@ import functools
import concurrent.futures
import glob
import logging
import math
import pickle
import sys
from sesamelib.geometry import IndexedGeometry, distance
from sesamelib.faust import BaseDynamicMessage
import faust
logging.basicConfig(level=logging.INFO)
APP_NAME = "ca1-015-csquare"
APP_NAME = "ca1-015-csquares-%s"
if __name__ == "__main__":
parser = argparse.ArgumentParser("Mutate the ais dynamic signal")
......@@ -27,9 +26,14 @@ if __name__ == "__main__":
help="Kafka bootstrap servers",
action="append",
required=True)
parser.add_argument("-c", "--cell_size", help="cell_size to use. Will look in cell_size subdirectory to load the geometries",
type=float, default=0.0)
args = parser.parse_args()
files = glob.glob("data/*.pickle")
logging.info(args)
files = glob.glob("/data/%s/*.pickle" % args.cell_size)
# will hold all the indexed geometries
geometries = []
for f in files:
......@@ -37,12 +41,14 @@ if __name__ == "__main__":
geometry = pickle.load(f)
geometries.append(geometry)
logging.info("Loaded %s geometries" % len(geometries))
loop = asyncio.get_event_loop()
pool = concurrent.futures.ThreadPoolExecutor()
app = faust.App(
APP_NAME,
APP_NAME % args.cell_size,
loop=loop,
broker="kafka://{}".format(",".join(args.bootstrap_servers))
)
......@@ -50,7 +56,7 @@ if __name__ == "__main__":
topic = app.topic("ais.dynamic",
key_type=str,
value_type=BaseDynamicMessage)
out_topic = app.topic(APP_NAME)
out_topic = app.topic(APP_NAME % args.cell_size)
pmsgs = {}
......@@ -67,6 +73,7 @@ if __name__ == "__main__":
y,
geometry))
if d <= 0.0:
print(msg)
await out_topic.send(key=str(msg.mmsi), value=msg)
app.finalize()
......
......@@ -33,7 +33,7 @@ services:
image: $NAMESPACE/ingestion:$TARGET_REF
command: --bootstrap_servers kafka:9092 "/bigdata/groups/sesame/orbcomm/aivdm/2017/03/01/*.nm4"
container_name: ingestion
# don't restart otherwise messages will be indexed several times...
# don':t restart otherwise messages will be indexed several times...
restart: on-failure
build:
context: ingestion
......@@ -41,7 +41,7 @@ services:
TARGET_REF: $TARGET_REF
NAMESPACE: $NAMESPACE
volumes:
- ../bigdata:/bigdata
- ../bigdata/:/bigdata
depends_on:
- base_gdal
......@@ -88,16 +88,18 @@ services:
depends_on:
- base_gdal
ca1-015-csquare:
image: $NAMESPACE/ca1-015-csquare:$TARGET_REF
command: --bootstrap_servers kafka:9092
container_name: ca1-015-csquare
ca1-015-csquares:
image: $NAMESPACE/ca1-015-csquares:$TARGET_REF
command: --bootstrap_servers kafka:9092 --cell_size 0.25
container_name: ca1-015-csquares
restart: always
build:
context: ca1-015-csquare
context: ca1-015-csquares
args:
TARGET_REF: $TARGET_REF
NAMESPACE: $NAMESPACE
volumes:
- ../bigdata/groups/sesame/csquares:/data
depends_on:
- base_gdal
############################
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment