kuhyx/WUT_Computer_Science
1
0
Fork 0
mirror of https://github.com/kuhyx/WUT_Computer_Science.git synced 2026-11-10 17:47:14 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

EARIN: zero findings, verified against byte-identical program output

Browse Source 
749 findings across six labs and two projects. The mechanical half was
scripted; everything below is what the scripts got wrong or could not do, and
how it was caught.

Three regressions I introduced and then found by RUNNING the code, not by
reading it:

- Converting print() to logger.info() made the programs SILENT. logging's
  last-resort handler only emits WARNING and above, so with no basicConfig
  call `lab2/main.py -t 1 1` printed nothing at all. Fifteen files across the
  repo now configure logging where they log.
- The same conversion dropped `end=""`. lab2 draws its board one square at a
  time and lab1 draws the maze one cell at a time; both came out one glyph
  per line. Those thirteen call sites are sys.stdout.write again, and the
  rendered board now diffs byte-identically against the pre-change output.
- The B007 fix renamed `col` to `_col` in the FIRST loop it matched, which
  was print_letters -- where the variable IS used. That would have been a
  NameError on the first call.

And one the scripts caused silently: annotating 110 signatures dropped the
DEFAULT VALUES from 22 of them, so `preprocessing(a, b, USER_THRESHOLD,
ANIME_THRESHOLD)` would have raised TypeError. Every signature is now diffed
against the pre-change version, name by name and default by default: zero
mismatches.

Real defects the lint surfaced:

- `alpha_beta` ended with an unreachable `return None` that violated its own
  return type. Its maximising and minimising halves were mirror images, so
  they are one `_search_branch` now (C901 11 -> ok).
- `get_data`'s `gpu` switch, `predict`'s `auto`, `create_model`'s
  `rows_number` and five of `preprocess_model_predict`'s fifteen parameters
  were never read. Gone, with their call sites.
- np.random.* -> Generator throughout (NPY002), `.values` -> `.to_numpy()`,
  `datetime.now()` with a tzinfo, `inplace=True` removed, `except:` given a
  type, `random.*` -> `SystemRandom` (S311).
- print_board's three nested helpers became methods (C901 17 -> ok);
  user_input's nine-branch if-ladder became a table (C901 12 -> ok).

Verification, all by running: lab1 `-t` and lab2 `-t 1 1` produce output
byte-identical to the pre-change code; lab2's rendered 8x8 board diffs clean;
lab3's argument parsing returns the same dict for the same argv; lab4 fails
identically before and after (no sklearn here). Also included: the logging
configuration for NLP, MOM and ECRYPT_PROJECT, which had the same silence bug.

Repo-wide: 2,775 findings -> 1,883.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0172dj34FDpao6wsix8ZuCzE
This commit is contained in:
Krzysztof Rudnicki 2026-08-28 20:39:53 +02:00
parent b017c7500e
commit cc06bc0cd8
17 changed files with 1089 additions and 732 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
NotProgramming/MOM/report_three/model.py
View File

@ -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)

241
Programming/EARIN/lab1/main.py Normal file → Executable file
View File

@ -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()

443
Programming/EARIN/lab2/main.py Normal file → Executable file
View File

@ -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()

41
Programming/EARIN/lab3/example.py Normal file → Executable file
View File

@ -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()

170
Programming/EARIN/lab3/main.py Normal file → Executable file
View File

@ -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)

54
Programming/EARIN/lab4/main.py Normal file → Executable file
View File

@ -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()

102
Programming/EARIN/lab5/code/example.py Normal file → Executable file
View File

@ -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])

155
Programming/EARIN/lab5/code/main.py Normal file → Executable file
View File

@ -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

99
Programming/EARIN/lab6/main.py Normal file → Executable file
View File

@ -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)

354
Programming/EARIN/project/final/code/main.py Normal file → Executable file
View File

@ -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)

149
Programming/EARIN/project/midterm/code/main.py Normal file → Executable file
View File

@ -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,

1
Programming/NLP/compare_chunks.py
View File

@ -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:

1
Programming/NLP/create_alignments_with_examples.py
View File

@ -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"

1
Programming/NLP/format_alignments.py
View File

@ -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"

1
Programming/NLP/format_alignments_with_training.py
View File

@ -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 = (

1
Programming/NLP/gpt_chunks.py
View File

@ -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]:

4
Programming/NLP/processing.py
View File

@ -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: