Cook Houdini TOP network from python or command line with progress output

houdini_cook_top_network.py

def cook_top_network(hip_filepath: str, top_node_path: str, parm_dict: Dict[str, Any]):
    print(f'Process ID: {os.getpid()}')

    import hou
    print(f'Loading hip file: {hip_filepath}')
    hou.hipFile.load(hip_filepath)
    top_node: hou.TopNode = hou.node(top_node_path)
    assert isinstance(top_node, hou.TopNode)
    for parm_name, value in parm_dict.items():
        print(f'setting {parm_name} to {value}')
        top_node.parm(parm_name).set(value)

    pdg_context = top_node.getPDGGraphContext()

    import pdg

    def node_progress_update(event):
        node: pdg.Node = event.node
        if node is None:
            return
        stats = node.stats()
        cooked_success_count = stats.stateCount(pdg.workItemState.CookedSuccess)
        cooked_cache_count = stats.stateCount(pdg.workItemState.CookedCache)
        all_count = stats.workItemCount()
        print(f'Node progress update: {event.node.name}: {cooked_success_count}+{cooked_cache_count}/{all_count}')
    
    def work_items_state_change(event):
        if event.currentState == pdg.workItemState.CookedFail:
            work_item = pdg_context.graph.workItemById(event.workItemId)
            print(f'WorkItem CookedFail: {work_item.name}, index: {work_item.index}')
            print(work_item.logMessages)

    def node_genertated(event):
        print(f'Node generated: {event.node.name}')
        node: pdg.Node = event.node
        node.addEventHandler(work_items_state_change, pdg.EventType.WorkItemStateChange)
        node.addEventHandler(node_progress_update, pdg.EventType.NodeProgressUpdate)
    
    def node_cooked(event):
        print(f'Node cooked: {event.node.name}')

    def cook_warning(event):
        node = event.node
        if node:
            print(f'Cook Warning: {node.name}: {event.message}')
        else:
            print(f'Cook Warning: {event.message}')
    
    def cook_error(event):
        node = event.node
        if node:
            print(f'Cook Error: {node.name}: {event.message}')
        else:
            print(f'Cook Error: {event.message}')

    def cook_complete(event):
        print('Graph cook complete!')

    # Register the handlers
    pdg_context.addEventHandler(node_genertated, pdg.EventType.NodeGenerated)
    pdg_context.addEventHandler(node_cooked, pdg.EventType.NodeCooked)
    pdg_context.addEventHandler(cook_warning, pdg.EventType.CookWarning)
    pdg_context.addEventHandler(cook_error, pdg.EventType.CookError)
    pdg_context.addEventHandler(cook_complete, pdg.EventType.CookComplete)

    top_node.cookOutputWorkItems(block=True, generate_only=True)
    top_node.cookOutputWorkItems(block=True)
    display_node = top_node.displayNode()
    pdg_node = display_node.getPDGNode()
    output_stats = pdg_node.stats()
    output_work_item_count = output_stats.workItemCount()
    cooked_success_count = output_stats.stateCount(pdg.workItemState.CookedSuccess)
    cooked_cache_count = output_stats.stateCount(pdg.workItemState.CookedCache)
    print(f'Output work item count: {output_work_item_count}, Success: {cooked_success_count}, Cache: {cooked_cache_count}')
    if display_node.getCookState(force=False) != hou.topCookState.Cooked or cooked_success_count + cooked_cache_count != output_work_item_count:
        print('PDG Cook Result: Failed')
        return False

    print('PDG Cook Result: Cooked')
    return True