Check which sinks are connected in the pipe system using Python

Question

It would be very helpful to me as a beginner if I could get feedback on my code, specifically about the efficiency of algorithm used and potential improvements in code quality.

Code context:

There is a pipe system represented by a 2D rectangular grid of cells. There are three different types of objects located in cells in the grid, with each cell having either 0 objects or 1 object: There is 1 source in the system(*). There is an arbitrary number of sinks in the system. They are each represented by a different uppercase letter . There are 10 different shapes of pipes: ╣ ╦ ╠ ╩ ║ ═ ╔ ╗ ╝ ╚ Each pipe has openings on 2 or 3 sides of its cell. Two adjacent cells are connected if both have a pipe opening at their shared edge. Source and sinks have openings on all 4 edges. A sink may be connected to the source through another sink. Task is to find sinks which are connected to the source in a given pipe system. link to input text file that contains rows of data indicating the location of the objects in the grid. Each row has character as object, x coordinate and y coordinate. Rows are given in arbitrary order. Example

• 0 2 (this is source *)

C 1 0

╠ 1 1

╣ 1 2

= 2 1

╚ 3 0

╝ 3 1

= 4 0

= 4 2

B 5 0

A 5 2 **Output grid according to coordinates **

• ╣ 0 ╔ ═ A

0 ╠ ═ ╝ 0 0

0 C 0 ╚ ═ B

I did not use breadth first search algo as there are multiple conditions to take care of . please guide me if there can be potential improvements to my code logic and style.

Step-by-Step Approach Objective:

Start from the sinks and trace the path to the source. Determine if a sink is disconnected early on. This approach is particularly useful when there are many sinks scattered away from the source.

Setup:

Matrix Initialization: Read pipe system data from a file and store it in a matrix (matrix). This matrix represents the grid layout of the pipes. Deque Initialization: Create a deque called sinks to store sinks along with their coordinates. Reading Input: Open and read from the file pro.txt. For each line, update the matrix with pipe characters and append sinks to the deque sinks with their coordinates. Populating Initial Data:

Matrix Population: Populate the matrix with pipe characters based on input. Sink Collection: Append each sink (uppercase letter) and its coordinates to the sinks deque. Initialize Traversal: Append the initial sink coordinates and possible directions (left, right, down) to the visited deque. Processing Each Sink:

Iterate through Sinks: For each sink in the sinks deque, initiate a search starting from the sink's coordinates. Searching:

Initialization in Search: Begin with the initial direction set to Up (U). Update Coordinates: In the outer while loop, update the current coordinates based on the direction. Bounds Check: Call check_bounds() to ensure coordinates are within grid limits. The check_bounds() function handles out-of-bound coordinates by popping from the visited deque until coordinates are valid or no valid moves remain. Processing Grid Cell: Assign the current grid cell to the variable grid and process it based on its type. Nested While Loop in Search: Conditions Handling: If grid is empty space, handle it by popping from visited. If grid is another sink, append its coordinates and possible next directions to visited. Check for blocked paths, and handle vertical and horizontal traversals. Handling Traversals:

Vertical Traversal: Based on the type of pipe and current direction, determine if vertical movement is possible and update the direction accordingly. If vertical movement fails, handle it by calling vertical_unsuccessful(), which checks if the direction is blocked and provides a backup option. Horizontal Traversal: Similar to vertical traversal, handle horizontal movement and update direction as needed. Call horizontal_unsuccessful() if horizontal movement is blocked. Updating States:

If Sink is Disconnected: If the search for a sink is unsuccessful, add the sink to the disconnected list. If Sink is Connected: If successful, add the sink to the connected list. Extend the previous deque with visited to keep track of all visited nodes for future iterations.

In main search function,I added if conditions which works as a switch case in C to check connections consistently .That's why the function was larger as compared to other functions

import time
from collections import deque

# Start timing the execution of the script
start_time=time.time()

# Define the dimensions of the grid
rows=30
cols=50

# Initialize the grid (matrix) to represent the pipe system
matrix=[[" " for _ in range(cols)] for _ in range(rows)]

# Initialize deques for managing sinks and other elements
sinks=deque()
letters=deque()
visited=deque()
disconnected=deque()
connected=deque()
previous=deque()

# Read the pipe system data from the file
with open("pro.txt","r",encoding="utf-8") as file:
    lines=file.readlines()

# Populate the matrix and sinks deque with data from the file
for line in lines:
    l=line.split()
    # (29-int(l[2]) is the offset to adjust coordinates according to array representation using indexes
    matrix[29-int(l[2])][int(l[1])]=l[0]

    # If the line starts with an alphabet, treat it as a sink
    if l[0].isalpha():
        sinks.append([l[0],29-int(l[2]),int(l[1])])

