python-fastapi-mm-zhcs-yj-api/routers/api/videoResource/location/__init__.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from fastapi import APIRouter, Request, Depends, Query, HTTPException, status
from common.security import valid_access_token
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from sqlalchemy.sql import func
from sqlalchemy.future import select
from common.auth_user import *
from pydantic import BaseModel
from database import get_db
from typing import List
from models import *
from utils import *
from utils.ry_system_util import *
from utils.video_util import *
from collections import defaultdict
import traceback
from concurrent.futures import ThreadPoolExecutor, as_completed
from multiprocessing import Pool, cpu_count
import json
import time
import math

router = APIRouter()



@router.get('/video/{video_code}')
async def get_video_url_by_id(
    video_code:str,
    db: Session = Depends(get_db),
    body=Depends(remove_xss_json),
    user_id=Depends(valid_access_token)
):
    # 大屏左下角视频及更多视频
    try:
        query = db.query(TPVideoInfo)
        query = query.filter(TPVideoInfo.gbIndexCode==video_code)
        videoIds = user_id_get_user_videoIds(db, user_id)
        video_list = [i.video_code_int for i in videoIds]
        video = query.first()
        tag_list =get_video_tag_list(db,video.gbIndexCode)
        tag = []
        tag_lable = []
        for info in tag_list:
            tag_info = get_dict_data_info(db, info.dict_type, info.dict_value)
            if tag_info:
                if tag_info.dict_label not in tag_lable and tag_info.dict_label!='全量视频':
                    tag.append({"id": info.id,
                                "video_code": video.gbIndexCode,
                                "dict_type": info.dict_type,
                                "dict_value": info.dict_value,
                                "dict_label": tag_info.dict_label,
                                "dict_code": tag_info.dict_code})
                    tag_lable.append(tag_info.dict_label)
        data={
            "name":video.name,
            "isUserVideos":video.gbIndexCode in video_list,
            "video_code": video.gbIndexCode,
            "isTag" : len(tag_list)>0,
            "tag" : tag,
            "tagLabels" : "、".join(tag_lable),
            "status":video.status,
            "longitude":video.longitude,
            "latitude":video.latitude,
            "statusName":video.statusName,
            "regionPath":video.regionPath,
            "installPlace":video.installPlace,
            "cameraTypeName":video.cameraTypeName,
            "cameraType":video.cameraType
        }
        return {
            "code": 200,
            "msg": "操作成功",
            "data": data}
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))


@router.get("/videos")
async def get_videos(
        zoom_level: float = Query(..., description="Zoom level for clustering"),
        latitude_min: float = Query(..., description="Minimum latitude"),
        latitude_max: float = Query(..., description="Maximum latitude"),
        longitude_min: float = Query(..., description="Minimum longitude"),
        longitude_max: float = Query(..., description="Maximum longitude"),
        dict_value: str = Query(None),
        db:  Session = Depends(get_db)
):
    try:
        # 根据缩放级别动态调整分组粒度
        distance_threshold = 1000 / (2 ** zoom_level)  # 例如：每缩放一级，距离阈值减半
        que = True
        print(time.time())
        if dict_value:
            tag_info = get_dict_data_info(db, 'video_type', dict_value)
            if tag_info:
                if tag_info.dict_label!='全量视频':
                    videolist = [i.video_code for i in tag_get_video_tag_list(db,dict_value)]
                    que =TPVideoInfo.gbIndexCode.in_(videolist)
        # 查询分组
        print("1",time.time())
        query = (
            select(
                TPVideoInfo.cameraIndexCode,
                TPVideoInfo.gbIndexCode,
                TPVideoInfo.pixel,
                TPVideoInfo.cameraType,
                TPVideoInfo.cameraTypeName,
                TPVideoInfo.installPlace,
                TPVideoInfo.status,
                TPVideoInfo.statusName,
                TPVideoInfo.latitude,
                TPVideoInfo.longitude,
                TPVideoInfo.name,
                TPVideoInfo.unitIndexCode,
                func.ST_AsText(TPVideoInfo.location).label("location")
            )
            .select_from(TPVideoInfo).where(
                and_(
                    TPVideoInfo.latitude >= latitude_min,
                    TPVideoInfo.latitude <= latitude_max,
                    TPVideoInfo.longitude >= longitude_min,
                    TPVideoInfo.longitude <= longitude_max,
                    TPVideoInfo.longitude>0,
                    TPVideoInfo.latitude>0,que
                )
            )
            .order_by(TPVideoInfo.cameraIndexCode)
        )

        result = db.execute(query)
        print("2",time.time())
        videos = result.fetchall()
        print("3",time.time())

        # 动态分组逻辑
        # groups = {}
        groups = group_videos(videos, distance_threshold)
        # for video in videos:
        #     grouped = False
        #     for group_id, group in list(groups.items()):
        #         for v in group["videos"]:
        #             distance = calculate_distance(video, v)
        #             if distance < distance_threshold:
        #                 groups[group_id]["videos"].append(video)
        #                 groups[group_id]["count"] += 1
        #                 grouped = True
        #                 break
        #         if grouped:
        #             break
        #     if not grouped:
        #         group_id = video.cameraIndexCode
        #         groups[group_id] = {"count": 1, "videos": [video]}

        print("4",time.time())

        return {"code": 200,
                "msg": "操作成功",
                "data": groups}
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
def calculate_grid_size(distance_threshold):
    # 假设地球半径为6371公里，将距离阈值转换为经纬度的差值
    # 这里假设纬度变化对距离的影响较小，仅根据经度计算网格大小
    earth_radius = 6371  # 地球半径，单位为公里
    grid_size = distance_threshold / earth_radius
    return grid_size

