plc-sim/server/program_public_tools.py

import os
import socket
from typing import Any, Optional

import time


class Timer:
    def __init__(self):
        self.start_time = None

    def start(self):
        if self.start_time is not None:
            print("Timer is already running. Use .elapsed_time() to check the time elapsed.")
        else:
            self.start_time = time.time()

    def elapsed_time(self):
        if self.start_time is None:
            print("Timer has not started yet. Use .start() to start the timer.")
        else:
            return time.time() - self.start_time

    def reset(self):
        self.start_time = None


class ProgramPublicTools(object):
    def __init__(self, enable_terminal_debug_output: bool = True):
        self.enable_terminal_debug_output = enable_terminal_debug_output

    def debug_output(self, *message: Any) -> None:
        self.__terminal_debug_output(*message)

    def __terminal_debug_output(self, *message: Any) -> None:
        if self.enable_terminal_debug_output is True:
            print(*message)

    @staticmethod
    def path_directory_exist_check(full_path: str) -> None:
        folder_path = os.path.dirname(full_path)
        os.makedirs(folder_path, exist_ok=True)

    @staticmethod
    def get_absolute_path(relative_path: str) -> str:
        script_dir = os.path.dirname(os.path.realpath(__file__))
        return os.path.abspath(os.path.join(script_dir, relative_path))

    @staticmethod
    def create_timer() -> Timer:
        return Timer()

    @staticmethod
    def is_convertible_to_int(num) -> tuple[bool, Optional[int]]:
        try:
            s = num
            s = int(s)
            return True, s
        except ValueError:
            return False, None

    @staticmethod
    def is_convertible_to_float(num) -> tuple[bool, Optional[float]]:
        try:
            s = num
            s = float(s)
            return True, s
        except ValueError:
            return False, None

    @staticmethod
    def find_available_server_port(number: int = 1, not_list: list[int] = None, address: str = "localhost",
                                   start_port: int = 1024, end_port: int = 65535) -> list[int]:
        get_port = []
        for port in range(start_port, end_port + 1):
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0)
                sock.bind((address, port))
                if not_list is not None:
                    if port in not_list:
                        pass
                    else:
                        get_port.append(port)
                else:
                    get_port.append(port)
                if len(get_port) == number:
                    break
            except OSError:
                continue
            finally:
                sock.close()
        if len(get_port) < number:
            raise Exception("Error: Do not find enough available port.")
        else:
            return get_port