ExternalVendorCode/sondehub-infra
1
0
Fork 0
mirror of https://github.com/projecthorus/sondehub-infra.git synced 2025-02-20 17:12:50 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

add listener endpoints for amateur

Browse Source
This commit is contained in:
xss 2022-04-10 15:18:28 +10:00
parent 2bbcefb203
commit 7080e6c70d
3 changed files with 189 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
ham_query.tf
View File

@ -97,4 +97,53 @@ resource "aws_apigatewayv2_integration" "ham_telem" {
integration_uri = aws_lambda_function.ham_telem.arn
timeout_milliseconds = 30000
payload_format_version = "2.0"
}
resource "aws_lambda_function" "ham_get_listener_telemetry" {
function_name = "ham_get_listener_telemetry"
handler = "query_ham.get_listener_telemetry"
s3_bucket = aws_s3_bucket_object.lambda.bucket
s3_key = aws_s3_bucket_object.lambda.key
source_code_hash = data.archive_file.lambda.output_base64sha256
publish = true
memory_size = 256
role = aws_iam_role.basic_lambda_role.arn
runtime = "python3.9"
timeout = 30
architectures = ["arm64"]
environment {
variables = {
"ES" = "es.${local.domain_name}"
}
}
tags = {
Name = "ham_get_listener_telemetry"
}
}
resource "aws_lambda_permission" "ham_get_listener_telemetry" {
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.ham_get_listener_telemetry.arn
principal = "apigateway.amazonaws.com"
source_arn = "arn:aws:execute-api:us-east-1:${data.aws_caller_identity.current.account_id}:${aws_apigatewayv2_api.main.id}/*/*/amateur/listeners/telemetry"
}
resource "aws_apigatewayv2_route" "ham_get_listener_telemetry" {
api_id = aws_apigatewayv2_api.main.id
api_key_required = false
authorization_type = "NONE"
route_key = "GET /amateur/listeners/telemetry"
target = "integrations/${aws_apigatewayv2_integration.ham_get_listener_telemetry.id}"
}
resource "aws_apigatewayv2_integration" "ham_get_listener_telemetry" {
api_id = aws_apigatewayv2_api.main.id
connection_type = "INTERNET"
integration_method = "POST"
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_function.ham_get_listener_telemetry.arn
timeout_milliseconds = 30000
payload_format_version = "2.0"
}

134
lambda/query_ham/__init__.py
View File

@ -85,11 +85,11 @@ def get(event, context):
def get_telem(event, context):
durations = { # ideally we shouldn't need to predefine these, but it's a shit load of data and we don't need want to overload ES
"3d": (259200, 1200), # 3d, 20m
"1d": (86400, 600), # 1d, 10m
"12h": (43200, 600), # 1d, 10m
"6h": (21600, 1), # 6h, 1m
"3h": (10800, 1), # 3h, 10s
"3d": (259200, 1),
"1d": (86400, 1),
"12h": (43200, 1),
"6h": (21600, 1),
"3h": (10800, 1),
"1h": (3600, 1),
"30m": (1800, 1),
"1m": (60, 1),
@ -213,3 +213,127 @@ def get_telem(event, context):
}
def get_listener_telemetry(event, context):
durations = { # ideally we shouldn't need to predefine these, but it's a shit load of data and we don't need want to overload ES
"3d": (259200, 1), # 3d, 20m
"1d": (86400, 1), # 1d, 10m
"12h": (43200, 1), # 1d, 10m
"6h": (21600, 1), # 6h, 1m
"3h": (10800, 1), # 3h, 10s
"1h": (3600, 1),
"30m": (1800, 1),
"1m": (60, 1),
"15s": (15, 1),
"0": (0, 1)
}
duration_query = "3h"
requested_time = datetime.now(timezone.utc)
if (
"queryStringParameters" in event
and "duration" in event["queryStringParameters"]
):
if event["queryStringParameters"]["duration"] in durations:
duration_query = event["queryStringParameters"]["duration"]
else:
return f"Duration must be either {', '.join(durations.keys())}"
if (
"queryStringParameters" in event
and "datetime" in event["queryStringParameters"]
):
requested_time = datetime.fromisoformat(
event["queryStringParameters"]["datetime"].replace("Z", "+00:00")
)
(duration, interval) = durations[duration_query]
if "queryStringParameters" in event and "uploader_callsign" in event["queryStringParameters"]:
interval = 1
lt = requested_time
gte = requested_time - timedelta(0, duration)
path = "ham-listeners-*/_search"
payload = {
"size": 0,
"timeout": "30s",
"aggs": {
"2": {
"terms": {
"field": "uploader_callsign.keyword",
"order": {"_key": "desc"},
"size": 10000,
},
"aggs": {
"3": {
"date_histogram": {
"field": "ts",
"fixed_interval": f"{str(interval)}s",
"min_doc_count": 1,
},
"aggs": {
"1": {
"top_hits": {
# "docvalue_fields": [
# {"field": "position"},
# {"field": "alt"},
# {"field": "datetime"},
# ],
# "_source": "position",
"size": 1,
"sort": [{"ts": {"order": "desc"}}],
}
}
},
}
},
}
},
"query": {
"bool": {
"filter": [
{"match_all": {}},
{"exists": { "field": "uploader_position"}},
{
"range": {
"ts": {"gte": gte.isoformat(), "lt": lt.isoformat()}
}
},
]
}
},
}
if "queryStringParameters" in event:
if "uploader_callsign" in event["queryStringParameters"]:
payload["query"]["bool"]["filter"].append(
{
"match_phrase": {
"uploader_callsign": str(event["queryStringParameters"]["uploader_callsign"])
}
}
)
results = es.request(json.dumps(payload), path, "POST")
output = {
sonde["key"]: {
data["key_as_string"]: data["1"]["hits"]["hits"][0]["_source"]
for data in sonde["3"]["buckets"]
}
for sonde in results["aggregations"]["2"]["buckets"]
}
compressed = BytesIO()
with gzip.GzipFile(fileobj=compressed, mode='w') as f:
json_response = json.dumps(output)
f.write(json_response.encode('utf-8'))
gzippedResponse = compressed.getvalue()
return {
"body": base64.b64encode(gzippedResponse).decode(),
"isBase64Encoded": True,
"statusCode": 200,
"headers": {
"Content-Encoding": "gzip",
"content-type": "application/json"
}
}

14
lambda/query_ham/__main__.py
View File

@ -3,11 +3,19 @@ import base64
import zlib
response = get_telem(
# response = get_telem(
# {
# "queryStringParameters":{
# # "payload_callsign": "HORUS-V2",
# "duration": "3d"
# }
# }, {})
response = get_listener_telemetry(
{
"queryStringParameters":{
"payload_callsign": "HORUS-V2",
"duration": "3d"
# "payload_callsign": "HORUS-V2",
"duration": "3h"
}
}, {})
compressed = base64.b64decode(response['body'])