Untitled

mail@pastecode.io avatar
unknown
plain_text
a month ago
25 kB
4
Indexable
Never
from ..repositories.productivity_repository import ProductivityRepository
from ..services.helpers.datetime_helper import DatetimeHelper
class ProductivityService():
    def __init__(self,repository:ProductivityRepository) -> None:
        self.repository = repository
    

    def get_productivity_filters(self,shop_id):
        data = self.repository.get_productivity_filters(shop_id=shop_id)
        return self.transform_productivity_filters(data)

    def transform_productivity_filters(self,data):
        result = []

        for each in data:
            line_id,line_name,model_id,model_name,variant_id,variant_name,subline_id,subline_name,station_id,station_name = each
            result.append({
                "line_id": line_id,
                "line_name": line_name,
                "model_id": model_id,
                "model_name": model_name,
                "variant_id": variant_id,
                "variant_name": variant_name,
                "subline_id": subline_id,
                "subline_name": subline_name,
                "station_id" : station_id,
                "station_name" : station_name
            })

        return result
    
    def get_productivity_cycle_time_metrics_live(self,shop_id,
                                                line_arr = [],
                                                subline_arr=[],
                                                station_arr=[],
                                                model_arr=[],
                                                variant_arr=[],
                                                is_concerned = False):
                                                

        data = self.repository.get_productivity_cycle_time_metrics_live(shop_id=shop_id,
                                                                        line_arr=line_arr,
                                                                        subline_arr=subline_arr,
                                                                        station_arr=station_arr,
                                                                        model_arr=model_arr,
                                                                        variant_arr=variant_arr)

        return self.tranform_productivity_cycle_time_metrics_live(data,is_concerned)





    def get_productivity_cycle_time_metrics_historic(self,shop_id,
                                                     line_arr = [],
                                                     subline_arr=[],
                                                     station_arr=[],
                                                     model_arr=[],
                                                     variant_arr=[],
                                                     from_date=None,
                                                     to_date=None,
                                                     is_concerned = False,
                                                     is_delay = None,
                                                    is_tip_dress = None,
                                                    is_fault=None,based_on = None):
        
        

        data = self.repository.get_productivity_cycle_time_metrics_historic(shop_id = shop_id,
                                                                            line_arr=line_arr,
                                                                            subline_arr=subline_arr,
                                                                            station_arr=station_arr,
                                                                            model_arr=model_arr,
                                                                            variant_arr=variant_arr,
                                                                            from_date=from_date,
                                                                            to_date=to_date,
                                                                            is_delay=is_delay,
                                                                            is_fault=is_fault,
                                                                            is_tip_dress=is_tip_dress,
                                                                            based_on=based_on
                                                                            )
        return self.tranform_productivity_cycle_time_metrics_live(data,is_concerned)

    def tranform_productivity_cycle_time_metrics_live(self,data,is_concerned):
        line_arr = []
        subline_cycle_time_dict = {}

        for each in data:
            (max_id,parameter_id,standard_station_cycle_time
            ,current_station_cycle_time,standard_line_cycle_time,line_id
            ,line_name,model_id,model_name,subline_id,subline_name,variant_id,
            variant_name,part_id,part_name,station_id,station_name,feeder_line,zone) = each
            if zone is None:
                zone = 'GREY'

            if is_concerned and zone == 'RED':
                line_obj = self.get_line_obj_for_cycle_time(line_id=line_id,line_name=line_name,line_arr = line_arr)
                subline_obj = self.get_subline_obj_for_cycle_time(subline_id=subline_id,
                                                                subline_name=subline_name,
                                                                subline_arr=line_obj['subline_arr'],
                                                                standard_value=standard_line_cycle_time,
                                                                current_value=0)
                
                station_obj = self.get_station_obj_for_cycle_time(station_id=station_id,station_name=station_name,
                                                                standard_value= standard_station_cycle_time,
                                                                current_value= current_station_cycle_time,
                                                                station_arr=subline_obj['station_arr'],zone=zone)
                
                self.get_feeder_line_arr(station_id,station_obj,feeder_line)
                
                part_obj = self.get_part_obj_for_cycle_time(part_id=part_id,part_name = part_name
                                                    ,variant_id=variant_id,variant_name=variant_name
                                                    ,model_id=model_id,model_name=model_name,part_arr=station_obj['part_arr'])
            elif is_concerned is False:
                line_obj = self.get_line_obj_for_cycle_time(line_id=line_id,line_name=line_name,line_arr = line_arr)
                subline_obj = self.get_subline_obj_for_cycle_time(subline_id=subline_id,
                                                                subline_name=subline_name,
                                                                subline_arr=line_obj['subline_arr'],
                                                                standard_value=standard_line_cycle_time,
                                                                current_value=0)
                
                station_obj = self.get_station_obj_for_cycle_time(station_id=station_id,station_name=station_name,
                                                                standard_value= standard_station_cycle_time,
                                                                current_value= current_station_cycle_time,
                                                                station_arr=subline_obj['station_arr'],zone=zone)
                
                self.get_feeder_line_arr(station_id,station_obj,feeder_line)
                
                part_obj = self.get_part_obj_for_cycle_time(part_id=part_id,part_name = part_name
                                                    ,variant_id=variant_id,variant_name=variant_name
                                                    ,model_id=model_id,model_name=model_name,part_arr=station_obj['part_arr'])

        

        for line_obj in line_arr:
            subline_arr = line_obj['subline_arr']
            for subline_obj in subline_arr:
                subline_obj['current_value'] = self.get_max_of_line_cycle_time(subline_obj['station_arr'])

        return line_arr
    
    def get_max_of_line_cycle_time(self,station_arr):
        maxx = 0
        for station_obj in station_arr:
            if station_obj['current_value'] != None and  maxx < station_obj['current_value'] :
                maxx = station_obj['current_value'] 
        return maxx


    def get_part_obj_for_cycle_time(self,part_id,part_name,model_id,model_name,variant_id,variant_name,part_arr):
        if(len(part_arr) > 0 and part_arr[-1].get('part_id') == part_id
            and part_arr[-1].get('model_id') == model_id
            and part_arr[-1].get('variant_id') == variant_id):
            station_obj = part_arr[-1] 
        else:
            station_obj = {
                'part_id' : part_id,
                'part_serial_no' : part_name,
                'model_id' : model_id,
                'model_name' : model_name,
                'variant_id' : variant_id,
                'variant_name' : variant_name,
            } 
            part_arr.append(station_obj)
        return station_obj
        

    def get_station_obj_for_cycle_time(self,station_id,station_name,
                                       standard_value,current_value,
                                       station_arr,zone):
        if(len(station_arr) > 0 and station_arr[-1].get('station_id') == station_id):
            station_obj = station_arr[-1] 
        else:
            station_obj = {
                'station_id' : station_id,
                'station_name' : station_name,
                'standard_value': standard_value,
                'current_value' : current_value,
                'zone' : zone,
                'feeder_line' : [],
                'part_arr' : [] ,
                'downtime_arr' : []
            } 
            station_arr.append(station_obj)
        return station_obj
        

    def get_subline_obj_for_cycle_time(self,subline_id,subline_name,subline_arr,
                                       standard_value,current_value):
        if(len(subline_arr) > 0 and subline_arr[-1].get('subline_id') == subline_id):
            subline_obj = subline_arr[-1] 
        else:
            subline_obj = {
                'subline_id' : subline_id,
                'subline_name' : subline_name,
                'standard_value': standard_value,
                'current_value' : current_value,
                'station_arr' : [] ,
                'downtime_arr' : []
            } 
            subline_arr.append(subline_obj)
        return subline_obj
    
    def get_line_obj_for_cycle_time(self,line_id,line_name,line_arr):
        if(len(line_arr) > 0 and line_arr[-1].get('line_id') == line_id):
            line_obj = line_arr[-1] 
        else:
            line_obj = {
                'line_id' : line_id,
                'line_name' : line_name,
                'subline_arr' : [] 
            } 
            line_arr.append(line_obj)
        return line_obj

    def productivity_production_metrics_live(self,shop_id,
                                            line_arr = [],
                                            subline_arr=[],
                                            station_arr=[],
                                            model_arr=[],
                                            variant_arr=[],
                                            from_date=None,
                                            to_date=None,
                                            is_concerned = False):
        
        data = self.repository.productivity_production_metrics_live(shop_id=shop_id,
                                                                    line_arr=line_arr,
                                                                    subline_arr=subline_arr,
                                                                    station_arr=station_arr,
                                                                    model_arr=model_arr,
                                                                    variant_arr=variant_arr,
                                                                    from_date=from_date,
                                                                    to_date=to_date)
        
        return self.tranform_productivity_production_metrics(data,is_concerned)
    

    def productivity_production_metrics_historic(self,shop_id,
                                            line_arr = [],
                                            subline_arr=[],
                                            station_arr=[],
                                            model_arr=[],
                                            variant_arr=[],
                                            from_date=None,
                                            to_date=None,
                                            is_concerned = False):
        
        data = self.repository.productivity_production_metrics_historic(shop_id=shop_id,
                                                                    line_arr=line_arr,
                                                                    subline_arr=subline_arr,
                                                                    station_arr=station_arr,
                                                                    model_arr=model_arr,
                                                                    variant_arr=variant_arr,
                                                                    from_date=from_date,
                                                                    to_date=to_date)
        
        return self.tranform_productivity_production_metrics_historic(data,is_concerned)
    
    def tranform_productivity_production_metrics_historic(self,data,is_concerned):
        line_arr = []

        for each in data:
            (line_id,line_name,subline_id,subline_name,station_id,
             station_name,planned_qty,subline_qty,station_qty,feeded_subline_id) = each
            if planned_qty is None or station_qty is None:
                zone = 'GREY'
            else :
                zone = 'RED' if planned_qty > station_qty else 'GREEN'
            if is_concerned and zone == 'RED':
                line_obj =  self.get_line_obj_for_cycle_time(line_id,line_name,line_arr)
                subline_obj = self.get_subline_obj_for_cycle_time(subline_id,subline_name,line_obj['subline_arr'],planned_qty,subline_qty)
                station_obj = self.get_station_obj_for_cycle_time(station_id,station_name,planned_qty,station_qty,subline_obj['station_arr'],zone)
                self.get_feeder_line_arr(station_id,station_obj,feeded_subline_id)
                # self.get_part_obj_for_cycle_time(None,None,None,None,None,None,station_obj['part_arr'])
            
            elif is_concerned is False:
                line_obj =  self.get_line_obj_for_cycle_time(line_id,line_name,line_arr)
                subline_obj = self.get_subline_obj_for_cycle_time(subline_id,subline_name,line_obj['subline_arr'],planned_qty,subline_qty)
                station_obj = self.get_station_obj_for_cycle_time(station_id,station_name,planned_qty,station_qty,subline_obj['station_arr'],zone)
                self.get_feeder_line_arr(station_id,station_obj,feeded_subline_id)
                # self.get_part_obj_for_cycle_time(None,None,None,None,None,None,station_obj['part_arr'])
        return line_arr    
    
    def tranform_productivity_production_metrics(self,data,is_concerned):
        line_arr = []

        for each in data:
            (line_id,line_name,planned_qty,subline_id,subline_name,station_id,
             station_name,model_id,model_name,variant_id,variant_name,part_id,
             part_name,subline_qty,responsible_station,feeded_subline_id,station_qty,part_serial_no) = each
            if planned_qty is None or station_qty is None:
                zone = 'GREY'
            else :
                zone = 'RED' if planned_qty > station_qty else 'GREEN'
            if is_concerned and zone == 'RED':
                line_obj =  self.get_line_obj_for_cycle_time(line_id,line_name,line_arr)
                subline_obj = self.get_subline_obj_for_cycle_time(subline_id,subline_name,line_obj['subline_arr'],planned_qty,subline_qty)
                station_obj = self.get_station_obj_for_cycle_time(station_id,station_name,planned_qty,station_qty,subline_obj['station_arr'],zone)
                self.get_feeder_line_arr(station_id,station_obj,feeded_subline_id)
                self.get_part_obj_for_cycle_time(part_id,part_serial_no,model_id,model_name,variant_id,variant_name,station_obj['part_arr'])
            
            elif is_concerned is False:
                line_obj =  self.get_line_obj_for_cycle_time(line_id,line_name,line_arr)
                subline_obj = self.get_subline_obj_for_cycle_time(subline_id,subline_name,line_obj['subline_arr'],planned_qty,subline_qty)
                station_obj = self.get_station_obj_for_cycle_time(station_id,station_name,planned_qty,station_qty,subline_obj['station_arr'],zone)
                self.get_feeder_line_arr(station_id,station_obj,feeded_subline_id)
                self.get_part_obj_for_cycle_time(part_id,part_serial_no,model_id,model_name,variant_id,variant_name,station_obj['part_arr'])
        return line_arr
            

    def get_feeder_line_arr(self,station_id,station_obj,feeded_subline_id):
        if(station_obj.get('station_id') == station_id):
            station_obj.get('feeder_line').append(feeded_subline_id)
 
    def get_productivity_efficiency(self,
                                    shop_id,
                                    line_arr = [],
                                    subline_arr=[],
                                    station_arr=[],
                                    model_arr=[],
                                    variant_arr=[]):
        
        downtime_data,break_data  = self.repository.productivity_efficiency(shop_id=shop_id,
                                                line_arr=line_arr,
                                                subline_arr=subline_arr,
                                                station_arr=station_arr,
                                                model_arr=model_arr,
                                                variant_arr=variant_arr)

        self.get_productivity_efficiency_metrics_live(downtime_data,break_data)

    

    def get_productivity_efficiency_metrics_live(self,downtime_data,break_data):
        line_arr = []

        for each in downtime_data:
            (line_id,line_name,subline_id,subline_name,
             station_id,station_name,model_id,model_name,
             variant_id,variant_name,part_id,part_name,
             start_time,end_time) = each
            
            line_obj = self.get_line_obj_for_cycle_time(line_arr=line_arr,line_id=line_id,line_name=line_name)
            subline_obj = self.get_subline_obj_for_productivity(subline_id,subline_name,line_obj['subline_arr'],0,0)
            self.add_downtime_in_obj(subline_obj,start_time,end_time)
            station_obj = self.get_station_obj_for_cycle_time(station_id,station_name,0,0,subline_obj['station_arr'],None)
            self.add_downtime_in_obj(station_obj,start_time,end_time)
            self.get_part_obj_for_cycle_time(part_id,part_name,model_id,model_name,variant_id,variant_name,station_obj['part_arr'])
        

        for line_obj in line_arr:
            subline_arr = line_obj['subline_arr']
            start_time = "06:30:00"
            total_break_time = self.get_break_time(break_data,start_time)
            for subline_obj in subline_arr:
                subline_obj['downtime_arr'] = self.remove_downtime_overlapping(subline_obj.get('downtime_arr'))
                station_arr = subline_obj['station_arr']
                total_subline_downtime = self.get_overall_downtime(subline_obj['downtime_arr'])
                total_working_time = self.get_total_working_time(start_time,total_break_time)
                subline_obj['current_value'] = ((total_working_time - total_subline_downtime)/total_working_time)*100
                for station_obj in station_arr:
                    station_obj['downtime_arr'] = self.remove_downtime_overlapping(station_obj.get('downtime_arr'))
                    total_station_downtime = self.get_overall_downtime(station_obj['downtime_arr'])
                    total_working_time = self.get_total_working_time(start_time,total_break_time)
                    station_obj['current_value'] = ((total_working_time - total_station_downtime)/total_station_downtime)*100



    def get_overall_downtime(downtime_arr):
        total_downtime = []
        for each in downtime_arr:
            total_downtime += each[1]-each[0]
        return total_downtime

    def get_total_working_time(self,start_time,break_time):
        total_working_time = 0
        current_time = DatetimeHelper.get_ist_datetime()
        if current_time <= "23:59:59" and current_time >= start_time:
            total_working_time += current_time - start_time
        elif (current_time < start_time and current_time > "00:00:00"):
            total_working_time += ("23:59:59"- start_time)
            total_working_time += current_time - "00:00:00"

        return total_working_time - break_time



    def get_break_time(self,break_data,start_time):
        total_break_time = 0
        for each in break_data:
            current_time = DatetimeHelper.get_ist_datetime()
            break_start_time,break_end_time,shop_id = each
            if current_time < "23:59:59" and current_time > start_time:
                if start_time < break_start_time <= current_time:
                    total_break_time += (break_end_time - break_start_time)
            elif(current_time < start_time and current_time > "00:00:00"):
                if "00:00:00" < break_start_time <= current_time:
                    total_break_time += (break_end_time - break_start_time)
        return total_break_time


    def get_station_obj_for_efficiency(self,station_id,station_name,
                                       standard_value,current_value,
                                       station_arr,zone):
        if(len(station_arr) > 0 and station_arr[-1].get('station_id') == station_id):
            station_obj = station_arr[-1] 
        else:
            station_obj = {
                'station_id' : station_id,
                'station_name' : station_name,
                'standard_value': standard_value,
                'current_value' : current_value,
                'zone' : zone,
                'feeder_line' : [],
                'part_arr' : [] ,
                'downtime_arr' : []
            } 
            station_arr.append(station_obj)
        return station_obj


    def get_subline_obj_for_productivity(self,subline_id,subline_name,subline_arr,
                                       standard_value,current_value):
        if(len(subline_arr) > 0 and subline_arr[-1].get('subline_id') == subline_id):
            subline_obj = subline_arr[-1] 
        else:
            subline_obj = {
                'subline_id' : subline_id,
                'subline_name' : subline_name,
                'standard_value': standard_value,
                'current_value' : current_value,
                'station_arr' : [] ,
                'downtime_arr' : []
            } 
            subline_arr.append(subline_obj)
        return subline_obj

    def add_downtime_in_obj(self,obj,start_time,end_time):
        obj.get('downtime_arr').append([start_time,end_time])

    def remove_downtime_overlapping(self,downtime_arr):
        n = len(downtime_arr)
        res = []
        for i in range(0,n):
            current = []
            current[0] = downtime_arr[i][0]
            current[1] = downtime_arr[i][1]
            start = i+1
            while(start<n):
                if(current[0] <= downtime_arr[start][0] and current[1] >= downtime_arr[start][1]):
                    i+=1
                elif(current[0] <= downtime_arr[start][0] and current[1] >= downtime_arr[start][0]):
                    if(current[1] <= downtime_arr[start][1]):
                        current[1] = downtime_arr[start][1]
                    i = i+1
                else:
                    i = start - 1
                    break
                start  =start+1
            res.append(current)
        return res
    
    # def get_line_obj(self,line_id,line_name,start_time,end_time):

Leave a Comment