o
    vh                     @   s   d dl m Z mZ d dlmZmZ d dlZd dlmZmZ d dl	m
Z
 d dlmZ d dlmZmZ d dlmZ d	d
 dd
 dd
 dd
 dd
 dd
 dZe ZddedefddZdd ZdS )    )datetime	timedelta)	json_utilObjectIdN)PointPolygon)AIOKafkaConsumer)database)connected_clientsdevice_subscriptions)notify_nearby_subscribersc                 C   s   | |kS N abr   r   ;/var/www/html/moveengine/app/v1/consumers/kafka_consumer.py<lambda>       r   c                 C   s   | |kS r   r   r   r   r   r   r      r   c                 C   s   | |k S r   r   r   r   r   r   r      r   c                 C   s   | |kS r   r   r   r   r   r   r      r   c                 C   s   | |kS r   r   r   r   r   r   r      r   c                 C   s   | |kS r   r   r   r   r   r   r      r   )z==><z>=z<=z!=payloadalertsc                    s   |  d}ttD ]r}t |}||krq
zD|tj| tjdI dH  |rY|D ]/}| d|kr3q)|tjd| dd| d| d| d| d	d
tjdI dH  q)W q
 t	y| } zt
| t|d td| W Y d}~q
d}~ww dS )zV
    Push tracking + alerts only to clients subscribed to the payload's device_id
    	device_id)defaultNalertmessage typegeofencetriggered_at)r   r   r   
alert_typer   r    u   ❌ WebSocket error:)getlistr
   r   	send_textjsondumpsr   r   	Exceptiondiscardpopprint)r   r   r   wssubscribed_devicer   er   r   r   broadcast_to_clients  s<   


	
r.   c            *         s  t dddddd} |  I d H  t }t|}|d }|d }|d	 }|d
 }|d }|d }|d }	|d }
ztd | 2 z3 d H W }zt|j	d}W n t
yj } ztd| W Y d }~qAd }~ww |d}|ssqAt |d< d|d< |di dg }g }|jd|idgd}|rt |d  }|tddkr|dd| dt d}|d d| d!|jd"  d#t d}|||g |||g || t|d$|i}|D ]}|d%}|d&}|d't|i}|r|sqt|}t|d( d d) }||}|d*d+kr|s|r&|d't|ind }d}|rw|d,rw|d, D ]?}|d-} |d.}!|d/}"|!tvrQd0} n&|| }#z|#d u sht|! t|#t|"smd0}W  n
W q7   d0}Y  |r|r|d1nd| d2}$|d3|d4 |$|d d5}%||% ||% q|d't|i}&||d6|d7|d8|d9d:t d;}'|&r|&d<}(|&d=})|(d>kr|)r|	d't|)id?|'i n|(dkr|)r|
d't|)id?|'i t|I d H  n|	d@|id?|'i |
d@|id?|'i t |I d H  qA6 W | ! I d H  zt| W d S  t"y6   Y d S w | ! I d H  zt| W w  t"yN   Y w w )ANvehicle_trackingzstagingapi.movex.ai:9092ztracking-groupearliestT)bootstrap_serversgroup_idauto_offset_resetenable_auto_committracking_logs
alert_logs	geofencesgeofence_rulesgeofence_rule_mappingiot_devicesfleets	workforceu   ✅ Kafka consumer startedzutf-8u!   ❌ Error decoding Kafka message:r   received_atWr!   locationcoordinates)r=   )sort   )minutes
inactivityzDevice z inactive for over 30 minutes.)r   r   r   r    recoveryz back online after <   z min inactivity.assigned_entity_idgeofence_idgeofence_rule_id_idlocr   trigger_eventsexit
conditions	parameteroperatorvalueFalert_messagez exited geofence.zgeofence-exitname)r   r   r   r   r    speedheadingfuel	timestamp)r@   rU   rV   rW   rX   )last_locationlast_updatedassociated_entity_typeassociated_entity_idvehiclez$setdevices)#r   startr	   get_mongo_dbnextr*   r%   loadsrR   decoder'   r"   r   utcnowfind_oner   secondsinsert_manyextend
insert_oner#   findr   r   r   contains	OPERATORSfloatappend
update_oner   update_manybroadcast_tracking_updatestopStopIteration)*consumer	mongo_genmongor5   r6   r7   r8   geofence_mappingsr^   r;   r<   msgdatar-   r   r@   r   
last_entrygapinactivity_alertrecovery_alertmappingsmappinggeo_idrule_idr   pointpolygoninsideruleconditions_passedcondparamopexpectedactual	alert_msgr   devicelast_location_update
assoc_typeassoc_idr   r   r   consume_tracking_data  s   











"



qr   r   )r   r   bsonr   r   r%   shapely.geometryr   r   aiokafkar   app.dbr	   app.utils.connectionsr
   r   app.v1.sockets.tracking_datar   rl   setdictr#   r.   r   r   r   r   r   <module>   s0    	     t 4