# Print the matrix to visualize the grid
for row in matrix:
    print(*row,sep="\t")
    print("\n")

# Define directions for movement
U=[-1,0]  # Up
D=[1,0]  # Down
R=[0,1]  # Right
L=[0,-1]  # Left

# Define the pipe connections
pipes={
    "╠":(U,L,D),
    "╣":(U,R,D),
    "║":(U,D),
    "╩":(D,R,L),
    "╦":(U,R,L),
    "╚":(D,L),
    "╝":(D,R),
    "╗":(U,R),
    "╔":(U,L),
    "═":(R,L)
}


# Function to check if the coordinates are within bounds of the grid
def check_bounds(index_coord):
    while (index_coord[0]<0 or index_coord[0]>29 or
           index_coord[1]<0 or index_coord[1]>49):
        if visited:
            v=visited.pop()
            index_coord=v[0]
            matrix[index_coord[0]][index_coord[1]]=" "
            traverse=v[1]
            index_coord[0]+=traverse[0]
            index_coord[1]+=traverse[1]
        else:
            return 0
    return index_coord


# Function to check if the current pipe being tested is present
#in previous which was populated with leftover special(3 openings) pipes in visited of previous connected sink
def backup(direct,backup_coord):
    if direct==U and [backup_coord,D] in previous:
        matrix[backup_coord[0]][backup_coord[1]]=" "
        return 1
    if direct==D and [backup_coord,U] in previous:
        matrix[backup_coord[0]][backup_coord[1]]=" "
        return 1
    if direct==R and [backup_coord,L] in previous:
        matrix[backup_coord[0]][backup_coord[1]]=" "
        return 1
    if direct==L and [backup_coord,R] in previous:
        matrix[backup_coord[0]][backup_coord[1]]=" "
        return 1
    return 0


# Function which appends special pipes' or other sinks' co-oridnates
#encountered in the path to visited
def visit_app(visited_obj,visited_coord,direction):
    # if passed directions are U or D that means horizontal directions should be appended
    if direction==U or direction==D:
        if visited_obj.isalpha():
            visited.append([visited_coord,L])
            visited.append([visited_coord,R])
        elif visited_obj in "╣╩╦":
            visited.append([visited_coord,L])
        else:
            visited.append([visited_coord,R])
    if direction==R or direction==L:
        if visited_obj.isalpha():
            visited.append([visited_coord,D])
            visited.append([visited_coord,U])
        elif visited_obj in "╣╦╠":
            visited.append([visited_coord,D])
        else:
            visited.append([visited_coord,U])


# Function to handle unsuccessful vertical moves
def vertical_unsuccessful(v_direct,vertical_grid):
    if v_direct==U:
        if D in pipes[vertical_grid] and U not in pipes[vertical_grid] or vertical_grid=="═":
            if visited:
                v=visited.pop()
                v_coord=v[0]
                matrix[v_coord[0]][v_coord[1]]=" "
                return [v[1],v_coord]
    if v_direct==D:
        if U in pipes[vertical_grid] and D not in pipes[vertical_grid] or vertical_grid=="═":
            if visited:
                v=visited.pop()
                v_coord=v[0]
                matrix[v_coord[0]][v_coord[1]]=" "
                return [v[1],v_coord]
    return 0


# Function to handle vertical moves
def vertical(vertical_direct,vertical_grid,v_coord):
    if vertical_direct==U:
        if U in pipes[vertical_grid]:
            if vertical_grid not in "╠╣╩╦":
                matrix[v_coord[0]][v_coord[1]]=" "
            if vertical_grid in "╠╣╩╦":
                visit_app(vertical_grid,v_coord,vertical_direct)
            if D in pipes[vertical_grid]:
                return U
            if L in pipes[vertical_grid]:
                return R
            else:
                return L
    if vertical_direct==D:
        if D in pipes[vertical_grid]:
            if vertical_grid not in "╠╣╩╦":
                matrix[v_coord[0]][v_coord[1]]=" "
            if vertical_grid in "╠╣╩╦":
                visit_app(vertical_grid,v_coord,vertical_direct)
            if U in pipes[vertical_grid]:
                return D
            if L in pipes[vertical_grid]:
                return R
            else:
                return L
    return 0