def get_grid_key(latitude, longitude, grid_size):
    # 根据经纬度和网格大小计算网格键
    return (math.floor(latitude / grid_size), math.floor(longitude / grid_size))

def calculate_distance(video1, video2):
    # 使用 Haversine 公式计算两点之间的距离
    from math import radians, sin, cos, sqrt, atan2
    R = 6371  # 地球半径（公里）
    lat1, lon1 = radians(video1.latitude), radians(video1.longitude)
    lat2, lon2 = radians(video2.latitude), radians(video2.longitude)
    dlat = lat2 - lat1
    dlon = lon2 - lon1
    a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
    c = 2 * atan2(sqrt(a), sqrt(1 - a))
    return R * c

def group_videos(videos, distance_threshold):
    grid_size = calculate_grid_size(distance_threshold)
    grid = defaultdict(lambda:{"count":0}) #,"list":[]
    groups = []

    for video in videos:
        grid_key = get_grid_key(video.latitude, video.longitude, grid_size)
        grid['%s-%s'%grid_key]['count']+=1
        grid['%s-%s'%grid_key]['latitude'] = (grid_key[0] + 0.5) * grid_size
        grid['%s-%s'%grid_key]['longitude'] = (grid_key[1] + 0.5) * grid_size
        # grid['%s-%s'%grid_key]['list'].append(video)
        if grid['%s-%s'%grid_key]['count']>1:
            grid['%s-%s' % grid_key]['type'] ='2'
        else:grid['%s-%s'%grid_key]['type'] ='1'
        groups = list(grid.values())
        # for group_id, group in list(grid.items()):
        #     groups.append(group)
    # #使用多线程计算距离
    # def process_video(video, grid_videos, groups, distance_threshold):
    #     grouped = False
    #     for group_id, group in list(groups.items()):
    #         for v in group["videos"]:
    #             if calculate_distance(video, v) < distance_threshold:
    #                 groups[group_id]["videos"].append(video)
    #                 groups[group_id]["count"] += 1
    #                 grouped = True
    #                 break
    #         if grouped:
    #             break
    #     if not grouped:
    #         group_id = video.cameraIndexCode
    #         groups[group_id] = {"count": 1, "videos": [video]}
    #
    # with ThreadPoolExecutor() as executor:
    #     futures = []
    #     for grid_key, grid_videos in grid.items():
    #         for video in grid_videos:
    #             futures.append(executor.submit(process_video, video, grid_videos, groups, distance_threshold))
    #     for future in as_completed(futures):
    #         future.result()  # 确保所有任务完成

    # for grid_key, grid_videos in grid.items():
    #     for video in grid_videos:
    #         grouped = False
    #         for group_id, group in list(groups.items()):
    #             for v in group["videos"]:
    #                 if calculate_distance(video, v) < distance_threshold:
    #                     groups[group_id]["videos"].append(video)
    #                     groups[group_id]["count"] += 1
    #                     grouped = True
    #                     break
    #             if grouped:
    #                 break
    #         if not grouped:
    #             group_id = video.cameraIndexCode
    #             groups[group_id] = {"count": 1, "videos": [video]}


        # 使用多进程处理每个网格中的视频

    # with Pool(processes=cpu_count()) as pool:
    #     for grid_key, grid_videos in grid.items():
    #         # 初始化局部分组
    #         partial_groups = pool.starmap(process_video, [(video, grid_videos, groups.copy(), distance_threshold) for video in grid_videos])
    #         # 合并局部分组结果
    #         for partial_group in partial_groups:
    #             for group_id, group in partial_group.items():
    #                 if group_id in groups:
    #                     groups[group_id]["videos"].extend(group["videos"])
    #                     groups[group_id]["count"] += group["count"]
    #                 else:
    #                     groups[group_id] = group
    # return groups
    return groups
def process_video(video, grid_videos, groups, distance_threshold):
    grouped = False
    for group_id, group in list(groups.items()):
        for v in group["videos"]:
            if calculate_distance(video, v) < distance_threshold:
                group["videos"].append(video)
                group["count"] += 1
                grouped = True
                break
        if grouped:
            break
    if not grouped:
        group_id = video.cameraIndexCode
        groups[group_id] = {"count": 1, "videos": [video]}
    return groups