Skip to content

Commit cb410d2

Browse files
committed
Attempt to fix the relays projection as one-off for now.
1 parent 110ed46 commit cb410d2

1 file changed

Lines changed: 62 additions & 0 deletions

File tree

opensensor/collection_apis.py

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -389,6 +389,58 @@ def get_vpd_pipeline(
389389
return pipeline
390390

391391

392+
def get_relay_board_pipeline(
393+
device_ids: List[str],
394+
device_name: str,
395+
start_date: datetime,
396+
end_date: datetime,
397+
resolution: int,
398+
):
399+
sampling_interval = timedelta(minutes=resolution)
400+
if start_date is None:
401+
start_date = datetime.utcnow() - timedelta(days=100)
402+
if end_date is None:
403+
end_date = datetime.utcnow()
404+
match_clause = get_initial_match_clause(device_ids, device_name, start_date, end_date)
405+
406+
# The MongoDB aggregation pipeline for Relay Board data
407+
pipeline = [
408+
{"$match": match_clause},
409+
{
410+
"$addFields": {
411+
"group": {
412+
"$floor": {
413+
"$divide": [
414+
{"$subtract": ["$timestamp", start_date]},
415+
sampling_interval.total_seconds() * 1000,
416+
]
417+
}
418+
}
419+
}
420+
},
421+
{
422+
"$group": {
423+
"_id": "$group",
424+
"doc": {"$first": "$$ROOT"},
425+
}
426+
},
427+
{
428+
"$replaceRoot": {
429+
"newRoot": "$doc",
430+
}
431+
},
432+
{
433+
"$project": {
434+
"_id": False,
435+
"timestamp": "$timestamp",
436+
"relays": "$relays",
437+
}
438+
},
439+
{"$sort": {"timestamp": 1}},
440+
]
441+
return pipeline
442+
443+
392444
def get_uniform_sample_pipeline(
393445
response_model: Type[T],
394446
device_ids: List[str],
@@ -478,6 +530,14 @@ def sample_and_paginate_collection(
478530
end_date,
479531
resolution,
480532
)
533+
elif response_model is RelayBoard:
534+
pipeline = get_relay_board_pipeline(
535+
device_ids,
536+
target_device_name,
537+
start_date,
538+
end_date,
539+
resolution,
540+
)
481541
else:
482542
pipeline = get_uniform_sample_pipeline(
483543
response_model,
@@ -507,6 +567,8 @@ def sample_and_paginate_collection(
507567
# If the response model is VPD, you already have VPD-related data from the pipeline.
508568
# So, you can directly use it to create the response model instances.
509569
data = [VPD(**item) for item in raw_data]
570+
elif response_model is RelayBoard:
571+
data = [RelayBoard(**item) for item in raw_data]
510572
else:
511573
data = [create_model_instance(response_model, item, unit) for item in raw_data]
512574

0 commit comments

Comments
 (0)