# Function to handle unsuccessful horizontal moves
def horizontal_unsuccessful(h_direct,h_grid):
    if h_direct==R:
        if L in pipes[h_grid] and R not in pipes[h_grid] or h_grid=="║":
            if visited:
                v=visited.pop()
                h_coord=v[0]
                matrix[h_coord[0]][h_coord[1]]=" "
                return [v[1],h_coord]
    if h_direct==L:
        if R in pipes[h_grid] and L not in pipes[h_grid] or h_grid=="║":
            if visited:
                v=visited.pop()
                h_coord=v[0]
                matrix[h_coord[0]][h_coord[1]]=" "
                return [v[1],h_coord]
    return 0


# Function to handle horizontal moves
def horizontal(h_traverse,h_grid,h_coord):
    if h_traverse==R:
        if R in pipes[h_grid]:
            if h_grid not in "╠╣╩╦":
                matrix[h_coord[0]][h_coord[1]]=" "
            if h_grid in "╠╣╩╦":
                visit_app(h_grid,h_coord,h_traverse)
            if L in pipes[h_grid]:
                return R
            if D in pipes[h_grid]:
                return U
            else:
                return D
    if h_traverse==L:
        if L in pipes[h_grid]:
            if h_grid not in "╠╣╩╦":
                matrix[h_coord[0]][h_coord[1]]=" "
            if h_grid in "╠╣╩╦":
                visit_app(h_grid,h_coord,h_traverse)
            if R in pipes[h_grid]:
                return L
            if D in pipes[h_grid]:
                return U
            else:
                return D
    return 0


# Function to search through the grid starting from a given coordinate
def search(coord):
    traverse=U  # Start by moving Up
    while True:
        coord=coord.copy()
        coord[0]+=traverse[0]
        coord[1]+=traverse[1]
        ch=check_bounds(coord)  # Check if the new coordinate is within bounds
        if not ch:
            return 0
        coord=ch
        grid=matrix[coord[0]][coord[1]]  # Get the value at the new coordinate

        while True:
            if grid==" ":
                if visited:
                    v=visited.pop()
                    coord=v[0]
                    matrix[coord[0]][coord[1]]=" "
                    traverse=v[1]
                    break
                return 0
            if grid in "╠╣╩╦" or grid.isalpha():
                if backup(traverse,coord):
                    return 1
            if grid.isalpha():
                letters.append(grid)
                visit_app(grid,coord,traverse)
                break
            if grid=="*":
                return 1
            if traverse==U or traverse==D:
                vert=vertical(traverse,grid,coord)
                if vert:
                    traverse=vert
                    break
                ver=vertical_unsuccessful(traverse,grid)
                if ver:
                    traverse=ver[0]
                    coord=ver[1]
                    break
                else:
                    return 0
            if traverse==R or traverse==L:
                h=horizontal(traverse,grid,coord)
                if h:
                    traverse=h
                    break
                h=horizontal_unsuccessful(traverse,grid)
                if h:
                    traverse=h[0]
                    coord=h[1]
                    break
                else:
                    return 0


# Main processing loop for each sink
for sink in sinks:
    s=sink[0]  # The sink identifier
    s_x=sink[1]  # The x-coordinate of the sink
    s_y=sink[2]  # The y-coordinate of the sink
    val=[s_x,s_y]

    # Skip if the sink is already processed
    if s in connected or s in disconnected:
        continue

    letters.append(s)
    visited.append([val,D])
    visited.append([val,R])
    visited.append([val,L])
    se=search(val)

    if not se:
        disconnected.extend(letters)  # Add the sink to the disconnected list
        letters.clear()  # Clear the letters deque for the next iteration
    else:
        # If the search is successful, update connected and previous lists
        connected.extend(letters)
        letters.clear()
        previous.extend(visited)
        visited.clear()

# Convert the connected deque to a set to remove duplicates, then back to deque
connected=set(connected)
connected=deque(connected)

# Print the sorted connected sinks
print(sorted(connected),"\n")

# Calculate and print the total time taken for execution
end_time=time.time()
execution_time=end_time-start_time
print(f"Total time taken: {execution_time:.2f} seconds")

Answer 1

algorithm

I did not use [BFS] as there are multiple conditions to take care of.

Sorry, I didn't follow the reasoning there. One could profitably apply either BFS or DFS to this problem.

There is an opportunity to "$ pip install networkx" and take advantage of its graph traversal algorithms and data structures. I understand you're doing this as a learning exercise; installing a package like NetworkX would let you learn new skills. Most python projects will eventually come to rely on the code and documentation of external packages.

main guard

You wrote lots of logic up at module level. Prefer to bury it within def main():, and use the customary guard:

if __name__ == "__main__":
    main()

Then the open cannot raise at import time, and we'll have less coupling from global variables.

When you get around to writing an automated unit test, it will want to be able to safely import this module without side effects, something the current code can't offer. For example if it happens there's no "pro.txt" file we'll raise FileNotFoundError, so "at import time" is the wrong time to be doing that. Defer it until run time, using a __main__ guard.

list vs tuple

I really like the U, D, R, L coordinate deltas. Please make them (dy, dx) tuples, rather than the [dy, dx] lists in OP code.

Why? In python we use list for arbitrary number of "same thing", e.g. names: list[str] = ["Alice", "Bob"]. Position doesn't matter, it doesn't change the meaning of an item.

We use tuple for fixed number of "different things", e.g. in delta: tuple(int, int) = (1, 0) the dy of 1 has an entirely different meaning from the dx of 0 -- they're distances in very different directions. We might even name the tuple elements:
Delta = namedtuple("Delta", "x, y")
delta = Delta(1, 0)

comment

I see why you have "inverted coordinate order". A casual reader might naïvely assume the conventional (dx, dy), so adding a # comment would be helpful. Or name those components.

Similarly, spelling out that "Y coordinates are inverted" from the usual "first quandrant" mathematical conventions wouldn't hurt. Or consider relying on numpy instead of the list-of-lists datastructure.

And on the topic of "backwards", I don't understand this:

    "╠":(U,L,D),

U, D? Perfect!

But L? It looks like R, to me.

Consistently doing mirror flip on X-coordinates results in the "same" problem to be solved. But there's room to better communicate technical details to colleagues collaborating with you.

In any event, using those four symbols is a big win over labeling the ten pieces with a bunch of integers, so that's well done.

side effects

Yikes, this is a bit terrible:

# Function to check if the coordinates are within bounds of the grid
def check_bounds(index_coord):
    while (index_coord[0]<0 or index_coord[0]>29 or
           index_coord[1]<0 or index_coord[1]>49):
        if visited: ...

The # comment is nice enough, I suppose, but please turn it into the function's """docstring""". It should describe the return value, or perhaps the signature could do that with something like
def check_bounds(index_coord) -> int:, or even
def check_bounds(index_coord: tuple[int, int]) -> int:

At first blush this function computes a value to be returned. But then it turns out we don't merely "check" the input. No, we "mutate" a pair of module level global variables which the comment didn't give a hint about at all. Presumably there's some (unmentioned) invariant we're trying to preserve.

The 29 and 49 magic numbers deserve names.

            return 0
    return index_coord

Yow! Holy type stability, Batman!

So it turns out the signature is more like
def check_bounds(index_coord) -> tuple[int, int] | int:,
for reasons I do not yet fathom. Typically we'll be making things easier for the caller if a function consistently returns answers of just one type. Python is a dynamic language, but refrain from the temptation to get carried away.

Avoid writing tricky code. Someone will have to debug it, probably you.

If it's unavoidable, then write a helpful docstring, and some unit tests.

predicate helper function

Rather than returning int, consider adopting this signature:
def backup(direction, backup_coord) -> bool:

It does the same thing. It just conveys the intent a little more clearly, with e.g. return True. Be sure to lint it with "$ mypy --strict *.py"

And since parameter names are part of the Public API you're designing, prefer to spell them out. As a local variable, direct might be fine, but not when it's exposed to callers.

meaningful names

def visit_app( ... ):

I was initially parsing that as "application", before settling on "append". But it's hard to work "visit append" into a sensible English sentence. Consider renaming this. Naming is a challenging, but important task. Sorry, I do not yet know what the right name for this function is.

This function appears to be working too hard, parsing inconvenient "pipe" symbols. Back when you read in the grid configuration, you could fill in two or three edges for each "pipe" node or grid cell. That happens just once. And then this function would have an easier time of it, simply iterating over a given node's edges.

choice of datastructure

def vertical_unsuccessful( ... ): ...
def vertical( ... ): ...
def horizontal_unsuccessful( ... ): ...
def horizontal( ... ): ...

I haven't waded in to find exactly what's going on in there. But when you see "same code, different direction", that should make you reconsider your approach, and maybe choose a datastructure where you can pass in a direction to change the behavior.

For example, with a numpy array, it would be straightforward to work with axis=0 or axis=1 in order to change direction.

automated tests

You have some complex logic and state variables. Go to the trouble of writing a unit test or two. Adding a second caller will help you notice where the documentation or calling convention could be improved. And tests are always a big help when you refactor, as they can reveal unexpected changes in behavior.