# Path: il2fb/ds/airbridge/compat.py
# IS_WINDOWS = (sys.platform == "win32")
# 
# Path: il2fb/ds/airbridge/typing.py

# 
# Path: il2fb/ds/airbridge/dedicated_server/path.py
# def normalize_exe_path(initial: str) -> Path:
#     return Path(initial).resolve()
# 
# def normalize_config_path(root_dir: Path, initial: str=None) -> Path:
#     return normalize_file_path(
#         root_dir=root_dir,
#         initial=initial,
#         default_name=DEFAULT_CONFIG_NAME,
#     )
# 
# def normalize_start_script_path(root_dir: Path, initial: str=None) -> Path:
#     return normalize_file_path(
#         root_dir=root_dir,
#         initial=initial,
#         default_name=DEFAULT_START_SCRIPT_NAME,
#     )
# 
# Path: il2fb/ds/airbridge/dedicated_server/validators.py
# def validate_dedicated_server_file_access(file_path: StringOrPath) -> None:
#     if not os.access(file_path, os.F_OK):
#         raise FileNotFoundError(
#             f"file does not exist (path='{file_path}')"
#         )
# 
#     if not os.access(file_path, os.R_OK):
#         raise PermissionError(
#             f"file does not have read access (path='{file_path}')"
#         )
# 
# def validate_dedicated_server_config(config: ServerConfig) -> None:
#     if not config.console.connection.port:
#         raise ValueError(
#             "server's console is disabled, please enable it "
#             "(see: https://github.com/IL2HorusTeam/il2fb-ds-config#console-section)"
#         )
# 
#     if not config.device_link.connection.port:
#         raise ValueError(
#             "server's device link is disabled, please enable it "
#             "(see: https://github.com/IL2HorusTeam/il2fb-ds-config#devicelink-section)"
#         )


import asyncio
import configparser
import functools
import io
import logging
import psutil
from pathlib import Path
from typing import List, Awaitable, Optional
from il2fb.config.ds import ServerConfig
from il2fb.ds.airbridge.compat import IS_WINDOWS
from il2fb.ds.airbridge.typing import IntOrNone, StringHandler, StringList
from il2fb.ds.airbridge.dedicated_server.path import (
    normalize_exe_path, normalize_config_path, normalize_start_script_path,
)
from il2fb.ds.airbridge.dedicated_server.validators import (
    validate_dedicated_server_file_access, validate_dedicated_server_config,
)
            self.config.console.connection.port,
            self.config.device_link.connection.port,
        }

        while timeout > 0:
            start_time = self._loop.time()

            actual_ports = {
                c.laddr.port
                for c in self._process_utils.connections('inet')
            }

            if actual_ports == expected_ports:
                return

            delay = min(timeout, polling_period)
            await asyncio.sleep(delay, loop=self._loop)

            time_delta = self._loop.time() - start_time
            timeout = max(0, timeout - time_delta)

        raise RuntimeError("expected ports of dedicated server are closed")

    async def ask_exit(self) -> Awaitable[None]:
        await self.input("exit\n")

    def terminate(self) -> None:
        if self._process and self._process.returncode is None:
            self._process.terminate()


# Path: il2fb/ds/airbridge/compat.py
# IS_WINDOWS = (sys.platform == "win32")
# 
# Path: il2fb/ds/airbridge/typing.py

# 
# Path: il2fb/ds/airbridge/dedicated_server/path.py
# def normalize_exe_path(initial: str) -> Path:
#     return Path(initial).resolve()
# 
# def normalize_config_path(root_dir: Path, initial: str=None) -> Path:
#     return normalize_file_path(
#         root_dir=root_dir,
#         initial=initial,
#         default_name=DEFAULT_CONFIG_NAME,
#     )
# 
# def normalize_start_script_path(root_dir: Path, initial: str=None) -> Path:
#     return normalize_file_path(
#         root_dir=root_dir,
#         initial=initial,
#         default_name=DEFAULT_START_SCRIPT_NAME,
#     )
# 
# Path: il2fb/ds/airbridge/dedicated_server/validators.py
# def validate_dedicated_server_file_access(file_path: StringOrPath) -> None:
#     if not os.access(file_path, os.F_OK):
#         raise FileNotFoundError(
#             f"file does not exist (path='{file_path}')"
#         )
# 
#     if not os.access(file_path, os.R_OK):
#         raise PermissionError(
#             f"file does not have read access (path='{file_path}')"
#         )
# 
# def validate_dedicated_server_config(config: ServerConfig) -> None:
#     if not config.console.connection.port:
#         raise ValueError(
#             "server's console is disabled, please enable it "
#             "(see: https://github.com/IL2HorusTeam/il2fb-ds-config#console-section)"
#         )
# 
#     if not config.device_link.connection.port:
#         raise ValueError(
#             "server's device link is disabled, please enable it "
#             "(see: https://github.com/IL2HorusTeam/il2fb-ds-config#devicelink-section)"
#         )


# Path: il2fb/ds/airbridge/dedicated_server/instance.py
import asyncio
import configparser
import functools
import io
import logging
import psutil
from pathlib import Path
from typing import List, Awaitable, Optional
from il2fb.config.ds import ServerConfig
from il2fb.ds.airbridge.compat import IS_WINDOWS
from il2fb.ds.airbridge.typing import IntOrNone, StringHandler, StringList
from il2fb.ds.airbridge.dedicated_server.path import (
    normalize_exe_path, normalize_config_path, normalize_start_script_path,
)
from il2fb.ds.airbridge.dedicated_server.validators import (
    validate_dedicated_server_file_access, validate_dedicated_server_config,
)

            self.config.console.connection.port,
            self.config.device_link.connection.port,
        }

        while timeout > 0:
            start_time = self._loop.time()

            actual_ports = {
                c.laddr.port
                for c in self._process_utils.connections('inet')
            }

            if actual_ports == expected_ports:
                return

            delay = min(timeout, polling_period)
            await asyncio.sleep(delay, loop=self._loop)

            time_delta = self._loop.time() - start_time
            timeout = max(0, timeout - time_delta)

        raise RuntimeError("expected ports of dedicated server are closed")

    async def ask_exit(self) -> Awaitable[None]:
        await self.input("exit\n")

    def terminate(self) -> None:
        if self._process and self._process.returncode is None:
            self._process.terminate()


    async def wait_finished(self) -> Awaitable[IntOrNone]: