@ -1,11 +1,12 @@
from datetime import datetime from datetime import datetime
from typing import Optional , Union
from sqlalchemy import select , func , cast , Numeric , text from sqlalchemy import select , func , cast , Numeric , text
from sqlalchemy . orm import Session from sqlalchemy . orm import Session
from sqlalchemy import and_ from sqlalchemy import and_
from sqlalchemy . sql import not_ from sqlalchemy . sql import not_
from src . maximo . model import WorkOrderData # Assuming this is where your model is from src . maximo . model import WorkOrderData # Assuming this is where your model is
from src . database . core import CollectorDbSession from src . database . core import CollectorDbSession , DbSession
from src . overhaul_scope . model import OverhaulScope
async def get_cm_cost_summary ( collector_db : CollectorDbSession , last_oh_date : datetime , upcoming_oh_date : datetime ) : async def get_cm_cost_summary ( collector_db : CollectorDbSession , last_oh_date : datetime , upcoming_oh_date : datetime ) :
query = text ( """ WITH part_costs AS (
query = text ( """ WITH part_costs AS (
@ -251,3 +252,81 @@ async def get_oh_cost_summary(collector_db: CollectorDbSession, last_oh_date:dat
return {
return {
item [ " location_tag " ] : item [ " avg_cost " ] for item in data
item [ " location_tag " ] : item [ " avg_cost " ] for item in data
}
}
from uuid import UUID
async def get_history_oh_wo ( * , db_session : DbSession , collector_db_session : CollectorDbSession , oh_session_id : UUID , parent_wo_num : Optional [ Union [ str , list ] ] = None ) :
## Get Parent wo num from oh session table
if not parent_wo_num :
query = select ( OverhaulScope . wo_parent ) . where ( OverhaulScope . id == oh_session_id )
result = await db_session . execute ( query )
parent_wo_num = result . scalar ( )
if not parent_wo_num :
return [ ]
# Ensure parent_wo_num is a list and removed duplicates if any
if isinstance ( parent_wo_num , str ) :
parent_wo_num = [ parent_wo_num ]
else :
parent_wo_num = list ( set ( parent_wo_num ) )
sql_query = text ( """
WITH target_wos AS (
SELECT
w . wonum ,
w . assetnum ,
COALESCE ( w . actmatcost , 0 ) as actmatcost ,
COALESCE ( w . actservcost , 0 ) as actservcost
FROM public . wo_maximo w
WHERE w . xx_parent = ANY ( : parent_wo_num )
) ,
wo_tasks AS (
SELECT
t . xx_parent AS parent_wonum ,
JSON_AGG ( t . description ) AS task_list
FROM public . wo_maximo t
JOIN target_wos tw ON t . xx_parent = tw . wonum
GROUP BY t . xx_parent
)
SELECT
w . assetnum ,
e . name AS equipment_name ,
e . location_tag ,
JSON_OBJECT_AGG ( w . wonum , COALESCE ( wt . task_list , ' [] ' : : json ) ) AS wonum_list ,
COUNT ( w . wonum ) AS total_wo_count ,
COALESCE ( SUM ( w . actmatcost ) , 0 ) AS total_material_cost ,
COALESCE ( SUM ( w . actservcost ) , 0 ) AS total_service_cost ,
COALESCE ( SUM ( w . actmatcost + w . actservcost ) , 0 ) AS total_actual_cost
FROM target_wos w
INNER JOIN public . ms_equipment_master e
ON w . assetnum = e . assetnum
LEFT JOIN wo_tasks wt
ON w . wonum = wt . parent_wonum
GROUP BY
w . assetnum ,
e . name ,
e . location_tag
ORDER BY total_actual_cost DESC ;
""" )
results = await collector_db_session . execute ( sql_query , { " parent_wo_num " : parent_wo_num } )
return [
{
" assetnum " : row . assetnum ,
" equipment_name " : row . equipment_name ,
" location_tag " : row . location_tag ,
" wonum_list " : row . wonum_list ,
" total_wo_count " : row . total_wo_count ,
" total_material_cost " : float ( row . total_material_cost ) ,
" total_service_cost " : float ( row . total_service_cost ) ,
" total_actual_cost " : float ( row . total_actual_cost )
}
for row in results
]
