"""CSC111 Winter 2023 Assignment 3: Graphs and Interconnection Networks Instructions (READ THIS FIRST!) =============================== This Python module contains the start of functions and/or classes you'll define for Part 3 of this assignment. You may, but are not required to, add doctest examples to help test your work. We strongly encourage you to do so! Copyright and Usage Information =============================== This file is provided solely for the personal and private use of students taking CSC111 at the University of Toronto St. George campus. All forms of distribution of this code, whether as given or with any changes, are expressly prohibited. For more information on copyright for CSC111 materials, please consult our Course Syllabus. This file is Copyright (c) 2023 Mario Badr and David Liu. """ import csv from typing import Optional import plotly.graph_objects as go from python_ta.contracts import check_contracts # NOTE: Node and NodeAddress must be imported for check_contracts # to work correctly, even if they aren't being used directly in this # module. Don't remove them (even if you get a warning about them in PyCharm)! from a3_network import AbstractNetwork, Packet, NodeAddress, Node from a3_simulation import NetworkSimulation, PacketStats # The different networks you're implementing on this assignment. Don't worry if # the a3_part4 networks are unused right now; you'll use them in this file after # completing Part 4. from a3_part2 import AlwaysRightRing, ShortestPathRing, ShortestPathTorus, ShortestPathStar from a3_part4 import GreedyChannelRing, GreedyChannelTorus, GreedyChannelStar,\ GreedyPathRing, GreedyPathTorus, GreedyPathStar def run_example() -> list[PacketStats]: """Run an example simulation. You may, but are not required to, change the code in this example to experiment with the simulation. """ network = AlwaysRightRing(5) simulation = NetworkSimulation(network) packets = [(0, Packet(1, 0, 4))] return simulation.run_with_initial_packets(packets, print_events=True) @check_contracts def read_packet_csv(csv_file: str) -> tuple[AbstractNetwork, list[tuple[int, Packet]]]: """Load network and packet data from a CSV file. Return a tuple of two values: - the first element is the network created from the specification in the first line of the CSV file - the second element is a list of tuples, where each tuple is of the form (timestamp, packet), created from all other lines of the CSV file Preconditions: - csv_file refers to a valid CSV file in the format described on the assignment handout Implementation hints: - Since it's the last assignment, we've deliberately not given you *any* startter code for reading CSV files! Refer back to past assignments/tutorials. Hint: treat the first line differently than all other lines. - You *may* use a big if statement to handle each different network type separately. - Remember to convert entries into ints where appropriate. """ with open(csv_file) as file: reader = csv.reader(file) network_data = next(reader) class_name = network_data[0] args = [int(arg) for arg in network_data[1:] if arg != ''] network = None # I DO NOT NEED THIS. HOWEVER, TA CAUSES ERROR WITHOUT THIS. if class_name == 'AlwaysRightRing': network = AlwaysRightRing(args[0]) elif class_name == 'ShortestPathRing': network = ShortestPathRing(args[0]) elif class_name == 'ShortestPathTorus': network = ShortestPathTorus(args[0]) elif class_name == 'ShortestPathStar': network = ShortestPathStar(args[0], args[1]) elif class_name == 'GreedyChannelRing': # Part 4 Q1 network = GreedyChannelRing(args[0]) elif class_name == 'GreedyChannelTorus': network = GreedyChannelTorus(args[0]) elif class_name == 'GreedyChannelStar': network = GreedyChannelStar(args[0], args[1]) elif class_name == 'GreedyPathRing': # Part 4 Q2 network = GreedyPathRing(args[0]) elif class_name == 'GreedyPathTorus': network = GreedyPathTorus(args[0]) elif class_name == 'GreedyPathStar': network = GreedyPathStar(args[0], args[1]) identifier = 0 packets = [] for row in reader: timestamp = int(row[0]) if class_name in ['ShortestPathTorus', 'GreedyChannelTorus', 'GreedyPathTorus']: packet = Packet(identifier, (int(row[1]), int(row[2])), (int(row[3]), int(row[4]))) else: packet = Packet(identifier, int(row[1]), int(row[2])) packets.append((timestamp, packet)) identifier += 1 return (network, packets) def plot_packet_latencies(packet_stats: list[PacketStats]) -> None: """Use plotly to plot a histogram of the packet latencies for the given stats. The packet latency is defined as the difference between the arrived_at and created_at times. It represents the total amount of time the packet spent in the network. We have provided some starter code for you. Preconditions: - packet_stats != [] - all(stats.arrived_at is not None for stats in packet_stats) - all(stats.route != [] for stats in packet_stats) """ fig = go.Figure(data=[ go.Histogram( x=[stats.arrived_at - stats.created_at for stats in packet_stats], ) ]) # Set the graph title and axis labels fig.update_layout( title='Packet Latency Histogram', xaxis_title_text='Packet Latency', yaxis_title_text='Count' ) fig.show() def plot_route_lengths(packet_stats: list[PacketStats]) -> None: """Use plotly to plot a histogram of the route lengths for the given stats. The route length is defined as the number of channels traversed by the packet to arrive at its destination. We have not provided any code, but your implementation should be pretty similar to plot_packet_latencies. Remember to update the histogram title and axis labels! Preconditions: - packet_stats != [] """ fig = go.Figure(data=[ go.Histogram( x=[len(stats.route) - 1 for stats in packet_stats], # "# of traversed", meaning len - 1. ) ]) fig.update_layout( title='Route Length Histogram', xaxis_title_text='Route Length', yaxis_title_text='Count' ) fig.show() @check_contracts def part3_runner(csv_file: str, plot_type: Optional[str] = None) -> dict[str, float]: """Run a simulation based on the data from the given csv file. If plot_type == 'latencies', plot a histogram of the packet latencies. If plot_type == 'route-lengths', plot a histogram of the packet route lengths. Return a dictionary with two keys: - 'average latency', whose associated value is the average packet latency - 'average route length', whose associated value is the average route length Preconditions: - csv_file refers to a valid CSV file in the format described on the assignment handout - plot_type in {None, 'latencies', 'route-lengths'} - the given CSV file contains at least one packet line """ network, packets = read_packet_csv(csv_file) network_simulation = NetworkSimulation(network) list_of_packet_stats = network_simulation.run_with_initial_packets(packets, print_events=False) # Turn this true. # Very cool to have print_events as True, but just in case MarkUs causes problem. if plot_type == 'latencies': plot_packet_latencies(list_of_packet_stats) elif plot_type == 'route-lengths': plot_route_lengths(list_of_packet_stats) avg_latency = sum( [stats.arrived_at - stats.created_at for stats in list_of_packet_stats]) / len(list_of_packet_stats) avg_route_length = sum([len(stats.route) - 1 for stats in list_of_packet_stats]) / len(list_of_packet_stats) return { 'average latency': avg_latency, 'average route length': avg_route_length } @check_contracts def part3_runner_optional(network: AbstractNetwork, plot_type: Optional[str] = None) -> dict[str, float]: """An optional runner. You may use this function to experiment with using a3_simulation.generate_random_initial_events, for example. You may modify the function header (e.g., by adding parameters or changing the return type) however you like. Your work in this function will not be graded (though it will still be checked by PythonTA). Preconditions: - the given CSV file contains at least one packet line """ # THIS IS TO AVOID TA ERROR. I'M NOT IMPLEMENTING part3_runner_optional. avoid_ta_error = (network, plot_type) if avoid_ta_error: return { 'average latency': ..., 'average route length': ... } return { 'average latency': ..., 'average route length': ... } if __name__ == '__main__': # Here is a sample call to part3_runner. Feel free to change it or add new calls! part3_runner('data/ring_single.csv', 'latencies') # When you are ready to check your work with python_ta, uncomment the following lines. # (In PyCharm, select the lines below and press Ctrl/Cmd + / to toggle comments.) # You can use "Run file in Python Console" to run PythonTA, # and then also test your methods manually in the console. import python_ta python_ta.check_all(config={ 'max-line-length': 120, 'extra-imports': ['csv', 'plotly.graph_objects', 'a3_network', 'a3_simulation', 'a3_part2', 'a3_part4'], 'disable': ['unused-import', 'too-many-branches'], 'allowed-io': ['read_packet_csv', 'part3_runner_optional'] })