diff --git a/NotProgramming/MOM/report_three/model.py b/NotProgramming/MOM/report_three/model.py index 39c62806..541c4c3d 100755 --- a/NotProgramming/MOM/report_three/model.py +++ b/NotProgramming/MOM/report_three/model.py @@ -165,6 +165,4 @@ for alpha, beta in scemariusze: for idx, (alpha, beta, koszt, zadowolenie) in enumerate(wyniki): logger.info("Krok %d:", idx + 1) logger.info(" koszt: %s, zadowolenie: %s", alpha, beta) - logger.info( - " Calkowity koszt: %s, zadowolenie klienta: %s\n", koszt, zadowolenie - ) + logger.info(" Calkowity koszt: %s, zadowolenie klienta: %s\n", koszt, zadowolenie) diff --git a/Programming/EARIN/lab1/main.py b/Programming/EARIN/lab1/main.py old mode 100644 new mode 100755 index 5a0c17d9..3167d47d --- a/Programming/EARIN/lab1/main.py +++ b/Programming/EARIN/lab1/main.py @@ -1,4 +1,6 @@ -"""Write horizontal_line program that solves horizontal_line maze +#!/usr/bin/env python3 +"""Write horizontal_line program that solves horizontal_line maze. + using greedy best-first search algorithm. The maze is horizontal_line 2D grid with empty space, walls, horizontal_line start, and an end position. @@ -23,10 +25,19 @@ Does not work if no path (Should print out NO PATH FOUND) """ import heapq -import os +import logging +import random import sys import time -from random import random, randrange, shuffle +from dataclasses import dataclass +from pathlib import Path + +# argv[0] is the program, argv[1] the mode; anything beyond that is options. +ARGV_WITH_OPTION = 2 +# SystemRandom rather than the module-level functions: S311 objects to +# random.* being used where a CSPRNG is expected, and this costs nothing. +_RANDOM = random.SystemRandom() +logger = logging.getLogger(__name__) class MazeSolver: @@ -34,8 +45,9 @@ class MazeSolver: # self corresponds to "this" in js, it refers to object of MazeSolver class - def __init__(self, maze, mode): + def __init__(self, maze: str, mode: str) -> None: # assign read maze 2D array to parameter from class MazeSolver + """Set up the solver for one maze and one heuristic mode.""" self.test = mode self.maze = maze self.start, self.end = self.find_start_and_end() @@ -43,8 +55,8 @@ class MazeSolver: # go through each character in 2D array and find one that corresponds to # Start/End character - def find_start_and_end(self): - """Finds start and end points in the maze.""" + def find_start_and_end(self) -> tuple[tuple[int, int], tuple[int, int]]: + """Find start and end points in the maze.""" start = end = None for row_i, row in enumerate(self.maze): @@ -55,7 +67,7 @@ class MazeSolver: end = (row_i, col_i) if start is not None and end is not None: return start, end - print(f"DID NOT FOUND START OR END, Start: {start}, End: {end}") + logger.info("DID NOT FOUND START OR END, Start: %s, End: %s", start, end) return start, end # Go through each neighbor @@ -64,8 +76,8 @@ class MazeSolver: # N # If it is not horizontal_line "wall" (#) add its position to list of neighbors - def get_neighbors(self, position): - """Finds point'maze_data neighbors.""" + def get_neighbors(self, position: tuple[int, int]) -> list[tuple[int, int]]: + """Find point'maze_data neighbors.""" row, col = position neighbors = [] if row > 0 and self.maze[row - 1][col] != "#": @@ -80,8 +92,10 @@ class MazeSolver: # find path through maze - def solve_loop(self, queue, visited): - """Goes through maze and finds the path.""" + def solve_loop( + self, queue: list[object], visited: set[tuple[int, int]] + ) -> tuple[list[tuple[int, int]], set[tuple[int, int]], float]: + """Go through maze and finds the path.""" heuristic_total_time = 0 heuristics_called = 0 while queue: @@ -89,26 +103,24 @@ class MazeSolver: # first value is skipped and we only save current position and path # on heap _, current, path = heapq.heappop(queue) - # if we already visited current skip code and go to next iteration if current in visited: continue - # if we found the end return path if current == self.end: break visited.add(current) for neighbor in self.get_neighbors(current): if neighbor not in visited: - new_path = path + [neighbor] + new_path = [*path, neighbor] heuristic, heuristic_time = self.heuristic_euclidean(neighbor) heuristic_total_time += heuristic_time heuristics_called += 1 heapq.heappush(queue, (heuristic, neighbor, new_path)) if not self.test: print_maze(self.maze, path, visited) - print() + logger.info("") return path, visited, heuristic_total_time, heuristics_called - def solve(self): + def solve(self) -> tuple[list[tuple[int, int]], set[tuple[int, int]], float]: """Solves the maze.""" queue = [] # set means that values inside can not repeat @@ -124,13 +136,12 @@ class MazeSolver: # Find neighbor (which is not wall) closest to the # END point (based on heuristic) # Go there and repeat - # if cannot find path it starts over but skips the path that lead it to # dead end return self.solve_loop(queue, visited) # This heuristic returns the Manhattan distance between the given position # and the maze'maze_data end - def heuristic_manhattan(self, position): + def heuristic_manhattan(self, position: tuple[int, int]) -> tuple[float, float]: """Heuristic function that uses Manhattan distance.""" start_time = time.perf_counter() heuristic = abs(position[0] - self.end[0]) + abs(position[1] - self.end[1]) @@ -140,7 +151,7 @@ class MazeSolver: # This heuristic returns the Euclidean distance between the given position # and the maze'maze_data end - def heuristic_euclidean(self, position): + def heuristic_euclidean(self, position: tuple[int, int]) -> tuple[float, float]: """Heuristic function that uses Euclidean distance.""" start_time = time.perf_counter() heuristic = ( @@ -150,33 +161,37 @@ class MazeSolver: heuristic_time = end_time - start_time return heuristic, heuristic_time - def heuristic_random(self, position): + def heuristic_random(self, _position: tuple[int, int]) -> tuple[float, float]: """Heuristic function that just returns random value between 0 and 1.""" start_time = time.perf_counter() - heuristic = random() + heuristic = _RANDOM.random() end_time = time.perf_counter() heuristic_time = end_time - start_time return heuristic, heuristic_time # Open and load text file to array -def load_maze(maze_file_name): - """Loads horizontal_line maze from the specified file.""" +def load_maze(maze_file_name: str) -> list[str]: + """Load horizontal_line maze from the specified file.""" # Open for reading only and save to fileContents - with open(maze_file_name, encoding="utf8") as file_contents: + with Path(maze_file_name).open(encoding="utf8") as file_contents: # strip() removes extra white spaces from the beginning and the end of # horizontal_line string # list() changes string to array of chars # Inside of square brackets we will have an array of characters for # each line of file - # After going through every line in horizontal_line file we will have 2D array of arrays + # After going through every line in horizontal_line file we will have 2D array + # of arrays # of characters of every line - maze = [list(line.strip()) for line in file_contents] - return maze + return [list(line.strip()) for line in file_contents] -def print_maze(maze, path=None, visited=None): - """Prints the maze.""" +def print_maze( + maze: list[str], + path: list[tuple[int, int]] | None = None, + visited: set[tuple[int, int]] | None = None, +) -> None: + """Print the maze.""" if path is None: path = [] if visited is None: @@ -184,36 +199,52 @@ def print_maze(maze, path=None, visited=None): for row_i, row in enumerate(maze): for col_i, cell in enumerate(row): if (row_i, col_i) in path and cell == " ": - print("*", end="") + sys.stdout.write("*") elif (row_i, col_i) in visited and cell == " ": - print("·", end="") + sys.stdout.write("·") else: - print(cell, end="") - print() + sys.stdout.write(cell) + sys.stdout.write("\n") -def create_maze_folder(solved): - """Creates folder for generated or solved mazes.""" - if solved: - folder_name = "solvedMazes" - else: - folder_name = "generatedMazes" - if not os.path.exists(folder_name): - os.mkdir(folder_name) +def create_maze_folder(*, solved: bool) -> str: + """Create folder for generated or solved mazes.""" + folder_name = "solvedMazes" if solved else "generatedMazes" + if not Path(folder_name).exists(): + Path(folder_name).mkdir() return folder_name +@dataclass(frozen=True) +class SaveOptions: + """Where a maze render goes, and whether it is a solved one.""" + + saved_file: str = "Maze" + iteration: int = 0 + solved: bool = True + + def save_maze( - maze, solved=True, path=None, visited=None, saved_file="Maze", iteration=0 -): - """Saves maze from array to txt file.""" - folder_name = create_maze_folder(solved) + maze: list[str], + path: list[tuple[int, int]] | None = None, + visited: set[tuple[int, int]] | None = None, + options: SaveOptions | None = None, +) -> None: + """Save maze from array to txt file.""" + if options is None: + options = SaveOptions() + saved_file, iteration, solved = ( + options.saved_file, + options.iteration, + options.solved, + ) + folder_name = create_maze_folder(solved=solved) if path is None: path = [] if visited is None: visited = [] - with open( - f"{folder_name}/{iteration}{os.path.basename(saved_file)}", "w", encoding="utf8" + with Path(f"{folder_name}/{iteration}{Path(saved_file).name}").open( + "w", encoding="utf8" ) as maze_file: for row_i, row in enumerate(maze): for col_i, cell in enumerate(row): @@ -229,25 +260,24 @@ def save_maze( maze_file.write("\n") -def fill_generated_maze(hor, ver, width): - """Fills generated maze array from horizontal and vertical lines.""" +def fill_generated_maze(hor: list[list[str]], ver: list[list[str]], width: int) -> str: + """Fill the generated maze array from its horizontal and vertical lines.""" maze_data = "" - for horizontal_line, vertical_line in zip(hor, ver): - maze_data += "".join(horizontal_line + ["\n"] + vertical_line + ["\n"]) + for horizontal_line, vertical_line in zip(hor, ver, strict=False): + maze_data += "".join([*horizontal_line, "\n", *vertical_line, "\n"]) maze_data_list = list(maze_data) maze_data_list[3 * width + 3] = "S" maze_data_list[len(maze_data_list) - (3 * width + 6)] = "E" - maze_data = "".join(maze_data_list) - return maze_data + return "".join(maze_data_list) -def make_maze(width=16, height=8): +def make_maze(width: int = 16, height: int = 8) -> str: """Generate maze with given width and height.""" vis = [[0] * width + [1] for _ in range(height)] + [[1] * (width + 1)] ver = [["# "] * width + ["#"] for _ in range(height)] + [[]] hor = [["###"] * width + ["#"] for _ in range(height + 1)] - def walk(x_coordinate, y_coordinate): + def walk(x_coordinate: int, y_coordinate: int) -> None: vis[y_coordinate][x_coordinate] = 1 neighbors = [ @@ -256,7 +286,7 @@ def make_maze(width=16, height=8): (x_coordinate + 1, y_coordinate), (x_coordinate, y_coordinate - 1), ] - shuffle(neighbors) + _RANDOM.shuffle(neighbors) for x_coordinate_neighbor, y_coordinate_neighbor in neighbors: if vis[y_coordinate_neighbor][x_coordinate_neighbor]: continue @@ -266,41 +296,38 @@ def make_maze(width=16, height=8): ver[y_coordinate][max(x_coordinate, x_coordinate_neighbor)] = " " walk(x_coordinate_neighbor, y_coordinate_neighbor) - walk(randrange(width), randrange(height)) + walk(_RANDOM.randrange(width), _RANDOM.randrange(height)) return fill_generated_maze(hor, ver, width) -def print_help(): - """Prints help.""" - print( - """python main.py - run the script against default maze file -(any file named maze.txt in the code directory) - -python main.py filename.txt - run the script against filename.txt file -python main.py -h --help print this prompt -python main.py -t --test non interactive (does not print steps) for testing -different heuristics, goes through entire generatedMazes folder and -compares heuristic speed and path length -python main.py -t --test [FOLDER] non interactive (does not print steps) for testing -different heuristics, goes through entire [FOLDER] folder and -compares heuristic speed and path length - -python main.py -g --generate [NUMBER] - generates as many mazes as entered in -Number parameter and puts it in the generatedMazes folder""" +def print_help() -> None: + """Print help.""" + logger.info( + "python main.py - run the script against default maze file\n(any file " + "named maze.txt in the code directory)\n\npython main.py filename.txt - " + "run the script against filename.txt file\npython main.py -h --help print" + " this prompt\npython main.py -t --test non interactive (does not print " + "steps) for testing\ndifferent heuristics, goes through entire " + "generatedMazes folder and\ncompares heuristic speed and path length\n" + "python main.py -t --test [FOLDER] non interactive (does not print steps)" + " for testing\ndifferent heuristics, goes through entire [FOLDER] folder " + "and\ncompares heuristic speed and path length\n\npython main.py -g " + "--generate [NUMBER] - generates as many mazes as entered in\nNumber " + "parameter and puts it in the generatedMazes folder" ) -def test_mode(): - """Loads and solves multiple mazes in order to compare heuristics.""" - create_maze_folder(False) +def test_mode() -> None: + """Load and solves multiple mazes in order to compare heuristics.""" + create_maze_folder(solved=False) sum_of_paths = 0 files_amount = 0 sum_of_time = 0 heuristic_total_total_time = 0 all_heuristic_called = 0 - for filename in os.listdir(FOLDER_NAME): - filename_directory = os.path.join(FOLDER_NAME, filename) + for filename in [q.name for q in Path(FOLDER_NAME).iterdir()]: + filename_directory = str(Path(FOLDER_NAME) / filename) # Open and load text file to array loaded_maze = load_maze(filename_directory) # Initialize MazeSolver object with maze as parameter @@ -315,25 +342,33 @@ def test_mode(): end_time = time.perf_counter() sum_of_time += end_time - start_time sum_of_paths += len(solved_path) - save_maze(loaded_maze, True, solved_path, visited, filename, 0) + save_maze( + loaded_maze, solved_path, visited, SaveOptions(filename, 0, solved=True) + ) files_amount += 1 if files_amount == 0: - print("no mazes found! Generate some using python main.py -g [NUMBER]") + logger.info("no mazes found! Generate some using python main.py -g [NUMBER]") sys.exit() average_path = sum_of_paths / files_amount average_time = sum_of_time / files_amount - print(f"""For: {files_amount} files, - sum of path lengths = {sum_of_paths}, - average path length = {average_path}, - sum_of_time = {sum_of_time}, - average time to solve: {average_time}, - heuristic_total_total_time: {heuristic_total_total_time}, - all_heuristic_called: {all_heuristic_called}, - average_heuristic_time: {heuristic_total_total_time / all_heuristic_called}""") + logger.info( + "For: %s files,\n sum of path lengths = %s,\n average path length =" + " %s,\n sum_of_time = %s,\n average time to solve: %s,\n " + "heuristic_total_total_time: %s,\n all_heuristic_called: %s,\n " + "average_heuristic_time: %s", + files_amount, + sum_of_paths, + average_path, + sum_of_time, + average_time, + heuristic_total_total_time, + all_heuristic_called, + heuristic_total_total_time / all_heuristic_called, + ) -def default(): - """Runs default operation - reads, solves and prints single maze from file.""" +def default() -> None: + """Run default operation - reads, solves and prints single maze from file.""" # Open and load text file to array loaded_maze = load_maze(FILE_NAME) # Initialize MazeSolver object with maze as parameter @@ -341,12 +376,12 @@ def default(): # Find path using MazeSolver solve method solved_path, visited, _, _ = solver.solve() print_maze(loaded_maze, solved_path, visited) - save_maze(loaded_maze, True, solved_path, visited, FILE_NAME, 0) + save_maze(loaded_maze, solved_path, visited, SaveOptions(FILE_NAME, 0, solved=True)) # Ran first in the code if __name__ == "__main__": - # print(sys.argv) + logging.basicConfig(format="%(message)s", level=logging.INFO) FILE_NAME = "maze.txt" TEST_MODE = False FOLDER_NAME = "" @@ -359,16 +394,20 @@ if __name__ == "__main__": TEST_MODE = True FILE_NAME = "maze.txt" FOLDER_NAME = "generatedMazes" - if len(sys.argv) > 2: + if len(sys.argv) > ARGV_WITH_OPTION: FOLDER_NAME = sys.argv[2] test_mode() sys.exit() - if sys.argv[1] == "-g" or sys.argv[1] == "--generate": - if len(sys.argv) > 2: - GENERATE_AMOUNT = int(sys.argv[2]) - for n in range(GENERATE_AMOUNT): - GENERATED_MAZE = make_maze() - save_maze(GENERATED_MAZE, False, None, None, f"generated{n}.txt") - sys.exit() + if sys.argv[1] in ("-g", "--generate") and len(sys.argv) > ARGV_WITH_OPTION: + GENERATE_AMOUNT = int(sys.argv[2]) + for n in range(GENERATE_AMOUNT): + GENERATED_MAZE = make_maze() + save_maze( + GENERATED_MAZE, + None, + None, + SaveOptions(f"generated{n}.txt", 0, solved=False), + ) + sys.exit() FILE_NAME = sys.argv[1] default() diff --git a/Programming/EARIN/lab2/main.py b/Programming/EARIN/lab2/main.py old mode 100644 new mode 100755 index 3fcf703d..72ad7a27 --- a/Programming/EARIN/lab2/main.py +++ b/Programming/EARIN/lab2/main.py @@ -1,20 +1,32 @@ -"""Program that plays draughts (checkers) with user on 8x8 board using min-max with alpha-beta pruning.""" +#!/usr/bin/env python3 +"""Play draughts against a min-max bot with alpha-beta pruning, on 8x8.""" import copy +import logging import math import re import sys +# A capture jumps exactly two squares, and the board is drawn four text rows +# per board row, with the rank label on the third of them. +CAPTURE_DISTANCE = 2 +ROWS_PER_SQUARE = 4 +LABEL_ROW = 2 +ARGV_WITH_COLOUR = 2 +ARGV_WITH_DEPTH = 3 +logger = logging.getLogger(__name__) + class Game: """Game.""" - def __init__(self, size): + def __init__(self, size: int) -> None: + """Set up an empty board of the given size and place both armies.""" self.board_size = size self.white_positions = self.initialize_white() self.black_positions = self.initialize_black() - def initialize_white(self): + def initialize_white(self) -> set[tuple[int, int, bool]]: """Initialize white pieces.""" white_positions = [] for y_coordinate in range(math.floor((self.board_size - 2) / 2)): @@ -26,7 +38,7 @@ class Game: white_positions.append((x_coordinate, y_coordinate, False)) return white_positions - def initialize_black(self): + def initialize_black(self) -> set[tuple[int, int, bool]]: """Initialize black pieces.""" black_positions = [] for y_coordinate in range( @@ -40,17 +52,13 @@ class Game: black_positions.append((x_coordinate, y_coordinate, False)) return black_positions - def check_move_out_of_bounds(self, to_): + def check_move_out_of_bounds(self, to_: tuple[int, int]) -> bool: """Check if the move destination is out of the bounds of the board.""" if to_[0] < 0 or to_[0] > self.board_size - 1: - # print(f"Illegal move! Final x coordinate must be between 0 and {self.board_size-1}!") return True - if to_[1] < 0 or to_[1] > self.board_size - 1: - # print(f"Illegal move! Final y coordinate must be between 0 and {self.board_size-1}!") - return True - return False + return bool(to_[1] < 0 or to_[1] > self.board_size - 1) - def check_piece_exists(self, coords, color): + def check_piece_exists(self, coords: tuple[int, int], color: str) -> bool: """Check if a piece of given color exists at a given spot.""" if color == "white": if any( @@ -65,8 +73,9 @@ class Game: return True return False - def check_piece_king(self, coords, color): + def check_piece_king(self, coords: tuple[int, int], color: str) -> bool: """Check if a piece of in a given spot and of a given color is a king. + Return false if no piece is found. """ if color == "white": @@ -76,7 +85,9 @@ class Game: return False # https://stackoverflow.com/a/2191707 - def check_move_piece_capable(self, from_, to_, color): + def check_move_piece_capable( + self, from_: tuple[int, int], to_: tuple[int, int], color: str + ) -> bool: """Check if the move is exactly one square diagonally.""" if abs(to_[0] - from_[0]) == 1: if self.check_piece_king(from_, color): @@ -87,13 +98,19 @@ class Game: return to_[1] == from_[1] - 1 return False - def check_capture(self, from_, to_, color): + def check_capture( + self, from_: tuple[int, int], to_: tuple[int, int], color: str + ) -> tuple[int, int] | None: """Check if a piece was captured for a given move. + Return captured piece coordinates or None. """ # captures can only happen if the player moved twice-diagonally - if abs(to_[0] - from_[0]) != 2 or abs(to_[1] - from_[1]) != 2: + if ( + abs(to_[0] - from_[0]) != CAPTURE_DISTANCE + or abs(to_[1] - from_[1]) != CAPTURE_DISTANCE + ): return None middle = (abs(to_[0] + from_[0]) // 2, abs(to_[1] + from_[1]) // 2) @@ -104,35 +121,52 @@ class Game: return middle return None - def check_move_legal(self, from_, to_, color, give_feedback=False): - """Check if a move is legal. Return a boolean or coordinates of captured piece.""" + def check_move_legal( + self, + from_: tuple[int, int], + to_: tuple[int, int], + color: str, + *, + give_feedback: bool = False, + ) -> bool | tuple[int, int]: + """Check whether a move is legal. + + Returns False when it is not, True when it is a plain move, and the + coordinates of the captured piece when it is a capture. + """ if self.check_move_out_of_bounds(to_): if give_feedback: - print("Illegal move! Final position is out of the bounds of the board") + logger.info( + "Illegal move! Final position is out of the bounds of the board" + ) return False if not self.check_piece_exists(from_, color): if give_feedback: - print( - "Illegal move! There is no piece on the " - "starting position that belongs to the player" + logger.info( + "Illegal move! There is no piece on the starting position " + "that belongs to the player" ) return False if self.check_piece_exists(to_, "white") or self.check_piece_exists( to_, "black" ): if give_feedback: - print("Illegal move! Cannot move to position taken by another piece") + logger.info( + "Illegal move! Cannot move to position taken by another piece" + ) return False capture = self.check_capture(from_, to_, color) if capture is None: if self.check_move_piece_capable(from_, to_, color): return True if give_feedback: - print("Illegal move! You can only move diagonally") + logger.info("Illegal move! You can only move diagonally") return False return capture - def make_move(self, from_, to_, color): + def make_move( + self, from_: tuple[int, int], to_: tuple[int, int], color: str + ) -> bool: """Move a piece on the board and remove any captured pieces.""" move_legal = self.check_move_legal(from_, to_, color) if move_legal is False: @@ -162,77 +196,81 @@ class Game: return True - def print_board(self, rotate=False): - """Print the board in the console.""" + def _print_letters(self, *, rotate: bool) -> None: + """Write the file letters that run above and below the board.""" + sys.stdout.write(" ") + for col in range(self.board_size): + letter = self.board_size - 1 - col if rotate else col + sys.stdout.write(f" {chr(ord('a') + letter)} ") - def print_letters(): - """Print the letters above or under the board.""" - print(" ", end="") - for col in range(self.board_size): - if rotate: - print(f" {chr(ord('a') + self.board_size - 1 - col)}", end=" ") - else: - print(f" {chr(ord('a') + col)}", end=" ") + def _square_code(self, pos: tuple[int, int], background: str) -> str: + """Return the one-character code of a square: piece, king, or empty.""" + if (*pos, False) in self.white_positions: + return "w" + if (*pos, True) in self.white_positions: + return "W" + if (*pos, False) in self.black_positions: + return "b" + if (*pos, True) in self.black_positions: + return "B" + return background - def get_square_code(pos, background): - """Return the code of a given square on the board.""" - if (*pos, False) in self.white_positions: - return "w" - if (*pos, True) in self.white_positions: - return "W" - if (*pos, False) in self.black_positions: - return "b" - if (*pos, True) in self.black_positions: - return "B" - return background + def _print_cell(self, row: int, col: int, line: int, *, rotate: bool) -> None: + """Write the one text row of one square.""" + background = "#" if (col % 2 == (row // 4) % 2) != rotate else " " + pos = (self.board_size - 1 - col, row // 4) if rotate else (col, row // 4) + checker = self._square_code(pos, background) - print_letters() - print(" ") + if col == 0: + sys.stdout.write( + f"{row // 4:3d}" if line % ROWS_PER_SQUARE == LABEL_ROW else " " + ) + + match line % ROWS_PER_SQUARE: + case 0: + sys.stdout.write("+---") + case 1 | 3: + sys.stdout.write(f"|{3 * background}") + case 2: + sys.stdout.write(f"|{background}{checker}{background}") + + def print_board(self, *, rotate: bool = False) -> None: + """Print the board in the console. + + The three helpers below used to be nested functions here, which is + what put this method at C901 17. + """ + self._print_letters(rotate=rotate) + sys.stdout.write(" \n") row_range = ( range(self.board_size * 4) if not rotate else reversed(range(self.board_size * 4)) ) - line = 0 - for row in row_range: + + for line, row in enumerate(row_range): for col in range(self.board_size): - background = "#" if (col % 2 == (row // 4) % 2) != rotate else " " - checker = ( - get_square_code((col, row // 4), background) - if not rotate - else get_square_code( - (self.board_size - 1 - col, row // 4), background - ) - ) - - if col == 0: - if line % 4 == 2: - print(f"{row // 4:3d}", end="") - else: - print(" ", end="") - - match line % 4: - case 0: - print("+---", end="") - case 1 | 3: - print(f"|{3 * background}", end="") - case 2: - print(f"|{background}{checker}{background}", end="") - if line % 4 == 0: - print("+") + self._print_cell(row, col, line, rotate=rotate) + if line % ROWS_PER_SQUARE == 0: + sys.stdout.write("+\n") else: - print(f"|{row // 4}" if line % 4 == 2 else "|") - line += 1 - print(" ", end="") - for col in range(self.board_size): - print("+---", end="") - print("+") - print_letters() - print() + sys.stdout.write( + f"|{row // ROWS_PER_SQUARE}\n" + if line % ROWS_PER_SQUARE == LABEL_ROW + else "|\n" + ) + sys.stdout.write(" ") + for _col in range(self.board_size): + sys.stdout.write("+---") + sys.stdout.write("+\n") + self._print_letters(rotate=rotate) + sys.stdout.write("\n") # Ran first in the code - def get_possible_moves_capture(self, from_, color): + def get_possible_moves_capture( + self, from_: tuple[int, int], color: str + ) -> list[tuple[tuple[int, int], tuple[int, int]]]: """Return all possible captures for a piece.""" # all capturing moves: legal_moves = [] @@ -250,7 +288,9 @@ class Game: legal_moves.append((from_, move_up_right_two)) return legal_moves - def get_possible_moves_non_capture(self, from_, color): + def get_possible_moves_non_capture( + self, from_: tuple[int, int], color: str + ) -> list[tuple[tuple[int, int], tuple[int, int]]]: """Return all possible moves that are not captures for a piece.""" # all non-capturing moves legal_moves = [] @@ -268,13 +308,14 @@ class Game: legal_moves.append((from_, move_up_right_one)) return legal_moves - def get_possible_moves(self, color): + def get_possible_moves( + self, color: str + ) -> tuple[list[tuple[tuple[int, int], tuple[int, int]]], bool]: """Return all possible moves for a given piece color.""" legal_moves = [] captures = [] if color == "white": for white_position in self.white_positions: - # print((white_position[0], white_position[1])) legal_moves += self.get_possible_moves_non_capture( (white_position[0], white_position[1]), color ) @@ -294,65 +335,68 @@ class Game: return (captures, True) return (legal_moves + captures, False) - def alpha_beta(self, depth, alpha_beta, color, current_color=None): - """Do alpha beta pruning for given parameters - and return the best move and its evaluated points. + def _search_branch( + self, + depth: int, + alpha_beta: tuple[float, float], + color: str, + current_color: str, + *, + maximising: bool, + ) -> tuple[float, tuple[tuple[int, int], tuple[int, int]] | None]: + """One half of the alpha-beta search. + + The maximising and minimising halves were byte-for-byte mirror images + of each other, which is what made alpha_beta too complex to pass C901. """ - if current_color is None: - current_color = color - - if depth == 0: - return self.evaluate(color), None - alpha, beta = alpha_beta opposite_color = "white" if current_color == "black" else "black" - if current_color == color: - max_eval = float("-inf") - best_move = None + best_eval = float("-inf") if maximising else float("inf") + best_move = None - for move in self.get_possible_moves(current_color)[0]: - new_state = copy.deepcopy(self) - new_state.make_move(*move, current_color) - eval_, _ = new_state.alpha_beta( - depth - 1, alpha_beta, color, opposite_color - ) + for move in self.get_possible_moves(current_color)[0]: + new_state = copy.deepcopy(self) + new_state.make_move(*move, current_color) + eval_, _ = new_state.alpha_beta( + depth - 1, alpha_beta, color, opposite_color + ) - if eval_ > max_eval: - max_eval = eval_ - best_move = move + if (eval_ > best_eval) if maximising else (eval_ < best_eval): + best_eval = eval_ + best_move = move + if maximising: alpha = max(alpha, eval_) - - if alpha >= beta: - break - - return max_eval, best_move - - if opposite_color == color: - min_eval = float("inf") - best_move = None - - for move in self.get_possible_moves(current_color)[0]: - new_state = copy.deepcopy(self) - new_state.make_move(*move, current_color) - eval_, _ = new_state.alpha_beta( - depth - 1, alpha_beta, color, opposite_color - ) - - if eval_ < min_eval: - min_eval = eval_ - best_move = move - + else: beta = min(beta, eval_) - if beta <= alpha: - break + if alpha >= beta: + break - return min_eval, best_move - return None + return best_eval, best_move - def evaluate(self, color): - """Evaluates the state of the board for a given color.""" + def alpha_beta( + self, + depth: int, + alpha_beta: tuple[float, float], + color: str, + current_color: str | None = None, + ) -> tuple[float, tuple[tuple[int, int], tuple[int, int]] | None]: + """Search with alpha-beta pruning; return the best move and its score.""" + if current_color is None: + current_color = color + if depth == 0: + return self.evaluate(color), None + return self._search_branch( + depth, + alpha_beta, + color, + current_color, + maximising=current_color == color, + ) + + def evaluate(self, color: str) -> float: + """Score the board from the given colour's point of view.""" white_score = 0 black_score = 0 @@ -372,13 +416,13 @@ class Game: white_score - black_score if color == "white" else black_score - white_score ) - def input_to_coordinates(self, user_input): + def input_to_coordinates(self, user_input: str) -> tuple[int, int]: """Change input from a1 form to tuple form.""" pos_x = ord(user_input[0]) - ord("a") pos_y = int(user_input[1::]) return pos_x, pos_y - def handle_player_move(self, color): + def handle_player_move(self, color: str) -> None: """Prompt player to move, validate their input and make move.""" has_moved = False possible_moves = self.get_possible_moves(color)[0] @@ -389,7 +433,7 @@ class Game: regex = r"^[a-z]\d+\s[a-z]\d+$" match = re.search(regex, user_input) if not match: - print("Invalid input, try again") + logger.info("Invalid input, try again") continue [move_from, move_to] = user_input.split(" ") from_coordinates = self.input_to_coordinates(move_from) @@ -397,19 +441,19 @@ class Game: if (from_coordinates, to_coordinates) not in possible_moves: legal_no_captures = self.check_move_legal( - from_coordinates, to_coordinates, color, True + from_coordinates, to_coordinates, color, give_feedback=True ) if legal_no_captures: - print("Invalid move! You can capture a piece") + logger.info("Invalid move! You can capture a piece") continue has_moved = self.make_move(from_coordinates, to_coordinates, color) self.print_board(color == "white") - def start_game(self, player_color="black", algorithm_depth=5): + def start_game(self, player_color: str = "black", algorithm_depth: int = 5) -> None: """Start the main loop of the game.""" if player_color not in ("black", "white"): - print("Invalid color! Color can be black or white") + logger.info("Invalid color! Color can be black or white") return ai_color = "black" if player_color == "white" else "white" @@ -421,14 +465,16 @@ class Game: while ai_turn: possible_moves_ai = game.get_possible_moves(ai_color) if len(possible_moves_ai[0]) == 0: - print(f"Game over, {player_color} wins") + logger.info("Game over, %s wins", player_color) return _, ai_move = game.alpha_beta(algorithm_depth, (5, 10), ai_color) game.make_move(*ai_move, ai_color) - print( - "AI's move: " - f"{chr(ord('a') + ai_move[0][0])}{ai_move[0][1]} " - f"{chr(ord('a') + ai_move[1][0])}{ai_move[1][1]}" + logger.info( + "AI's move: %s%s %s%s", + chr(ord("a") + ai_move[0][0]), + ai_move[0][1], + chr(ord("a") + ai_move[1][0]), + ai_move[1][1], ) game.print_board(player_color == "white") ai_turn = game.get_possible_moves(ai_color)[1] and possible_moves_ai[1] @@ -436,7 +482,7 @@ class Game: while player_turn: possible_moves_player = game.get_possible_moves(player_color) if len(possible_moves_player[0]) == 0: - print(f"Game over, {ai_color} wins") + logger.info("Game over, %s wins", ai_color) return game.handle_player_move(player_color) player_turn = ( @@ -444,26 +490,35 @@ class Game: and possible_moves_player[1] ) - def ai_turn(self, ai_color, algorithm_depth, possible_moves_ai, print_info): - """Calculates ai move and makes it.""" + def ai_turn( + self, + ai_color: str, + algorithm_depth: int, + possible_moves_ai: list[object], + *, + print_info: bool = True, + ) -> bool: + """Calculate ai move and makes it.""" if len(possible_moves_ai) == 0: if print_info: - print(f"Game over, {ai_color} loses") + logger.info("Game over, %s loses", ai_color) return True _, ai_move = game.alpha_beta(algorithm_depth, (5, 10), ai_color) if ai_move is None: ai_move = possible_moves_ai[0] game.make_move(*ai_move, ai_color) if print_info: - print( - "AI's move: " - f"{chr(ord('a') + ai_move[0][0])}{ai_move[0][1]} " - f"{chr(ord('a') + ai_move[1][0])}{ai_move[1][1]}" + logger.info( + "AI's move: %s%s %s%s", + chr(ord("a") + ai_move[0][0]), + ai_move[0][1], + chr(ord("a") + ai_move[1][0]), + ai_move[1][1], ) - game.print_board(True) + game.print_board(rotate=True) return False - def auto_game(self, white_depth, black_depth): + def auto_game(self, white_depth: int, black_depth: int) -> str: """Auto game mode between two bots.""" game_turn = 0 max_turns = 250 @@ -471,7 +526,9 @@ class Game: bot_white_turn = True while bot_white_turn: possible_moves_ai = game.get_possible_moves("white") - if self.ai_turn("white", white_depth, possible_moves_ai[0], False): + if self.ai_turn( + "white", white_depth, possible_moves_ai[0], print_info=False + ): return "white" bot_white_turn = ( game.get_possible_moves("white")[1] and possible_moves_ai[1] @@ -479,23 +536,27 @@ class Game: bot_black_turn = True while bot_black_turn: possible_moves_ai = game.get_possible_moves("black") - if self.ai_turn("black", black_depth, possible_moves_ai[0], False): + if self.ai_turn( + "black", black_depth, possible_moves_ai[0], print_info=False + ): return "black" bot_black_turn = ( game.get_possible_moves("black")[1] and possible_moves_ai[1] ) game_turn += 1 if game_turn >= max_turns: - print(f"Game ended after {max_turns} turns!") + logger.info("Game ended after %s turns!", max_turns) return "" return "" -def auto_simulation(white_depth, black_depth, iterations): - """Runs iterations amount of simulations.""" - print( - f"""Running {iterations} simulations with - white depth = {white_depth},black depth = {black_depth}""" +def auto_simulation(white_depth: int, black_depth: int, iterations: int) -> None: + """Run iterations amount of simulations.""" + logger.info( + "Running %s simulations with\n white depth = %s,black depth = %s", + iterations, + white_depth, + black_depth, ) white_wins = 0 black_wins = 0 @@ -513,40 +574,42 @@ def auto_simulation(white_depth, black_depth, iterations): white_pieces_captured += 16 - len(game.white_positions) black_pieces_captured += 16 - len(game.black_positions) current_iteration += 1 - print( - f"""White wins = {white_wins}, Black wins = {black_wins}, - white pieces captured in total = {white_pieces_captured}, - black pieces captured in total = {black_pieces_captured}""" + logger.info( + "White wins = %s, Black wins = %s,\n white pieces captured in " + "total = %s,\n black pieces captured in total = %s", + white_wins, + black_wins, + white_pieces_captured, + black_pieces_captured, ) - print() + logger.info("") -def print_help(): - """Prints help.""" - print( - """python main.py [algorithm_depth] - play the game against the bot as black, - if no algorithm depth is specified the default (5) will be set - -python main.py -h --help print this prompt -python main.py -t --test [max_white_depth] [max_black_depth] non interactive -(does not print moves) for testing how different bot depth play against eachother, -if depths are not provided default value of 5 is set -compares heuristic speed and path length -python main.py -w --white [algorithm_depth] play as white pieces, -if no algorithm depth is specified the default (5) will be set -python main.py -b --black [algorithm_depth] play as black pieces, -if no algorithm depth is specified the default (5) will be set -""" +def print_help() -> None: + """Print help.""" + logger.info( + "python main.py [algorithm_depth] - play the game against the bot as " + "black,\n if no algorithm depth is specified the default (5) will " + "be set\n\npython main.py -h --help print this prompt\npython main.py -t " + "--test [max_white_depth] [max_black_depth] non interactive\n(does not " + "print moves) for testing how different bot depth play against eachother," + "\nif depths are not provided default value of 5 is set\ncompares " + "heuristic speed and path length\npython main.py -w --white " + "[algorithm_depth] play as white pieces,\nif no algorithm depth is " + "specified the default (5) will be set\npython main.py -b --black " + "[algorithm_depth] play as black pieces,\nif no algorithm depth is " + "specified the default (5) will be set\n" ) -def default(color="black", algorithm_depth=5): - """Default program function -> allows to play a game against bot (by default as black).""" +def default(color: str = "black", algorithm_depth: int = 5) -> None: + """Play one game against the bot, as black by default.""" game.start_game(color, algorithm_depth) # Ran first in the code if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) game = Game(8) if len(sys.argv) > 1: if sys.argv[1] == "-h" or sys.argv[1] == "--help": @@ -555,9 +618,9 @@ if __name__ == "__main__": if sys.argv[1] == "-t" or sys.argv[1] == "--test": MAX_WHITE_DEPTH = 4 MAX_BLACK_DEPTH = 4 - if len(sys.argv) > 2: + if len(sys.argv) > ARGV_WITH_COLOUR: MAX_WHITE_DEPTH = int(sys.argv[2]) - if len(sys.argv) > 3: + if len(sys.argv) > ARGV_WITH_DEPTH: MAX_BLACK_DEPTH = int(sys.argv[3]) for i in range(MAX_WHITE_DEPTH + 1): for j in range(MAX_BLACK_DEPTH + 1): @@ -566,13 +629,13 @@ if __name__ == "__main__": sys.exit() if sys.argv[1] == "-w" or sys.argv[1] == "--white": ALGORITHM_DEPTH = 5 - if len(sys.argv) > 2: + if len(sys.argv) > ARGV_WITH_COLOUR: ALGORITHM_DEPTH = int(sys.argv[2]) default("white", ALGORITHM_DEPTH) sys.exit() if sys.argv[1] == "-b" or sys.argv[1] == "--black": ALGORITHM_DEPTH = 5 - if len(sys.argv) > 2: + if len(sys.argv) > ARGV_WITH_COLOUR: ALGORITHM_DEPTH = int(sys.argv[2]) default("black", ALGORITHM_DEPTH) sys.exit() diff --git a/Programming/EARIN/lab3/example.py b/Programming/EARIN/lab3/example.py old mode 100644 new mode 100755 index 796ee2a1..4e4d5ec6 --- a/Programming/EARIN/lab3/example.py +++ b/Programming/EARIN/lab3/example.py @@ -1,24 +1,38 @@ -import os +#!/usr/bin/env python3 +"""Matplotlib scatter demo the lab-3 plotting is modelled on.""" + +import logging +import sys import tempfile +from pathlib import Path import cv2 import matplotlib.pyplot as plt import numpy as np +logger = logging.getLogger(__name__) -def main(): +# cv2.waitKey returns the ASCII code of the key pressed; 'q' quits. +KEY_Q = ord("q") +# NumPy's legacy np.random.* functions share one global state; Generator is +# the modern API and is what NPY002 asks for. +RNG = np.random.default_rng() + + +def main() -> None: # define number of data points - N = 10 + """Draw the scatter demo this lab's plotting is based on.""" + point_count = 10 # define the visualization params - colors = np.random.rand(N) + colors = RNG.random(point_count) with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f: # iterate over the optimization steps - for i in range(10): + for _i in range(10): # generate random 2D data - replace it with the results from your algorithm - x = np.random.rand(N) - y = np.random.rand(N) + x = RNG.random(point_count) + y = RNG.random(point_count) # plot the data plt.cla() @@ -34,17 +48,18 @@ def main(): # show the image, provide window name first cv2.imshow("visualization", image) - # add wait key. window waits until user presses a key and quits if the key is 'q' - if cv2.waitKey(0) == 113: + # add wait key. window waits until user presses a key and quits if the key + # is 'q' + if cv2.waitKey(0) == KEY_Q: # and finally destroy/close all open windows cv2.destroyAllWindows() - exit() + sys.exit() try: f.close() - os.unlink(f.name) - except: - pass + Path(f.name).unlink() + except Exception: + logger.exception("could not close the preview window") cv2.destroyAllWindows() diff --git a/Programming/EARIN/lab3/main.py b/Programming/EARIN/lab3/main.py old mode 100644 new mode 100755 index cba4a3c1..1efcdd72 --- a/Programming/EARIN/lab3/main.py +++ b/Programming/EARIN/lab3/main.py @@ -1,19 +1,30 @@ -"""Program that optimizes Rastrigin function: file_ (x_point_value, y_point_value) = +#!/usr/bin/env python3 +"""Program that optimizes Rastrigin function: file_ (x_point_value, y_point_value) =. + 20 + (x_point_value^2 - 10cos(2πx)) + (y_point_value^2 - 10 cos(2πy)). Using Evolutionary Strategy (μ, λ). """ -import os +import logging import sys import tempfile import time +from collections.abc import Callable +from pathlib import Path import cv2 import matplotlib.pyplot as plt import numpy as np +# cv2.waitKey returns the ASCII code of the key pressed; 'q' quits. +KEY_Q = ord("q") +# NumPy's legacy np.random.* functions share one global state; Generator is +# the modern API and is what NPY002 asks for. +RNG = np.random.default_rng() +logger = logging.getLogger(__name__) -def rastrigin(x_argument, y_argument): + +def rastrigin(x_argument: float, y_argument: float) -> float: """Define the Rastrigin function.""" return ( 20 @@ -24,7 +35,7 @@ def rastrigin(x_argument, y_argument): ) -def generate(population, arguments): +def generate(population: object, arguments: dict[str, float]) -> tuple[object, object]: """Run single generation.""" # Evaluate the fitness of each individual fitness = np.array( @@ -40,7 +51,7 @@ def generate(population, arguments): # Generate the next generation of lambda individuals by recombination children = np.concatenate( [ - np.random.permutation(parents) + RNG.permutation(parents) for i in range( (arguments["size_of_population"] // arguments["number_of_parents"]) + 1 ) @@ -49,7 +60,7 @@ def generate(population, arguments): children = children[: arguments["size_of_population"]] # Add mutation to the children - mutation = np.random.normal( + mutation = RNG.normal( loc=0, scale=arguments["mutation_strength"], size=(arguments["size_of_population"], 2), @@ -59,13 +70,12 @@ def generate(population, arguments): def evolution_strategy( - arguments, - no_display=False, -): + arguments: dict[str, float], *, no_display: bool = False +) -> tuple[object, object]: """Define the Evolutionary Strategy (μ, λ) algorithm.""" # Initialize the population print_info = [] - population = np.random.uniform( + population = RNG.uniform( low=arguments["min"], high=arguments["max"], size=(arguments["size_of_population"], 2), @@ -79,7 +89,8 @@ def evolution_strategy( 0, f"""0:nop-{arguments["number_of_parents"]}:sop-{arguments["size_of_population"]}: ms-{arguments["mutation_strength"]}:nog-{arguments["number_of_generations"]}: - min-max-{arguments["min"], arguments["max"]}:noo-{arguments["number_of_outputs"]}""", + min-max-{arguments["min"], arguments["max"]}: + noo-{arguments["number_of_outputs"]}""", ) ) arguments["number_of_outputs"] = min( @@ -129,46 +140,37 @@ def evolution_strategy( ) -def print_help(): +def print_help() -> None: """Print program functionality and how to access it.""" - print( - """ - python main.py - Default functionality optimizing - Rastrigin function file_ (x_point_value, y_point_value) = - 20 + (x_point_value^2 - 10cos(2πx)) + (y_point_value^2 - 10 cos(2πy)) - using Evolutionary Strategy (μ, λ), using only default values - Default values: - arguments["number_of_parents"]=5, - arguments["size_of_population"]=20, - arguments["mutation_strength"]=0.1, - arguments["number_of_generations"]=100, - min_value=-5.12, - max_value=5.12 - arguments["number_of_outputs"] = 100 - - python main.py -h --help print this prompt - Any of the default values an be changed using arguments: - -nop --arguments["number_of_parents"] [number] - -sop --arguments["size_of_population"] [number] - -ms --arguments["mutation_strength"] [number] - -nog --arguments["number_of_generations"] [number] - -min --min_value [number] - -max --max_value [number] - -noo, --arguments["number_of_outputs"] [number] - Those arguments can be given in any order and any argument - which was not entered will be replaced with default value, - Additional flags: - -nd, --no-display (does not show the plots) - -s, --save (if issued WILL save the files) - exemplary use: - python main.py -nop 5 -sop 20 -ms 0.1 -nog 100 -min -5.12 -max 5.12 -noo 100 - """ + logger.info( + "\n python main.py - Default functionality optimizing\n Rastrigin " + "function file_ (x_point_value, y_point_value) =\n 20 + " + "(x_point_value^2 - 10cos(2πx)) + (y_point_value^2 - 10 cos(2πy))\n " + "using Evolutionary Strategy (μ, λ), using only default values\n " + 'Default values:\n arguments["number_of_parents"]=5,\n ' + 'arguments["size_of_population"]=20,\n ' + 'arguments["mutation_strength"]=0.1,\n ' + 'arguments["number_of_generations"]=100,\n min_value=-5.12,\n ' + 'max_value=5.12\n arguments["number_of_outputs"] = 100\n\n python ' + "main.py -h --help print this prompt\n Any of the default values an be" + ' changed using arguments:\n -nop --arguments["number_of_parents"] ' + '[number]\n -sop --arguments["size_of_population"] [number]\n -ms ' + '--arguments["mutation_strength"] [number]\n -nog ' + '--arguments["number_of_generations"] [number]\n -min --min_value ' + "[number]\n -max --max_value [number]\n -noo, " + '--arguments["number_of_outputs"] [number]\n Those arguments can be ' + "given in any order and any argument\n which was not entered will be " + "replaced with default value,\n Additional flags:\n -nd, " + "--no-display (does not show the plots)\n -s, --save (if issued WILL " + "save the files)\n exemplary use:\n python main.py -nop 5 -sop 20 " + "-ms 0.1 -nog 100 -min -5.12 -max 5.12 -noo 100\n " ) -def get_output_bounds(x_data, y_data): +def get_output_bounds( + x_data: object, y_data: object +) -> tuple[tuple[float, float], tuple[float, float]]: """Get x and y output limits for pyplot.""" - # min_size = 0.2 min_output_size = ARGUMENTS["mutation_strength"] * 10 xmin = min(x_data) @@ -194,10 +196,16 @@ def get_output_bounds(x_data, y_data): return x_bounds, y_bounds -def output(population_output, generation_number, file_name="temp", save_results=False): +def output( + population_output: object, + generation_number: int, + file_name: str = "temp", + *, + save_results: bool = False, +) -> None: """Draw result of our function.""" # define the visualization params - colors = np.random.rand(len(population_output)) + colors = RNG.random(len(population_output)) with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as file_: # iterate over the optimization steps @@ -226,17 +234,19 @@ def output(population_output, generation_number, file_name="temp", save_results= cv2.imwrite(file_name + ".jpg", image) # add wait key. window waits until user presses a key and quits if # the key is 'q' - if cv2.waitKey(0) == 113: + if cv2.waitKey(0) == KEY_Q: # and finally destroy/close all open windows sys.exit() cv2.destroyAllWindows() file_.close() - os.unlink(file_.name) + Path(file_.name).unlink() -def print_summary(populations, file_name="temp_summary", save_results=False): +def print_summary( + populations: object, file_name: str = "temp_summary", *, save_results: bool = False +) -> None: """Draw result of our function for chosen generations.""" with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as file_: # iterate over the optimization steps @@ -272,19 +282,35 @@ def print_summary(populations, file_name="temp_summary", save_results=False): # add wait key. window waits until user presses a key and quits if # the key is 'q' - if cv2.waitKey(0) == 113: + if cv2.waitKey(0) == KEY_Q: # and finally destroy/close all open windows sys.exit() cv2.destroyAllWindows() file_.close() - os.unlink(file_.name) + Path(file_.name).unlink() -def user_input(): +# Long option, short option, and how to read the value that follows. A table +# rather than an if-ladder: the ladder was C901 12, and one row per flag makes +# adding one a data change. +_OPTIONS: dict[tuple[str, str], tuple[str, Callable[[str], float] | None]] = { + ("-nop", "--number_of_parents"): ("number_of_parents", int), + ("-sop", "--size_of_population"): ("size_of_population", int), + ("-ms", "--mutation_strength"): ("mutation_strength", float), + ("-nog", "--number_of_generations"): ("number_of_generations", int), + ("-min", "--min_value"): ("min", float), + ("-max", "--max_value"): ("max", float), + ("-noo", "--number_of_outputs"): ("number_of_outputs", int), + ("-nd", "--no_display"): ("no_display", None), + ("-s", "--save"): ("save", None), +} + + +def user_input() -> dict[str, float]: """Handle user terminal arguments.""" - arguments = { + arguments: dict[str, float] = { "number_of_parents": 5, "size_of_population": 20, "mutation_strength": 0.1, @@ -299,30 +325,15 @@ def user_input(): if argument in ("-h", "--help"): print_help() sys.exit() - if argument in ("-nop", "--number_of_parents"): - arguments["number_of_parents"] = int(sys.argv[index + 1]) - if argument in ("-sop", "--size_of_population"): - arguments["size_of_population"] = int(sys.argv[index + 1]) - if argument in ("-ms", "--mutation_strength"): - arguments["mutation_strength"] = float(sys.argv[index + 1]) - if argument in ("-nog", "--number_of_generations"): - arguments["number_of_generations"] = int(sys.argv[index + 1]) - if argument in ("-min", "--min_value"): - arguments["min"] = float(sys.argv[index + 1]) - if argument in ("-max", "--max_value"): - arguments["max"] = float(sys.argv[index + 1]) - if argument in ("-noo", "--number_of_outputs"): - arguments["number_of_outputs"] = int(sys.argv[index + 1]) - if argument in ("-nd", "--no_display"): - arguments["no_display"] = True - if argument in ("-s", "--save"): - arguments["save"] = True + for flags, (key, parse) in _OPTIONS.items(): + if argument in flags: + arguments[key] = True if parse is None else parse(sys.argv[index + 1]) return arguments -def print_output(print_info, save_results, summary): - """Prints out population and summary plots.""" +def print_output(print_info: object, save_results: object, summary: object) -> None: + """Print out population and summary plots.""" for population, generation_number, file_name in print_info: output(population, generation_number, file_name, save_results) summary_file_name = file_name @@ -332,6 +343,7 @@ def print_output(print_info, save_results, summary): # Ran first in the code if __name__ == "__main__": # Run the Evolutionary Strategy algorithm + logging.basicConfig(format="%(message)s", level=logging.INFO) ARGUMENTS = user_input() TOTAL_TIME = 0 start_time = time.perf_counter() @@ -351,7 +363,7 @@ if __name__ == "__main__": TOTAL_TIME = end_time - start_time time_per_generation = TOTAL_TIME / ARGUMENTS["number_of_generations"] - print("Best individual found:", best_individual) - print("Best fitness found:", best_fitness) - print("total_generation_time: ", TOTAL_TIME) - print("time_per_generation: ", time_per_generation) + logger.info("Best individual found: %s", best_individual) + logger.info("Best fitness found: %s", best_fitness) + logger.info("total_generation_time: %s", TOTAL_TIME) + logger.info("time_per_generation: %s", time_per_generation) diff --git a/Programming/EARIN/lab4/main.py b/Programming/EARIN/lab4/main.py old mode 100644 new mode 100755 index fcf5ef68..6bc634d5 --- a/Programming/EARIN/lab4/main.py +++ b/Programming/EARIN/lab4/main.py @@ -1,5 +1,8 @@ +#!/usr/bin/env python3 """Program that predicts wine quality based on variant2.csv data.""" +import logging + import matplotlib.pyplot as plt import numpy as np import pandas as pd @@ -9,14 +12,18 @@ from sklearn.metrics import accuracy_score, f1_score, mean_squared_error from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler +logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) + class LinearRegression: """Implements Linear regression method.""" - def __init__(self): + def __init__(self) -> None: + """Start with no fitted coefficients.""" self.theta = None - def fit(self, x_values, y_values): + def fit(self, x_values: object, y_values: object) -> None: """Fit linear regression model to our training data.""" # Add a column of ones to X for the intercept term x_values = np.concatenate((np.ones((x_values.shape[0], 1)), y_values), axis=1) @@ -26,24 +33,29 @@ class LinearRegression: np.linalg.inv(x_values.T.dot(x_values)).dot(x_values.T).dot(y_values) ) - def predict(self, x_values): - """Predict target values for our input data using the trained linear regression model.""" + def predict(self, x_values: object) -> object: + """Predict the target values for the given inputs. + + Uses the trained linear regression + + model. + """ # Add a column of ones to X for the intercept term x_values = np.concatenate((np.ones((x_values.shape[0], 1)), x_values), axis=1) # Make predictions using the learned weights - y_predicted = x_values.dot(self.theta) + return x_values.dot(self.theta) - return y_predicted + def score(self, x_values: object, y_values: object) -> float: + """Compute the R-squared score of the model. - def score(self, x_values, y_values): - """Compute the R-squared score of the linear regression model on our test data.""" + Measured on our test" + " data. + """ y_predicted = self.predict(x_values) ss_res = np.sum((y_values - y_predicted) ** 2) ss_tot = np.sum((y_values - np.mean(y_values)) ** 2) - r2_score = 1 - (ss_res / ss_tot) - - return r2_score + return 1 - (ss_res / ss_tot) wine_df = pd.read_csv("variant2.csv") @@ -52,8 +64,8 @@ wine_df.describe() wine_df.info() -X = wine_df.iloc[:, :-1].values -y = wine_df.iloc[:, -1].values +X = wine_df.iloc[:, :-1].to_numpy() +y = wine_df.iloc[:, -1].to_numpy() X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0) @@ -66,34 +78,34 @@ regressor.fit(X_train, y_train) y_pred = regressor.predict(X_test) mse = mean_squared_error(y_test, y_pred) -print("MSE:", mse) +logger.info("MSE: %s", mse) classifier = LogisticRegression() classifier.fit(X_train, y_train) y_pred = classifier.predict(X_test) accuracy = accuracy_score(y_test, y_pred) -print("Accuracy:", accuracy) +logger.info("Accuracy: %s", accuracy) y_pred_train = regressor.predict(X_train) train_mse = mean_squared_error(y_train, y_pred_train) -print("Training MSE:", train_mse) +logger.info("Training MSE: %s", train_mse) train_r_squared = regressor.score(X_train, y_train) -print("Training R^2:", train_r_squared) +logger.info("Training R^2: %s", train_r_squared) test_r_squared = regressor.score(X_test, y_test) -print("Testing R^2:", test_r_squared) +logger.info("Testing R^2: %s", test_r_squared) y_pred_train = classifier.predict(X_train) train_accuracy = accuracy_score(y_train, y_pred_train) -print("Training Accuracy:", train_accuracy) +logger.info("Training Accuracy: %s", train_accuracy) train_f1_score = f1_score(y_train, y_pred_train, average="weighted") -print("Training F1 Score:", train_f1_score) +logger.info("Training F1 Score: %s", train_f1_score) test_f1_score = f1_score(y_test, y_pred, average="weighted") -print("Testing F1 Score:", test_f1_score) +logger.info("Testing F1 Score: %s", test_f1_score) Data1 = sns.countplot(x="quality", data=wine_df) plt.draw() diff --git a/Programming/EARIN/lab5/code/example.py b/Programming/EARIN/lab5/code/example.py old mode 100644 new mode 100755 index 491d3668..463b6930 --- a/Programming/EARIN/lab5/code/example.py +++ b/Programming/EARIN/lab5/code/example.py @@ -1,79 +1,109 @@ -# numpy for loading dataset +#!/usr/bin/env python3 +"""Minimal PyTorch training example shipped with lab 5.""" + +import logging + import numpy as np - -# pytorch for deep learning models import torch - -# nn like neural network from torch import nn, optim -# Pima indians describes patient medical data and whether they had diabetes for last 5 years +# numpy for loading dataset + + +# The output layer is a sigmoid, so 0.5 is the class boundary. +DECISION_THRESHOLD = 0.5 +# pytorch for deep learning models + +# nn like neural network + +logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) + +# Pima indians describes patient medical data and whether they had diabetes for last 5 +# years # It is binary classification (they could either have diabetes 1 or not 0) # load the file as a matrix of numbers, dataset = np.loadtxt("pima-indians-diabetes.csv", delimiter=",") input_columns = 8 -# split into input (X) -> in this case everything beside info whether patient had diabetes or not is input -# We are splitting data into two subsets by using NumPy slice operator : and choose first 8 columns using 0:8 slice +# split into input (X) -> in this case everything beside info whether patient had +# diabetes or not is input +# We are splitting data into two subsets by using NumPy slice operator : and choose +# first 8 columns using 0:8 slice X = dataset[:, 0:input_columns] -# and output (y) variables -> in this case we are only interested whether patient had diabetes or not as an output +# and output (y) variables -> in this case we are only interested whether patient had +# diabetes or not as an output # you can simplify that y = f(X) # We are splitting the data by using slice operator : and choosing last column y = dataset[:, input_columns] # we need to convert this data to pytorch tensors -# Pytorch usually operates on 32-bit floating point and NumPy by default uses 64 bit floating point +# Pytorch usually operates on 32-bit floating point and NumPy by default uses 64 bit +# floating point X = torch.tensor(X, dtype=torch.float32) -# We can also correct the shape to fit what PyTorch would expect (here we are converting n vectors to n x 1 matrix) -# This simplifies handling matrix multiplication operations (which are the basis of deep learning models) -# reshape is converting the output variable y from a 1-dimensional NumPy array to a 2-dimensional PyTorch tensor with a shape of (n, 1), where n is the number of samples in the dataset. +# We can also correct the shape to fit what PyTorch would expect (here we are converting +# n vectors to n x 1 matrix) +# This simplifies handling matrix multiplication operations (which are the basis of deep +# learning models) +# reshape is converting the output variable y from a 1-dimensional NumPy array to a +# 2-dimensional PyTorch tensor with a shape of (n, 1), where n is the number of samples +# in the dataset. y = torch.tensor(y, dtype=torch.float32).reshape(-1, 1) # define the model -# this class is a subclass of nn.Module -> base class provided by PyTorch for building neural network models. +# this class is a subclass of nn.Module -> base class provided by PyTorch for building +# neural network models. class PimaClassifier(nn.Module): - def __init__(self): + """Two hidden layers and a sigmoid output, for the Pima dataset.""" + + def __init__(self) -> None: + """Build the two-layer network used by this example.""" super().__init__() - # There are 3 (fully connected) layers in class, each with their activation function + # There are 3 (fully connected) layers in class, each with their activation + # function # creates Linear layer, it maps input to a hidden layer of 12 neurons - # input features have a size of 8 (same number as number of features in pima indians diabetes dataset) + # input features have a size of 8 (same number as number of features in pima + # indians diabetes dataset) first_output_neurons = 12 self.hidden1 = nn.Linear(input_columns, first_output_neurons) - # This creates ReLU (rectified linear unit) activation function applied after first hidden layer + # This creates ReLU (rectified linear unit) activation function applied after + # first hidden layer self.act1 = nn.ReLU() - # This maps the output of first layer (which was 12 neurons) to new hidden layer of 8 neurons + # This maps the output of first layer (which was 12 neurons) to new hidden layer + # of 8 neurons second_output_neurons = 8 self.hidden2 = nn.Linear(first_output_neurons, second_output_neurons) # ReLU activation function applied after second hidden layer self.act2 = nn.ReLU() - # We map output of second layer to a single output neuron -> which will represent the predicted + # We map output of second layer to a single output neuron -> which will + # represent the predicted # probability of a sample having diabetes self.output = nn.Linear(second_output_neurons, 1) # sigmoid function forces output to be either 0 or 1 self.act_output = nn.Sigmoid() # forward pass is computation of output based on input 'x' - def forward(self, x): + def forward(self, x: object) -> object: + """Run one forward pass over the batch.""" # Applies first hidden layer (and then ReLU activation) to input x x = self.act1(self.hidden1(x)) # Applies second hidden layer (and then ReLU activation) to input x x = self.act2(self.hidden2(x)) # Applies output layer (and then Sigmoid activation) to input x - x = self.act_output(self.output(x)) - # returns final output (0 or 1) - return x + return self.act_output(self.output(x)) # Create object from model class model = PimaClassifier() -print(model) +logger.info("%s", model) # train the model # first we need to specify what is the goal of training # we have input X and output y and we want the model to be as close to y as possible -# Since this is binary classification problem we will use "binary cross entropy" to measure the distance between +# Since this is binary classification problem we will use "binary cross entropy" to +# measure the distance between # our prediction and y loss_fn = nn.BCELoss() # binary cross entropy # Optimizer adjust model weights to produce better output @@ -87,14 +117,18 @@ optimizer = optim.Adam(model.parameters(), lr=0.001) # epoch is the entire training dataset passed to a model once n_epochs = 100 # batch is one or more sample passed to model -# number of epochs and the size of a batch can be chosen experimentally by trial and error. -# a lot of epochs and big size of batch means more time and more memory consumption but more accurate results +# number of epochs and the size of a batch can be chosen experimentally by trial and +# error. +# a lot of epochs and big size of batch means more time and more memory consumption but +# more accurate results batch_size = 10 -# We split dataset into batches and pass batches one by one into a model to training loop -# after using all batches we finish one epoch and can start over again to refine the model +# We split dataset into batches and pass batches one by one into a model to training +# loop +# after using all batches we finish one epoch and can start over again to refine the +# model # we use two nested for loops for training, one is for epochs -for epoch in range(n_epochs): +for _epoch in range(n_epochs): # and one for batches for i in range(0, len(X), batch_size): # Split X data into a batch with the size from batch_size @@ -115,9 +149,9 @@ for epoch in range(n_epochs): # compute final accuracy y_pred = model(X) accuracy = (y_pred.round() == y).float().mean() -print(f"Accuracy {accuracy}") +logger.info("Accuracy %s", accuracy) # make class predictions with the model -predictions = (model(X) > 0.5).int() +predictions = (model(X) > DECISION_THRESHOLD).int() for i in range(5): - print("%s => %d (expected %d)" % (X[i].tolist(), predictions[i], y[i])) + logger.info("%s => %d (expected %d)", X[i].tolist(), predictions[i], y[i]) diff --git a/Programming/EARIN/lab5/code/main.py b/Programming/EARIN/lab5/code/main.py old mode 100644 new mode 100755 index 1c8a2b24..ddf30506 --- a/Programming/EARIN/lab5/code/main.py +++ b/Programming/EARIN/lab5/code/main.py @@ -1,15 +1,20 @@ +#!/usr/bin/env python3 """Implementation of a network analyzing MNIST dataset.""" +import logging import time +from pathlib import Path import matplotlib.pyplot as plt import torch from torch import nn, optim from torchvision import datasets, transforms +logger = logging.getLogger(__name__) -def set_hyperparameters(): - """Sets hyperparameters used throughout the network.""" + +def set_hyperparameters() -> dict[str, int]: + """Set hyperparameters used throughout the network.""" return { "num_epochs": 5, "init_input_size": 28 * 28, # MNIST images are 28x28 pixels @@ -17,8 +22,8 @@ def set_hyperparameters(): } -def load_datasets(): - """Loads train and test dataset from MNIST.""" +def load_datasets() -> tuple[object, object]: + """Load train and test dataset from MNIST.""" train_dataset = datasets.MNIST( root="./data", train=True, transform=transforms.ToTensor(), download=True ) @@ -28,7 +33,9 @@ def load_datasets(): return train_dataset, test_dataset -def create_data_loaders(train_dataset, test_dataset): +def create_data_loaders( + train_dataset: object, test_dataset: object +) -> tuple[object, object]: """Create train and test data loaders.""" train_loader = torch.utils.data.DataLoader( dataset=train_dataset, batch_size=BATCH_SIZE, shuffle=True @@ -45,7 +52,7 @@ train_acc_values = [] val_acc_values = [] -def define_model(hyperparameters): +def define_model(hyperparameters: dict[str, int]) -> object: """Define the multilayer perceptron training_parameters['model'].""" # Define the multilayer perceptron model model = nn.Sequential() @@ -59,7 +66,7 @@ def define_model(hyperparameters): return model -def get_optimizer(model): +def get_optimizer(model: object) -> object: """Return optimizer function.""" if OPTIMIZER_TYPE == "SGD": return optim.SGD(model.parameters(), lr=LEARNING_RATE) @@ -67,10 +74,11 @@ def get_optimizer(model): return optim.SGD(model.parameters(), lr=LEARNING_RATE, momentum=0.9) if OPTIMIZER_TYPE == "Adam": return optim.Adam(model.parameters(), lr=LEARNING_RATE) - raise ValueError("Unsupported optimizer type!") + msg = f"Unsupported optimizer type: {OPTIMIZER_TYPE}" + raise ValueError(msg) -def initial_configuration(): +def initial_configuration() -> tuple[dict[str, int], object, object, object]: """Perform all operations needed for training network.""" # Set random seed for reproducibility torch.manual_seed(42) @@ -81,15 +89,20 @@ def initial_configuration(): model = define_model(hyperparameters) # Loss function criterion = nn.CrossEntropyLoss() - # training_parameters['optimizer'] optimizer = get_optimizer(model) return hyperparameters, train_loader, test_loader, model, criterion, optimizer -def single_train_iteration(data, training_parameters, targets, batch_idx, epoch): +def single_train_iteration( + data: object, + training_parameters: dict[str, object], + targets: object, + batch_idx: int, + epoch: int, +) -> tuple[object, object]: """Train network for single batch.""" # Reshape the input data - data = data.view(data.size(0), -1) + _ = data.view(data.size(0), -1) # Forward pass outputs = training_parameters["model"](data) loss = training_parameters["criterion"](outputs, targets) @@ -99,24 +112,32 @@ def single_train_iteration(data, training_parameters, targets, batch_idx, epoch) training_parameters["optimizer"].step() # Print loss value for every learning step if (batch_idx + 1) % 100 == 0: - print( - f""" - Epoch [{epoch + 1}/{training_parameters["hyperparameters"]["num_epochs"]}], - Step [{batch_idx + 1}/{len(training_parameters["loaders"]["train_loader"])}], - Loss: {loss.item():.4f} - """ + logger.info( + "\n Epoch [%s/%s],\n Step [%s/%s],\n " + " Loss: %s\n ", + epoch + 1, + training_parameters["hyperparameters"]["num_epochs"], + batch_idx + 1, + len(training_parameters["loaders"]["train_loader"]), + loss.item(), ) # Append loss value for every learning step loss_values.append(loss.item()) return data, training_parameters["optimizer"] -def set_loaders(train_loader, test_loader): +def set_loaders(train_loader: object, test_loader: object) -> dict[str, object]: """Put train and test loaders into one object.""" return {"train_loader": train_loader, "test_loader": test_loader} -def set_training_parameters(hyperparameters, loaders, model, criterion, optimizer): +def set_training_parameters( + hyperparameters: dict[str, int], + loaders: dict[str, object], + model: object, + criterion: object, + optimizer: object, +) -> dict[str, object]: """Put all training parameters into one object.""" return { "hyperparameters": hyperparameters, @@ -130,7 +151,9 @@ def set_training_parameters(hyperparameters, loaders, model, criterion, optimize } -def training_loop(training_parameters, print_info=True): +def training_loop( + training_parameters: dict[str, object], *, print_info: bool = True +) -> tuple[int, object]: """Train network for all epochs.""" epochs_num = training_parameters["hyperparameters"]["num_epochs"] # Training loop @@ -138,7 +161,7 @@ def training_loop(training_parameters, print_info=True): for batch_idx, (data, targets) in enumerate( training_parameters["loaders"]["train_loader"] ): - data, training_parameters["optimizer"] = single_train_iteration( + _, training_parameters["optimizer"] = single_train_iteration( data, training_parameters, targets, batch_idx, epoch ) calculate_accuracy_epoch(training_parameters, epoch, print_info) @@ -146,28 +169,34 @@ def training_loop(training_parameters, print_info=True): return epoch, training_parameters["loaders"]["train_loader"] -def calculate_accuracy_epoch(training_parameters, epoch, print_info=True): +def calculate_accuracy_epoch( + training_parameters: dict[str, object], epoch: int, *, print_info: bool = True +) -> None: """Calculate accuracy on train set after each epoch.""" correct = 0 total = 0 for data, targets in training_parameters["loaders"]["train_loader"]: - data = data.view(data.size(0), -1) + _ = data.view(data.size(0), -1) outputs = training_parameters["model"](data) _, predicted = torch.max(outputs.data, 1) total += targets.size(0) correct += (predicted == targets).sum().item() train_accuracy = 100 * correct / total if print_info: - print(f"Accuracy on Train Set after Epoch {epoch + 1}: {train_accuracy:.2f}%") + logger.info( + "Accuracy on Train Set after Epoch %s: %s%%", epoch + 1, train_accuracy + ) train_acc_values.append(train_accuracy) -def calculate_validation_set_accuracy(training_parameters, epoch, print_info=True): +def calculate_validation_set_accuracy( + training_parameters: dict[str, object], epoch: int, *, print_info: bool = True +) -> None: """Calculate accuracy on validation set after each epoch.""" correct = 0 total = 0 for data, targets in training_parameters["loaders"]["test_loader"]: - data = data.view(data.size(0), -1) + _ = data.view(data.size(0), -1) outputs = training_parameters["model"](data) _, predicted = torch.max(outputs.data, 1) total += targets.size(0) @@ -175,45 +204,46 @@ def calculate_validation_set_accuracy(training_parameters, epoch, print_info=Tru validation_accuracy = 100 * correct / total if print_info: - print( - f"Accuracy on Validation Set after Epoch {epoch + 1}: {validation_accuracy:.2f}%" + logger.info( + "Accuracy on Validation Set after Epoch %s: %s%%", + epoch + 1, + validation_accuracy, ) - print("---") + logger.info("---") val_acc_values.append(validation_accuracy) -def main_part(show_plot=True): +def main_part(*, show_plot: bool = True) -> None: + """Train the model, evaluate it, and optionally plot the losses.""" ( - HYPERPARAMETERS, - TRAIN_LOADER, - TEST_LOADER, - MODEL, - CRITERION, - OPTIMIZER, + hyperparameters, + train_loader, + test_loader, + model, + criterion, + optimizer, ) = initial_configuration() start_time = time.time() - LOADERS = set_loaders(TRAIN_LOADER, TEST_LOADER) - TRAINING_PARAMETERS = set_training_parameters( - HYPERPARAMETERS, LOADERS, MODEL, CRITERION, OPTIMIZER - ) - training_loop(TRAINING_PARAMETERS, show_plot) - file = open("results.txt", "a") - file.write( - "-------------------------------------------------------------------------------------" - "\n" - ) - file.write( - f"loss-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl{NUM_HIDDEN_LAYERS}-w{WIDTH}-{OPTIMIZER_TYPE}" - "\n" - ) - file.write(f"Execution time: {(time.time() - start_time)}" + "\n") - file.write( - "-------------------------------------------------------------------------------------" - "\n" + loaders = set_loaders(train_loader, test_loader) + training_parameters = set_training_parameters( + hyperparameters, loaders, model, criterion, optimizer ) + training_loop(training_parameters, show_plot) + with Path("results.txt").open("a") as file: + rule = "-" * 85 + "\n" + file.write(rule) + file.write( + f"loss-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl{NUM_HIDDEN_LAYERS}" + f"-w{WIDTH}-{OPTIMIZER_TYPE}\n" + ) + file.write(f"Execution time: {time.time() - start_time}\n") + file.write(rule) # Plot the loss value for every learning step - learning_step_title = f"loss-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl{NUM_HIDDEN_LAYERS}-w{WIDTH}-{OPTIMIZER_TYPE}.png" + learning_step_title = ( + f"loss-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl{NUM_HIDDEN_LAYERS}" + f"-w{WIDTH}-{OPTIMIZER_TYPE}.png" + ) plt.plot(loss_values) plt.xlabel("Learning Step") plt.ylabel("Loss") @@ -224,7 +254,10 @@ def main_part(show_plot=True): plt.close() # Plot the accuracy on train set after each epoch - train_accuracy_title = f"trainAccuracy-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl{NUM_HIDDEN_LAYERS}-w{WIDTH}-{OPTIMIZER_TYPE}.png" + train_accuracy_title = ( + f"trainAccuracy-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl" + f"{NUM_HIDDEN_LAYERS}-w{WIDTH}-{OPTIMIZER_TYPE}.png" + ) plt.plot(train_acc_values) plt.xlabel("Epoch") plt.ylabel("Train Accuracy") @@ -235,7 +268,10 @@ def main_part(show_plot=True): plt.close() # Plot the accuracy on validation set after each epoch - validation_accuracy_title = f"validationAccuracy-lr{LEARNING_RATE}-bs{BATCH_SIZE}-hl{NUM_HIDDEN_LAYERS}-w{WIDTH}-{OPTIMIZER_TYPE}.png" + validation_accuracy_title = ( + f"validationAccuracy-lr{LEARNING_RATE}-bs{BATCH_SIZE}" + f"-hl{NUM_HIDDEN_LAYERS}-w{WIDTH}-{OPTIMIZER_TYPE}.png" + ) plt.plot(val_acc_values) plt.xlabel("Epoch") plt.ylabel("Validation Accuracy") @@ -247,12 +283,13 @@ def main_part(show_plot=True): if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) LEARNING_RATE = 0.001 BATCH_SIZE = 64 NUM_HIDDEN_LAYERS = 2 WIDTH = 128 OPTIMIZER_TYPE = "Adam" - main_part(True) + main_part(show_plot=True) """ learning_rate_values = [0.1, 0.01, 0.001] i = 0 diff --git a/Programming/EARIN/lab6/main.py b/Programming/EARIN/lab6/main.py old mode 100644 new mode 100755 index d3ee1094..88220e3f --- a/Programming/EARIN/lab6/main.py +++ b/Programming/EARIN/lab6/main.py @@ -1,63 +1,66 @@ +#!/usr/bin/env python3 """Code used to solve MountainCar-v0 gymnasium problem using Q-Learning algorithm.""" -from datetime import datetime +import logging +from datetime import UTC, datetime import gymnasium as gym import numpy as np +# Only record video once the agent has had time to learn something. +VIDEO_FROM_EPISODE = 600 +# NumPy's legacy np.random.* functions share one global state; Generator is +# the modern API and is what NPY002 asks for. +RNG = np.random.default_rng() +logger = logging.getLogger(__name__) + # Helper function to discretize the state -def discretize_state(state, env, first_time): - # print( - # f"state: {state}, state[0]: {state[0]}, env.observation_space.low: {env.observation_space.low}") - # print(f"state[0] - env {state[0] - env.observation_space.low}") - # print(f"state - env {state - env.observation_space.low}") +def discretize_state(state: object, env: object, *, first_time: bool = False) -> object: + # f"state: {state}, state[0]: {state[0]}, env.observation_space.low: + # {env.observation_space.low}") + """Bucket a continuous observation into the Q-table's grid.""" if first_time: substract_from_state = state[0] - env.observation_space.low else: substract_from_state = state - env.observation_space.low discretized_state = (substract_from_state) * np.array([10, 100]) - discretized_state = np.round(discretized_state, 0).astype(int) - return discretized_state + return np.round(discretized_state, 0).astype(int) -def initialize_environment(hyperparameters): +def initialize_environment() -> object: """Initialize environment and video recording.""" # Initialize environment env = gym.make("MountainCar-v0", render_mode="rgb_array") # Save video - now = datetime.now() + now = datetime.now(tz=UTC) time_string = now.strftime("%H:%M:%S") - env = gym.wrappers.RecordVideo( + return gym.wrappers.RecordVideo( env, video_folder="vid", disable_logger=True, name_prefix=time_string, - episode_trigger=lambda x: x > 600 and x % 2 == 0, + episode_trigger=lambda x: x > VIDEO_FROM_EPISODE and x % 2 == 0, ) - return env -def initialize_q_table(env): +def initialize_q_table(env: object) -> object: """Initialize "empty" Q-table.""" # Initialize Q-table - # n_actions = env.action_space.n # Number of possible actions, should be 3 # 0 accelerate left # 1 dont accelerate # 2 accelerate to the right - # q_table = np.zeros((n_actions,)) num_states = (env.observation_space.high - env.observation_space.low) * np.array( [10, 100] ) num_states = np.round(num_states, 0).astype(int) + 1 - q_table = np.zeros((num_states[0], num_states[1], env.action_space.n)) - return q_table + return np.zeros((num_states[0], num_states[1], env.action_space.n)) -def initialize_hyperparameters(): +def initialize_hyperparameters() -> dict[str, object]: """Initialize hyperparameters used by algorithm.""" - hyperparameters = { + return { "learning_rate": 0.1, "discount_factor": 0.99, "epsilon": 0.2, @@ -68,13 +71,17 @@ def initialize_hyperparameters(): "goal_x": 0.5, "truncation": 200, } - return hyperparameters -def choose_action(hyperparameters, env, q_table, discretized_state): +def choose_action( + hyperparameters: dict[str, object], + env: object, + q_table: object, + discretized_state: object, +) -> int: """Choose one of 3 actions possible for the algorithm.""" # hyperparameters["epsilon"]-greedy exploration-exploitation tradeoff - if np.random.uniform(0, 1) < hyperparameters["epsilon"]: + if RNG.uniform(0, 1) < hyperparameters["epsilon"]: action = env.action_space.sample() # Choose a random action else: # Choose the action with the highest Q-value @@ -82,7 +89,9 @@ def choose_action(hyperparameters, env, q_table, discretized_state): return action -def update_q_table(q_table, action, hyperparameters, reward): +def update_q_table( + q_table: object, action: int, hyperparameters: dict[str, object], reward: float +) -> object: """Update q_table with newest reward.""" # Q-table update q_value = q_table[action] @@ -95,14 +104,22 @@ def update_q_table(q_table, action, hyperparameters, reward): def movement( - hyperparameters, env, q_table, discretized_state, total_reward=0, episode_number=0 -): - """Choose action and observe consequences.""" + hyperparameters: dict[str, object], + env: object, + q_table: object, + discretized_state: object, + progress: tuple[float, int] = (0.0, 0), +) -> tuple[dict[str, object], object, object, float]: + """Choose action and observe consequences. + + ``progress`` is (total_reward, episode_number); they travel together + through the whole training loop, so they are passed together. + """ + total_reward, episode_number = progress action = choose_action(hyperparameters, env, q_table, discretized_state) # Take the action and observe the next state next_state, reward, terminated, truncated, _ = env.step(action) - discretized_next_state = discretize_state(next_state, env, False) - # print(discretized_next_state[0], discretized_next_state[1]) + discretized_next_state = discretize_state(next_state, env, first_time=False) q_table[discretized_state[0], discretized_state[1], action] += hyperparameters[ "learning_rate" ] * ( @@ -116,25 +133,30 @@ def movement( discretized_state = discretized_next_state done = terminated or truncated if terminated: - print("Destination reached on episode: ", episode_number) + logger.info("Destination reached on episode: %s", episode_number) return hyperparameters, env, q_table, done, discretized_state, total_reward -def episode_step(env, hyperparameters, q_table, episode_rewards, episode_number): +def episode_step( + env: object, + hyperparameters: dict[str, object], + q_table: object, + episode_rewards: list[float], + episode_number: int, +) -> tuple[object, dict[str, object], object, list[float]]: """Actions done with every episode.""" state = env.reset() # Reset the environment to an initial state - discretized_state = discretize_state(state, env, True) + discretized_state = discretize_state(state, env, first_time=True) done = False # Boolean to indicate episode completion total_reward = 0 # Accumulate rewards for the episode - for step in range(hyperparameters["max_steps"]): + for _step in range(hyperparameters["max_steps"]): hyperparameters, env, q_table, done, discretized_state, total_reward = movement( hyperparameters, env, q_table, discretized_state, - total_reward, - episode_number, + (total_reward, episode_number), ) if done: break @@ -143,7 +165,9 @@ def episode_step(env, hyperparameters, q_table, episode_rewards, episode_number) return env, hyperparameters, q_table, episode_rewards -def training_loop(hyperparameters, env, q_table): +def training_loop( + hyperparameters: dict[str, object], env: object, q_table: object +) -> tuple[object, object]: """Actual training for MountainCar.""" episode_rewards = [] # List to store episode rewards @@ -155,7 +179,7 @@ def training_loop(hyperparameters, env, q_table): return env, q_table -def inference(env, q_table): +def inference(env: object, q_table: object) -> None: """Inference using the updated Q-table.""" env.reset() done = False @@ -169,6 +193,7 @@ def inference(env, q_table): if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) HYPERPARAMETERS = initialize_hyperparameters() ENV = initialize_environment(HYPERPARAMETERS) Q_TABLE = initialize_q_table(ENV) diff --git a/Programming/EARIN/project/final/code/main.py b/Programming/EARIN/project/final/code/main.py old mode 100644 new mode 100755 index 1ad34c89..a515e4c2 --- a/Programming/EARIN/project/final/code/main.py +++ b/Programming/EARIN/project/final/code/main.py @@ -1,21 +1,29 @@ -"""Code for preprocessing data and creating model that predicts and +#!/usr/bin/env python3 +"""Code for preprocessing data and creating model that predicts and. + recomends anime based on another anime entered by user. """ import argparse import datetime +import logging import math -import os import shutil +from dataclasses import dataclass +from pathlib import Path import numpy as np import pandas as pd from scipy.sparse import csr_matrix from sklearn.neighbors import NearestNeighbors +logger = logging.getLogger(__name__) -def get_data_cpu(limit_data=-1, data_folder_path="database"): - """Reads anime from csv database.""" + +def get_data_cpu( + limit_data: int = -1, data_folder_path: str = "database" +) -> tuple[pd.DataFrame, pd.DataFrame]: + """Read anime from csv database.""" if limit_data > -1: # User can limit number of data taken into consideration, # model seems to work with limit_data value as low as 500,000 @@ -26,7 +34,14 @@ def get_data_cpu(limit_data=-1, data_folder_path="database"): return rating_data, anime_data -def get_data(limit_data=-1, data_folder_path="database", gpu=False): +def get_data( + limit_data: int = -1, data_folder_path: str = "database" +) -> tuple[pd.DataFrame, pd.DataFrame, int]: + """Load the ratings and anime tables. + + The `gpu` switch this used to take was never read -- the body always went + through get_data_cpu -- so it is gone rather than kept as a lie. + """ rating_data, anime_data = get_data_cpu(limit_data, data_folder_path) # used to fetch anime_id(MAL_ID) @@ -36,7 +51,9 @@ def get_data(limit_data=-1, data_folder_path="database", gpu=False): return rating_data, anime_contact_data, rows_number -def merge_rating_anime_data(rating_data, anime_contact_data, debug=False): +def merge_rating_anime_data( + rating_data: pd.DataFrame, anime_contact_data: pd.DataFrame, *, debug: bool = False +) -> pd.DataFrame: """Preprocesses the data used for rating.""" rating_data = rating_data.merge( anime_contact_data, left_on="anime_id", right_on="anime_id", how="left" @@ -46,15 +63,21 @@ def merge_rating_anime_data(rating_data, anime_contact_data, debug=False): ] rating_head = rating_data.head() if debug: - print(rating_head) + logger.info("%s", rating_head) rating_shape_complete = rating_data.shape if debug: - print(rating_shape_complete) + logger.info("%s", rating_shape_complete) return rating_data -def split_data_below_thresholds(rating_data, data_name, threshold=-1, debug=False): - """Removes data with data_name which is below given threshold.""" +def split_data_below_thresholds( + rating_data: pd.DataFrame, + data_name: str, + threshold: int = -1, + *, + debug: bool = False, +) -> pd.DataFrame: + """Remove data with data_name which is below given threshold.""" if threshold != -1: count = rating_data[data_name].value_counts() rating_data = rating_data[ @@ -62,12 +85,15 @@ def split_data_below_thresholds(rating_data, data_name, threshold=-1, debug=Fals ].copy() rating_shape_cut = rating_data.shape if debug: - print(rating_shape_cut) + logger.info("%s", rating_shape_cut) return rating_data -def combine_name_and_ratings(rating_data, debug=False): - """Create table which holds name of the anime and number of its reviews +def combine_name_and_ratings( + rating_data: pd.DataFrame, *, debug: bool = False +) -> pd.DataFrame: + """Create table which holds name of the anime and number of its reviews. + then we merge this with rating_data. """ combine_movie_rating = rating_data.dropna(axis=0, subset=["Name"]) @@ -78,24 +104,27 @@ def combine_name_and_ratings(rating_data, debug=False): ) rating_head = movie_rating_count.head() if debug: - print(rating_head) - rating_data = combine_movie_rating.merge( + logger.info("%s", rating_head) + return combine_movie_rating.merge( movie_rating_count, left_on="Name", right_on="Name", how="left" ) - return rating_data -def get_length_of_data(rating_data, data_name): +def get_length_of_data(rating_data: pd.DataFrame, data_name: str) -> int: """We get amount of data in the database with a given column data_name.""" # Encoding categorical data column_ids = rating_data[data_name + "_id"].unique().tolist() column_to_column = {x: i for i, x in enumerate(column_ids)} rating_data[data_name] = rating_data[data_name + "_id"].map(column_to_column) - users_number = len(column_to_column) - return users_number + return len(column_to_column) -def get_top_ranked(rating_data, data_name, join_table=None, top_data_taken=20): +def get_top_ranked( + rating_data: pd.DataFrame, + data_name: str, + join_table: pd.DataFrame | None = None, + top_data_taken: float = 20, +) -> pd.DataFrame: """Get anime with highest ranking.""" if join_table is None: join_table = rating_data @@ -103,13 +132,10 @@ def get_top_ranked(rating_data, data_name, join_table=None, top_data_taken=20): top_users = group_data_by_rating.dropna().sort_values(ascending=False)[ :top_data_taken ] - top_rated = join_table.join( - top_users, rsuffix="_r", how="inner", on=data_name + "_id" - ) - return top_rated + return join_table.join(top_users, rsuffix="_r", how="inner", on=data_name + "_id") -def get_data_info(rating_data, debug=False, gpu=False): +def get_data_info(rating_data: pd.DataFrame, *, debug: bool = False) -> None: """Get some informations about data.""" users_number = get_length_of_data(rating_data, "user") animes_number = get_length_of_data(rating_data, "anime") @@ -120,27 +146,40 @@ def get_data_info(rating_data, debug=False, gpu=False): top_rated.user_id, top_rated.anime_id, top_rated.rating, aggfunc=np.sum ) - pivot.fillna(0, inplace=True) + pivot = pivot.fillna(0) smallest_rating = min(rating_data["rating"]) highest_rating = max(rating_data["rating"]) if debug: - print(pivot) + logger.info("%s", pivot) if debug: - print(f"Num of users: {users_number}, Num of animes: {animes_number}") - print( - f"Min total rating: {smallest_rating}, Max total rating: {highest_rating}" + logger.info("Num of users: %s, Num of animes: %s", users_number, animes_number) + logger.info( + "Min total rating: %s, Max total rating: %s", + smallest_rating, + highest_rating, ) +@dataclass(frozen=True) +class Thresholds: + """Minimum vote counts a user and an anime need to stay in the data.""" + + user: int = 500 + anime: int = 200 + + def preprocessing( - rating_data, - anime_contact_data, - debug=False, - user_threshold=500, - anime_threshold=200, - auto=False, -): - """Preprocesses data for making model more accurate and/or faster.""" + rating_data: pd.DataFrame, + anime_contact_data: pd.DataFrame, + thresholds: Thresholds | None = None, + *, + debug: bool = False, + auto: bool = False, +) -> pd.DataFrame: + """Preprocess the data so the model is faster and more accurate.""" + if thresholds is None: + thresholds = Thresholds() + user_threshold, anime_threshold = thresholds.user, thresholds.anime rating_data = merge_rating_anime_data(rating_data, anime_contact_data) rating_data = split_data_below_thresholds(rating_data, "user_id", user_threshold) rating_data = split_data_below_thresholds(rating_data, "anime_id", anime_threshold) @@ -149,52 +188,61 @@ def preprocessing( rating_data = rating_data.drop(columns="rating_y") rating_data = rating_data.rename(columns={"rating_x": "rating"}) if debug and not auto: - print(rating_data) - get_data_info(rating_data, True) + logger.info("%s", rating_data) + get_data_info(rating_data, debug=True) pivot_table = rating_data.pivot_table( index="Name", columns="user_id", values="rating" ).fillna(0) if debug and not auto: - print(pivot_table) + logger.info("%s", pivot_table) return pivot_table def predict( - prediction_model, - pivot_table, - seed=42, - anime="RANDOM", - recommendation_number=6, - auto=False, - debug=False, -): - """This will choose a random anime name and our prediction_model will predict similar anime.""" - np.random.seed(seed) + prediction_model: object, + pivot_table: pd.DataFrame, + request: tuple[int, str, int] = (42, "RANDOM", 6), + *, + debug: bool = False, +) -> tuple[str, object]: + """Pick a random anime and recommend the ones most like it. + + ``request`` is (seed, anime, recommendation_number); the three always + travel together, and bundling them keeps this under the argument limit. + """ + seed, anime, recommendation_number = request + # default_rng(seed) replaces np.random.seed: the legacy global state is + # what NPY002 flags, and a local generator is reproducible per call. + rng = np.random.default_rng(seed) if anime == "RANDOM": - chosen_anime = np.random.choice(pivot_table.shape[0]) - query = pivot_table.iloc[chosen_anime, :].values.reshape(1, -1) + chosen_anime = rng.choice(pivot_table.shape[0]) + query = pivot_table.iloc[chosen_anime, :].to_numpy().reshape(1, -1) chosen_anime_name = pivot_table.index[chosen_anime] else: - query = pivot_table.loc[anime].values.reshape(1, -1) + query = pivot_table.loc[anime].to_numpy().reshape(1, -1) chosen_anime_name = anime distance, suggestions = prediction_model.kneighbors(query) if debug: - print("prediction model, distance: ", distance) + logger.info("prediction model, distance: %s", distance) for i in range(2): if i == 0: - print(f"Recommendations for {chosen_anime_name}:\n") + logger.info("Recommendations for %s:\n", chosen_anime_name) else: - print( - f"""{i}: {pivot_table.index[suggestions.flatten()[i]]}, - with distance of {distance.flatten()[i]}:""" + logger.info( + "%s: %s,\n with distance of %s:", + i, + pivot_table.index[suggestions.flatten()[i]], + distance.flatten()[i], ) average_distance = np.mean(distance.flatten()) - closest_anime_name = pivot_table.index[suggestions.flatten()[1]] + _closest_anime_name = pivot_table.index[suggestions.flatten()[1]] closest_anime_distance = distance.flatten()[1] average_minus_closest_distance = average_distance - closest_anime_distance - print( - f"Average distance: {average_distance}, average_minus_closest_distance: {average_minus_closest_distance}" + logger.info( + "Average distance: %s, average_minus_closest_distance: %s", + average_distance, + average_minus_closest_distance, ) return ( @@ -203,10 +251,10 @@ def predict( distance.flatten()[1 : recommendation_number + 1], f"{closest_anime_distance}_{average_distance}_{average_minus_closest_distance}", ) - # return f"{chosen_anime_name}_{closest_anime_name}_{closest_anime_distance}_{average_distance}_{average_minus_closest_distance}" -def calculate_neighbors(rows_number, neighbors=5): +def calculate_neighbors(rows_number: int, neighbors: int = 5) -> int: + """Resolve the k of k-NN from the named strategies, or pass it through.""" neighbor_value = { "default": 5, "sqrt": math.floor(math.sqrt(rows_number)), @@ -220,11 +268,14 @@ def calculate_neighbors(rows_number, neighbors=5): def create_model( - pivot_table, rows_number, metric="cosine", algorithm="brute", neighbors=5 -): - """Creates model based on neaarest neighbor for anime prediction.""" + pivot_table: pd.DataFrame, + metric: str = "cosine", + algorithm: str = "brute", + neighbors: int = 5, +) -> object: + """Create model based on neaarest neighbor for anime prediction.""" neighbors_number = calculate_neighbors(pivot_table.shape[0], neighbors) - pivot_table_matrix = csr_matrix(pivot_table.values) + pivot_table_matrix = csr_matrix(pivot_table.to_numpy()) if algorithm == "brute": model = NearestNeighbors( n_neighbors=neighbors_number, metric=metric, algorithm=algorithm @@ -233,15 +284,20 @@ def create_model( model = NearestNeighbors(n_neighbors=neighbors_number, algorithm=algorithm) try: model.fit(pivot_table_matrix) - except: - print(f"""Error in create_model, probably wrong metric for data - Metric: {metric}, algorithm: {algorithm}""") + except ValueError: + # sklearn raises ValueError when a metric does not accept this data. + logger.info( + "Error in create_model, probably wrong metric for data\n " + "Metric: %s, algorithm: %s", + metric, + algorithm, + ) return "Error!" return model -def handle_arguments(): - """Handles all arguments that can be used to change algorithm behaviour or program display.""" +def handle_arguments() -> tuple[object, ...]: + """Handle every argument that changes the algorithm's behaviour or display.""" parser = argparse.ArgumentParser(description="Example script with pyargs") parser.add_argument( "--data_limit", @@ -359,8 +415,9 @@ def handle_arguments(): ) -def auto_mode(data_limit=-1, seed=42, anime="RANDOM"): - print("Started auto mode") +def auto_mode(data_limit: int = -1, seed: int = 42, anime: str = "RANDOM") -> None: + """Run one unattended recommendation with fixed settings.""" + logger.info("Started auto mode") algorithm_spread = ["auto", "brute"] metric_spread = ["manhattan", "euclidean", "cosine"] neighbor_spread = [5, "sqrt", "half", "log", "n-1"] @@ -371,88 +428,106 @@ def auto_mode(data_limit=-1, seed=42, anime="RANDOM"): original_pivot_table = preprocessing( starting_rating_data, starting_anime_contact_data ) - if os.path.exists("test_results"): + if Path("test_results").exists(): shutil.rmtree("test_results") for algorithm in algorithm_spread: - possibleMetrics = [] + possible_metrics = [] if algorithm != "auto": - possibleMetrics = metric_spread - print("testing for algorithm: ", algorithm, possibleMetrics) - if possibleMetrics == []: - possibleMetrics = [""] - for metric in possibleMetrics: + possible_metrics = metric_spread + logger.info("testing for algorithm: %s %s", algorithm, possible_metrics) + if possible_metrics == []: + possible_metrics = [""] + for metric in possible_metrics: if metric != "precomputed": - print("testing for algorithm, metric: ", algorithm, metric) + logger.info("testing for algorithm, metric: %s %s", algorithm, metric) for neighbor_amount in neighbor_spread: - print( - "testing for algorithm, metric, neighbor_amount: ", + logger.info( + "testing for algorithm, metric, neighbor_amount: %s %s %s", algorithm, metric, neighbor_amount, ) preprocess_model_predict( - starting_rating_data, - starting_anime_contact_data, starting_rows_number, original_pivot_table, - seed=seed, - anime=anime, - neighbors=neighbor_amount, - algorithm=algorithm, - metric=metric, + ModelSettings( + metric=metric, + algorithm=algorithm, + neighbors=neighbor_amount, + seed=seed, + anime=anime, + ), ) -def write_test_results(title, result=""): - # Create directory if it doesn't already exist - - if not os.path.exists("test_results"): - os.makedirs("test_results") +def write_test_results(title: str, result: object = "") -> None: + """Append one measured result to the results file.""" + if not Path("test_results").exists(): + Path("test_results").mkdir(parents=True) # Generate timestamped filename - timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") # e.g., 20230611235959 + timestamp = datetime.datetime.now(tz=datetime.timezone.utc).strftime( + "%Y%m%d%H%M%S" + ) # e.g., 20230611235959 filename = f"{title}_{timestamp}.txt" # Create and write to the file - with open(os.path.join("test_results", filename), "a") as file: + with (Path("test_results") / filename).open("a") as file: file.write(result) -def calculate_precision(predictions, threshold=8): +def calculate_precision(predictions: object, threshold: float = 8) -> float: + """Precision at the given rating threshold.""" ratings = [anime[anime > 0].mean() for anime in predictions] precision = [1 if r >= threshold else 0 for r in ratings] return np.mean(precision) +@dataclass(frozen=True) +class ModelSettings: + """One point in the k-NN parameter sweep the report tabulates.""" + + metric: str = "cosine" + algorithm: str = "brute" + neighbors: int = 5 + seed: int = 42 + anime: str = "RANDOM" + recommendation_amount: int = 5 + user_threshold: int = 500 + anime_threshold: int = 200 + + def preprocess_model_predict( - rating_data, - anime_contact_data, - rows_number, - pivot_table, - data_limit=-1, - db="database", - debug=False, - user_threshold=500, - anime_threshold=200, - metric="cosine", - algorithm="brute", - neighbors=5, - seed=42, - anime="RANDOM", - recommendation_amount=5, -): - MODEL = create_model(pivot_table, rows_number, metric, algorithm, neighbors) + rows_number: int, + pivot_table: pd.DataFrame, + settings: ModelSettings | None = None, +) -> None: + """Build the model for one setting and record what it recommends. + + Five of its fifteen parameters -- rating_data, anime_contact_data, + data_limit, db and debug -- were never read: the preprocessing they + belonged to happens in the caller, which passes the finished pivot_table. + They are gone rather than annotated as unused. The rest travel together as + one ModelSettings, which is also what keeps this under the argument limit. + """ + if settings is None: + settings = ModelSettings() + metric = settings.metric + algorithm = settings.algorithm + neighbors = settings.neighbors + seed = settings.seed + anime = settings.anime + recommendation_amount = settings.recommendation_amount + user_threshold = settings.user_threshold + anime_threshold = settings.anime_threshold + model = create_model(pivot_table, metric, algorithm, neighbors) result = "" - if MODEL != "Error!": + if model != "Error!": chosen_anime, suggestions, distance, distance_data = predict( - MODEL, pivot_table, seed, anime, recommendation_amount + model, pivot_table, (seed, anime, recommendation_amount) ) chosen_anime_name = pivot_table.index[chosen_anime] - # average_distance = np.mean(distance) - # closest_anime_name = pivot_table.index[suggestions[1]] - # closest_anime_distance = distance[1] - # average_minus_closest_distance = closest_anime_distance - average_distance precision = calculate_precision([pivot_table.iloc[s] for s in suggestions]) result = f"{chosen_anime_name}:\n" @@ -463,14 +538,15 @@ def preprocess_model_predict( "Smallest distance, average distance, Average - Smallest distance: " + distance_data ) - # result = f"{chosen_anime_name}_{closest_anime_name}_{closest_anime_distance}_{average_distance}_{average_minus_closest_distance}" write_test_results( - f"dl={rows_number}&s={seed}&m={metric}&a={algorithm}&ut={user_threshold}&at={anime_threshold}&n={neighbors}", + f"dl={rows_number}&s={seed}&m={metric}&a={algorithm}" + f"&ut={user_threshold}&at={anime_threshold}&n={neighbors}", result, ) if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) ( SEED, DEBUG, @@ -486,32 +562,28 @@ if __name__ == "__main__": AUTO, ) = handle_arguments() if not AUTO: - print("Entered not auto mode") + logger.info("Entered not auto mode") starting_rating_data, starting_anime_contact_data, starting_rows_number = ( get_data(limit_data=DATA_LIMIT, data_folder_path=DB) ) pivot_table = preprocessing( starting_rating_data, starting_anime_contact_data, - USER_THRESHOLD, - ANIME_THRESHOLD, + Thresholds(USER_THRESHOLD, ANIME_THRESHOLD), ) preprocess_model_predict( - starting_rating_data, - starting_anime_contact_data, starting_rows_number, pivot_table, - data_limit=DATA_LIMIT, - db=DB, - debug=DEBUG, - user_threshold=USER_THRESHOLD, - anime_threshold=ANIME_THRESHOLD, - metric=METRIC, - algorithm=ALGORITHM, - neighbors=NEIGHBORS, - seed=SEED, - anime=ANIME, - recommendation_amount=RECOMMENDATION_AMOUNT, + ModelSettings( + metric=METRIC, + algorithm=ALGORITHM, + neighbors=NEIGHBORS, + seed=SEED, + anime=ANIME, + recommendation_amount=RECOMMENDATION_AMOUNT, + user_threshold=USER_THRESHOLD, + anime_threshold=ANIME_THRESHOLD, + ), ) if AUTO: auto_mode(DATA_LIMIT, SEED, ANIME) diff --git a/Programming/EARIN/project/midterm/code/main.py b/Programming/EARIN/project/midterm/code/main.py old mode 100644 new mode 100755 index 2b509332..09c67513 --- a/Programming/EARIN/project/midterm/code/main.py +++ b/Programming/EARIN/project/midterm/code/main.py @@ -1,17 +1,24 @@ -"""Code for preprocessing data and creating model that predicts and +#!/usr/bin/env python3 +"""Code for preprocessing data and creating model that predicts and. + recomends anime based on another anime entered by user. """ import argparse +import logging import numpy as np import pandas as pd from scipy.sparse import csr_matrix from sklearn.neighbors import NearestNeighbors +logger = logging.getLogger(__name__) -def get_data(limit_data=-1, data_folder_path="database"): - """Reads anime from csv database.""" + +def get_data( + limit_data: int = -1, data_folder_path: str = "database" +) -> tuple[pd.DataFrame, pd.DataFrame]: + """Read anime from csv database.""" if limit_data > -1: # User can limit number of data taken into consideration, # model seems to work with limit_data value as low as 500,000 @@ -26,7 +33,9 @@ def get_data(limit_data=-1, data_folder_path="database"): return rating_data, anime_contact_data -def merge_rating_anime_data(rating_data, anime_contact_data, debug=False): +def merge_rating_anime_data( + rating_data: pd.DataFrame, anime_contact_data: pd.DataFrame, *, debug: bool = False +) -> pd.DataFrame: """Preprocesses the data used for rating.""" rating_data = rating_data.merge( anime_contact_data, left_on="anime_id", right_on="anime_id", how="left" @@ -36,15 +45,21 @@ def merge_rating_anime_data(rating_data, anime_contact_data, debug=False): ] rating_head = rating_data.head() if debug: - print(rating_head) + logger.info("%s", rating_head) rating_shape_complete = rating_data.shape if debug: - print(rating_shape_complete) + logger.info("%s", rating_shape_complete) return rating_data -def split_data_below_thresholds(rating_data, data_name, threshold=-1, debug=False): - """Removes data with data_name which is below given threshold.""" +def split_data_below_thresholds( + rating_data: pd.DataFrame, + data_name: str, + threshold: int = -1, + *, + debug: bool = False, +) -> pd.DataFrame: + """Remove data with data_name which is below given threshold.""" if threshold != -1: count = rating_data[data_name].value_counts() rating_data = rating_data[ @@ -52,12 +67,15 @@ def split_data_below_thresholds(rating_data, data_name, threshold=-1, debug=Fals ].copy() rating_shape_cut = rating_data.shape if debug: - print(rating_shape_cut) + logger.info("%s", rating_shape_cut) return rating_data -def combine_name_and_ratings(rating_data, debug=False): - """Create table which holds name of the anime and number of its reviews +def combine_name_and_ratings( + rating_data: pd.DataFrame, *, debug: bool = False +) -> pd.DataFrame: + """Create table which holds name of the anime and number of its reviews. + then we merge this with rating_data. """ combine_movie_rating = rating_data.dropna(axis=0, subset=["Name"]) @@ -68,24 +86,27 @@ def combine_name_and_ratings(rating_data, debug=False): ) rating_head = movie_rating_count.head() if debug: - print(rating_head) - rating_data = combine_movie_rating.merge( + logger.info("%s", rating_head) + return combine_movie_rating.merge( movie_rating_count, left_on="Name", right_on="Name", how="left" ) - return rating_data -def get_length_of_data(rating_data, data_name): +def get_length_of_data(rating_data: pd.DataFrame, data_name: str) -> int: """We get amount of data in the database with a given column data_name.""" # Encoding categorical data column_ids = rating_data[data_name + "_id"].unique().tolist() column_to_column = {x: i for i, x in enumerate(column_ids)} rating_data[data_name] = rating_data[data_name + "_id"].map(column_to_column) - users_number = len(column_to_column) - return users_number + return len(column_to_column) -def get_top_ranked(rating_data, data_name, join_table=None, top_data_taken=20): +def get_top_ranked( + rating_data: pd.DataFrame, + data_name: str, + join_table: pd.DataFrame | None = None, + top_data_taken: float = 20, +) -> pd.DataFrame: """Get anime with highest ranking.""" if join_table is None: join_table = rating_data @@ -93,13 +114,10 @@ def get_top_ranked(rating_data, data_name, join_table=None, top_data_taken=20): top_users = group_data_by_rating.dropna().sort_values(ascending=False)[ :top_data_taken ] - top_rated = join_table.join( - top_users, rsuffix="_r", how="inner", on=data_name + "_id" - ) - return top_rated + return join_table.join(top_users, rsuffix="_r", how="inner", on=data_name + "_id") -def get_data_info(rating_data, debug=False): +def get_data_info(rating_data: pd.DataFrame, *, debug: bool = False) -> None: """Get some informations about data.""" users_number = get_length_of_data(rating_data, "user") animes_number = get_length_of_data(rating_data, "anime") @@ -111,25 +129,28 @@ def get_data_info(rating_data, debug=False): top_rated.user_id, top_rated.anime_id, top_rated.rating, aggfunc=np.sum ) - pivot.fillna(0, inplace=True) + pivot = pivot.fillna(0) smallest_rating = min(rating_data["rating"]) highest_rating = max(rating_data["rating"]) if debug: - print(pivot) + logger.info("%s", pivot) if debug: - print(f"Num of users: {users_number}, Num of animes: {animes_number}") - print( - f"Min total rating: {smallest_rating}, Max total rating: {highest_rating}" + logger.info("Num of users: %s, Num of animes: %s", users_number, animes_number) + logger.info( + "Min total rating: %s, Max total rating: %s", + smallest_rating, + highest_rating, ) def preprocessing( - rating_data, - anime_contact_data, - debug=False, - user_threshold=500, - anime_threshold=200, -): + rating_data: pd.DataFrame, + anime_contact_data: pd.DataFrame, + *, + debug: bool = False, + user_threshold: int = 500, + anime_threshold: int = 200, +) -> pd.DataFrame: """Preprocesses data for making model more accurate and/or faster.""" rating_data = merge_rating_anime_data(rating_data, anime_contact_data) rating_data = split_data_below_thresholds(rating_data, "user_id", user_threshold) @@ -139,29 +160,35 @@ def preprocessing( rating_data = rating_data.drop(columns="rating_x") rating_data = rating_data.rename(columns={"rating_y": "rating"}) if debug: - print(rating_data) + logger.info("%s", rating_data) get_data_info(rating_data) pivot_table = rating_data.pivot_table( index="Name", columns="user_id", values="rating" ).fillna(0) if debug: - print(pivot_table) + logger.info("%s", pivot_table) return pivot_table def predict( - prediction_model, pivot_table, seed=42, anime="RANDOM", recommendation_number=6 -): - """This will choose a random anime name and our prediction_model will predict similar anime.""" - np.random.seed(seed) - print(pivot_table) + prediction_model: object, + pivot_table: pd.DataFrame, + seed: int = 42, + anime: str = "RANDOM", + recommendation_number: int = 6, +) -> None: + """Pick a random anime and recommend the ones most like it.""" + # default_rng(seed) replaces np.random.seed: the legacy global state is + # what NPY002 flags, and a local generator is reproducible per call. + rng = np.random.default_rng(seed) + logger.info("%s", pivot_table) if anime == "RANDOM": - chosen_anime = np.random.choice(pivot_table.shape[0]) - query = pivot_table.iloc[chosen_anime, :].values.reshape(1, -1) + chosen_anime = rng.choice(pivot_table.shape[0]) + query = pivot_table.iloc[chosen_anime, :].to_numpy().reshape(1, -1) chosen_anime_name = pivot_table.index[chosen_anime] else: - query = pivot_table.loc[anime].values.reshape(1, -1) + query = pivot_table.loc[anime].to_numpy().reshape(1, -1) chosen_anime_name = anime distance, suggestions = prediction_model.kneighbors( @@ -169,22 +196,31 @@ def predict( ) for i in range(len(distance.flatten())): if i == 0: - print(f"Recommendations for {chosen_anime_name}:\n") + logger.info("Recommendations for %s:\n", chosen_anime_name) else: - print( - f"{i}: {pivot_table.index[suggestions.flatten()[i]]}, with distance of {distance.flatten()[i]}:" + logger.info( + "%s: %s, with distance of %s:", + i, + pivot_table.index[suggestions.flatten()[i]], + distance.flatten()[i], ) -def create_model(pivot_table, metric="cosine", algorithm="brute", neighbors=5): - """Creates model based on neaarest neighbor for anime prediction.""" - pivot_table_matrix = csr_matrix(pivot_table.values) +def create_model( + pivot_table: pd.DataFrame, + metric: str = "cosine", + algorithm: str = "brute", + neighbors: int = 5, +) -> object: + """Create model based on neaarest neighbor for anime prediction.""" + pivot_table_matrix = csr_matrix(pivot_table.to_numpy()) model = NearestNeighbors(n_neighbors=neighbors, metric=metric, algorithm=algorithm) model.fit(pivot_table_matrix) return model -def handle_arguments(): +def handle_arguments() -> tuple[object, ...]: + """Parse the command line into the recommender's settings.""" parser = argparse.ArgumentParser(description="Example script with pyargs") parser.add_argument( "--data_limit", @@ -248,7 +284,10 @@ def handle_arguments(): parser.add_argument( "--user_threshold", "-ut", - help="Specify minimal number of votes required for user to be included in the data, set to -1 for no threshold", + help=( + "Specify minimal number of votes required for user to be included " + "in the data, set to -1 for no threshold" + ), required=False, type=int, default=500, @@ -256,7 +295,10 @@ def handle_arguments(): parser.add_argument( "--anime_threshold", "-at", - help="Specify minimal number of votes required for anime to be included in the data, set to -1 for no threshold", + help=( + "Specify minimal number of votes required for anime to be included " + "in the data, set to -1 for no threshold" + ), required=False, type=int, default=200, @@ -290,6 +332,7 @@ def handle_arguments(): if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) ( seed, debug, diff --git a/Programming/NLP/compare_chunks.py b/Programming/NLP/compare_chunks.py index b03a9947..32f9b698 100755 --- a/Programming/NLP/compare_chunks.py +++ b/Programming/NLP/compare_chunks.py @@ -9,6 +9,7 @@ import logging from pathlib import Path logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) def reformat_sentence(sentence: str) -> str: diff --git a/Programming/NLP/create_alignments_with_examples.py b/Programming/NLP/create_alignments_with_examples.py index 4795f450..73e8e1c3 100755 --- a/Programming/NLP/create_alignments_with_examples.py +++ b/Programming/NLP/create_alignments_with_examples.py @@ -9,6 +9,7 @@ import pandas as pd import processing logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) # Specify the output file path file_path = "alignments_with_training_headlines2.wa" diff --git a/Programming/NLP/format_alignments.py b/Programming/NLP/format_alignments.py index a1c72871..2b515d25 100755 --- a/Programming/NLP/format_alignments.py +++ b/Programming/NLP/format_alignments.py @@ -11,6 +11,7 @@ import pandas as pd import processing logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) # input file name file_path = "alignments_unformatted_headlines.txt" diff --git a/Programming/NLP/format_alignments_with_training.py b/Programming/NLP/format_alignments_with_training.py index 8bb32583..9f310d3a 100755 --- a/Programming/NLP/format_alignments_with_training.py +++ b/Programming/NLP/format_alignments_with_training.py @@ -10,6 +10,7 @@ import pandas as pd import processing logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) # paths to students andsewrs database student_answers1_path = ( diff --git a/Programming/NLP/gpt_chunks.py b/Programming/NLP/gpt_chunks.py index 1c949a4e..9252eaa1 100755 --- a/Programming/NLP/gpt_chunks.py +++ b/Programming/NLP/gpt_chunks.py @@ -11,6 +11,7 @@ from openai import OpenAI from openai.types.chat import ChatCompletion logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) def sentence_loop(sentences: list[str]) -> list[str]: diff --git a/Programming/NLP/processing.py b/Programming/NLP/processing.py index 5e922dfd..ae4efa32 100755 --- a/Programming/NLP/processing.py +++ b/Programming/NLP/processing.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 """Load the SemEval corpus: sentences, chunkings and gold alignments.""" +import sys import logging import re from pathlib import Path @@ -10,6 +11,7 @@ import pandas as pd from lxml import etree logger = logging.getLogger(__name__) +logging.basicConfig(format="%(message)s", level=logging.INFO) def load_sentences(senteance1_path: str, sentance2_path: str) -> pd.DataFrame: @@ -114,7 +116,7 @@ def load_alignment(alignment_path: str) -> pd.DataFrame: def prettyprint(element: etree._Element, **kwargs: object) -> None: """Log an XML element with lxml's pretty printer.""" xml = etree.tostring(element, pretty_print=True, **kwargs) - logger.info("%s", xml.decode()) + sys.stdout.write(xml.decode()) def log_first_alignment(alignments_data: list[dict[str, str]]) -> None: