#!/usr/bin/env python3

# Copyright (C) 2020  Braiins Systems s.r.o.
#
# This file is part of Braiins Open-Source Initiative (BOSI).
#
# BOSI is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
# Please, keep in mind that we may also license BOSI or any part thereof
# under a proprietary license. For more information on the terms and conditions
# of such proprietary license or if you have any other questions, please
# contact us at opensource@braiins.com.

import argparse
import base64
import csv
import html.parser
import json
import logging
import os
import os.path
import re
import requests
import sys
import urllib.parse

from bos_toolbox.platform.descriptor import TargetTriple

LOG = logging.getLogger(__name__)

# these are in roughly same order as in web UI
fields = [
    'host',
    # format
    'model',
    # BOS management
    'bos_mgmt_id',
    # pool group
    'pool0_user',
    'pool0_password',
    'pool0_url',
    'pool1_user',
    'pool1_password',
    'pool1_url',
    'pool2_user',
    'pool2_password',
    'pool2_url',
    # hashchain global
    'asic_boost',
    'global_frequency',
    'global_voltage',
    # hashchain specifics
    'hashchainA_enabled',
    'hashchainA_frequency',
    'hashchainA_voltage',
    'hashchainB_enabled',
    'hashchainB_frequency',
    'hashchainB_voltage',
    'hashchainC_enabled',
    'hashchainC_frequency',
    'hashchainC_voltage',
    # thermal control
    'thermal_mode',
    'target_temp',
    'hot_temp',
    'dangerous_temp',
    # fanc control
    'fan_speed',
    'fan_min',
    # autotuning
    'autotuning',
    'power_limit',
    # power scaling
    'power_scaling',
    'power_step',
    'min_power_limit',
    'shutdown',
    'shutdown_duration',
    # non-bosminer options
    'auto_upgrade',
    # password is never loaded, but can be changed
    'password',
]


def main(parser, args):

    if args.password == 'prompt':
        from getpass import getpass

        args.password = getpass()

    total = len(open(args.table, 'r').readlines())

    if args.action == 'load':
        tmp_table = args.table + '.tmp'
        with open(args.table, newline='') as reader_fd, open(
            tmp_table, 'w', newline=''
        ) as writer_fd:
            reader = csv.DictReader(reader_fd, fields, restval='')
            writer = csv.DictWriter(
                writer_fd, fields, restval='', dialect=reader.dialect
            )
            writer.writerow({k: k for k in fields})

            for line_no, row in enumerate(reader, 1):
                host = row['host'].strip()
                out = {'host': host}
                try:
                    if (
                        line_no == 1 and host == fields[0] or host == ''
                    ):  # skip first line with headers
                        continue
                    LOG.info('Pulling from %s (%d/%d)' % (host, line_no, total))
                    api = BosApi(host, args.user, args.password)
                    out['bos_mgmt_id'] = api.get_bos_mgmt_id()
                    out['auto_upgrade'] = toggle2str(
                        api.uci_get_bool_legacy('bos', 'auto_upgrade', 'enable')
                    )
                    cfg, triple = api.get_config()

                    # output row data
                    csvizer = Csvizer(cfg, triple)
                    csvizer.pull(out)
                    writer.writerow(out)
                except Exception as ex:
                    log_error(row)
                    if args.ignore:
                        LOG.warning(str(ex))
                        writer.writerow(out)
                    else:
                        raise

        if not args.check:
            # only replace original file once we are all done so as not to clobber it if error occurs
            os.replace(args.table, args.table + '.bak')
            os.replace(tmp_table, args.table)
        else:
            os.unlink(tmp_table)

    elif args.action in ('save', 'save_apply'):
        with open(args.table, newline='') as reader_fd:
            reader = csv.DictReader(reader_fd, fields, restval='')

            for line_no, row in enumerate(reader, 1):
                try:
                    host = row['host'].strip()
                    password = row['password'].strip()
                    if line_no == 1 and host == fields[0]:
                        # if column headers are present, make sure they fit
                        for name in fields:
                            if name != row[name]:
                                sys.exit(
                                    'Mismatched header: %s (expected %s)'
                                    % (row[name], name)
                                )
                        continue  # skip header row
                    if host == '':
                        continue  # skip over empty hosts
                    bos_mgmt_id = row['bos_mgmt_id']
                    auto_upgrade = str2toggle(row['auto_upgrade'])
                    LOG.info('Pushing to %s (%d/%d)' % (host, line_no, total))

                    api = BosApi(host, args.user, args.password)
                    cfg, triple = api.get_config()
                    csvizer = Csvizer(cfg, triple)
                    csvizer.push(row)
                    if not args.check:
                        api.set_config(csvizer.cfg)
                        if bos_mgmt_id:
                            api.set_bos_mgmt_id(bos_mgmt_id)
                        if auto_upgrade is not None:
                            api.uci_init_section('bos', 'auto_upgrade', 'bos')
                            ret = api.uci_set_bool(
                                'bos', 'auto_upgrade', 'enable', auto_upgrade
                            )
                            api.uci_commit('bos')
                        if args.action == 'save_apply':
                            api.apply_config()
                        # password is part of system administration, not miner config
                        if password and args.change_password:
                            LOG.info(
                                'Changing password for %s (%d/%d)'
                                % (host, line_no, total)
                            )
                            api.set_password(password)
                except Exception as ex:
                    log_error(row)
                    if args.ignore:
                        LOG.warning(str(ex))
                    else:
                        raise

    elif args.action == 'apply':
        with open(args.table, newline='') as reader_fd:
            reader = csv.DictReader(reader_fd, fields, restval='')
            for line_no, row in enumerate(reader, 1):
                try:
                    host = row['host'].strip()
                    if (
                        line_no == 1 and host == fields[0] or host == ''
                    ):  # skip first line with headers
                        continue
                    LOG.info('Applying on %s (%d/%d)' % (host, line_no, total))
                    api = BosApi(host, args.user, args.password)
                    if not args.check:
                        api.apply_config()
                except Exception as ex:
                    log_error(row)
                    if args.ignore:
                        LOG.warning(str(ex))
                    else:
                        raise

    else:
        raise RuntimeError('unknown action')


class Csvizer:
    GROUP_SIZE = 3  # how many pools to have in our group
    DEFAULT_GROUP = {'name': 'default', 'pool': []}  # defaults for created pool group

    def __init__(self, cfg, triple):
        cfg_format = cfg['format']
        cfg_format_version = cfg_format['version']
        cfg_format_model = cfg_format['model']

        # json struct with config to work on
        self.cfg = cfg
        self._cfg_format_version = cfg_format_version
        self._cfg_format_model = cfg_format_model
        self._triple = triple

    def is_bos_plus(self) -> bool:
        return self._cfg_format_version.endswith('+')

    def is_antminer_s9(self) -> bool:
        return self._triple.target_full_name == 'am1-s9'

    def is_antminer_x17(self) -> bool:
        return self._triple.target_full_name == 'am2-x17'

    def _get_first_chain_index(self):
        return 1 if self.is_antminer_x17() else 6

    def _has_chain_voltage(self):
        return self.is_antminer_s9()

    def pull(self, row):
        """
        Gather data from existing config into dict suitable for csvwriter.
        row: dict to be populated with values pulled from config
        """
        row['model'] = self._cfg_format_model
        self.pull_groups(row)
        self.pull_hashchain_global(row)
        self.pull_fanctl(row)
        self.pull_tempctl(row)
        # hashchains are currently hardcoded in bosminer
        for chain in range(ord('A'), ord('C') + 1):
            self.pull_hashchain(row, chr(chain))
        # bosminer plus additional features
        if self.is_bos_plus():
            self.pull_power_scaling(row)
            self.pull_autotuning(row)

    def push(self, row):
        """
        Use data in dict to update existing config.
        row: dict to be populated with values pulled from config
        """
        self.push_groups(row)
        self.push_hashchain_global(row)
        self.push_fanctl(row)
        self.push_tempctl(row)
        for chain in range(ord('A'), ord('C') + 1):
            self.push_hashchain(row, chr(chain))
        # options from plus will bug out regular bosminer
        if self.is_bos_plus():
            self.push_power_scaling(row)
            self.push_autotuning(row)

    # pools --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def pull_groups(self, row):
        """
        row: dict to be populated with values pulled from config
        for now, only four pools in first group are considered, rest is ignored
        """
        if 'group' in self.cfg:
            # gather the first group, rest is ignored
            for i, pool in enumerate(
                self.cfg['group'][0].get('pool', [])[: self.GROUP_SIZE]
            ):
                row['pool%d_url' % i] = pool.get('url')
                row['pool%d_user' % i] = pool.get('user')
                row['pool%d_password' % i] = pool.get('password')

    def push_groups(self, row):
        """
        row: dict to pull new values from
        no fancy merging yet, if there is anything in four pools in csv, first group is overwriten with that
        """
        if 'group' not in self.cfg:
            self.cfg['group'] = [{'name': 'default', 'pool': []}]
        pools = []
        for i in range(self.GROUP_SIZE):
            pool = {
                'enabled': True,
                'user': row.get('pool%d_user' % i, ''),
                'password': row.get('pool%d_password' % i, ''),
                'url': str2url(row.get('pool%d_url' % i, '')),
            }
            if pool['url'] and pool['user']:
                pools.append(pool)
            elif pool['url'] and not pool['user']:
                raise InvalidUser('(unspecified)')
            elif not pool['url'] and pool['user']:
                raise InvalidUrl('(unspecified)')
            else:
                pass  # neither url nor user == just ignore this
        if pools:
            self.cfg.setdefault('group', [self.DEFAULT_GROUP])[0]['pool'] = pools

    # autotuning --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def pull_autotuning(self, row):
        """
        row: dict to be populated with values pulled from config
        Otherwise logic is reverse of push.
        """
        row['autotuning'] = toggle2str(self.cfg.get('autotuning', {}).get('enabled'))
        row['power_limit'] = power_limit = self.cfg.get('autotuning', {}).get(
            'psu_power_limit', ''
        )

    def push_autotuning(self, row):
        """
        row: dict to pull new values from
        Depending on power_limit column:
            zero = disable autotuning,
            empty = enable with default wattage
            other number = fixed wattage
        """
        enabled = str2toggle(row['autotuning'])
        power_limit = str2int(row['power_limit'], allow_empty=True)

        if enabled is not None:
            self.cfg.setdefault('autotuning', {})['enabled'] = enabled
        if power_limit is not None:
            self.cfg.setdefault('autotuning', {})['psu_power_limit'] = power_limit

    # fan control --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def pull_fanctl(self, row):
        """
        row: dict to be populated with values pulled from config
        """
        row['fan_speed'] = self.cfg.get('fan_control', {}).get('speed', '')
        row['fan_min'] = self.cfg.get('fan_control', {}).get('min_fans', '')

    def push_fanctl(self, row):
        """
        row: dict to pull new values from
        TODO: how to handle empty fields?
        """

        speed = str2int(row['fan_speed'], allow_empty=True)
        if speed is not None:
            self.cfg.setdefault('fan_control', {})['speed'] = speed

        min_fans = str2int(row['fan_min'], allow_empty=True)
        if min_fans is not None:
            self.cfg.setdefault('fan_control', {})['min_fans'] = min_fans

    # thermal  --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def pull_tempctl(self, row):
        """
        row: dict to be populated with values pulled from config
        """
        row['thermal_mode'] = self.cfg.get('temp_control', {}).get('mode', '')
        row['target_temp'] = self.cfg.get('temp_control', {}).get('target_temp', '')
        row['hot_temp'] = self.cfg.get('temp_control', {}).get('hot_temp', '')
        row['dangerous_temp'] = self.cfg.get('temp_control', {}).get(
            'dangerous_temp', ''
        )

    def push_tempctl(self, row):
        """
        row: dict to pull new values from
        TODO: how to handle empty fields again?
        """
        mode = row['thermal_mode'].strip().lower()
        if mode not in ('auto', 'manual', 'disabled', ''):
            raise InvalidThermalMode(mode)
        target_temp = str2float(row['target_temp'], allow_empty=True)
        hot_temp = str2float(row['hot_temp'], allow_empty=True)
        dangerous_temp = str2float(row['dangerous_temp'], allow_empty=True)

        if mode:
            self.cfg.setdefault('temp_control', {})['mode'] = mode
        if target_temp is not None:
            self.cfg.setdefault('temp_control', {})['target_temp'] = target_temp
        if hot_temp is not None:
            self.cfg.setdefault('temp_control', {})['hot_temp'] = hot_temp
        if dangerous_temp is not None:
            self.cfg.setdefault('temp_control', {})['dangerous_temp'] = dangerous_temp

    # global hashchain  --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def pull_hashchain_global(self, row):
        """
        row: dict to be populated with values pulled from config
        """
        row['asic_boost'] = toggle2str(
            self.cfg.get('hash_chain_global', {}).get('asic_boost')
        )
        row['global_frequency'] = self.cfg.get('hash_chain_global', {}).get(
            'frequency', ''
        )
        row['global_voltage'] = self.cfg.get('hash_chain_global', {}).get('voltage', '')

    def push_hashchain_global(self, row):
        """
        row: dict to pull new values from
        TODO: how to handle empty fields again?
        """
        asic_boost = str2toggle(row['asic_boost'])
        frequency = str2float(row['global_frequency'], allow_empty=True)
        voltage = str2float(row['global_voltage'], allow_empty=True)

        if asic_boost is not None:
            self.cfg.setdefault('hash_chain_global', {})['asic_boost'] = asic_boost
        if frequency is not None:
            self.cfg.setdefault('hash_chain_global', {})['frequency'] = frequency
        if voltage is not None:
            self.cfg.setdefault('hash_chain_global', {})['voltage'] = voltage

    # specific hashchain  --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def get_chain_index(self, chain) -> str:
        return str(self._get_first_chain_index() + ord(chain) - ord('A'))

    def pull_hashchain(self, row, chain):
        """
        row: dict to be populated with values pulled from config
        chain: chain number (6-8)
        """
        chain_index = self.get_chain_index(chain)
        hash_chain = self.cfg.get('hash_chain', {}).get(chain_index, {})
        row['hashchain{}_enabled'.format(chain)] = toggle2str(hash_chain.get('enabled'))
        row['hashchain{}_frequency'.format(chain)] = hash_chain.get('frequency', '')
        row['hashchain{}_voltage'.format(chain)] = hash_chain.get('voltage', '')

    def push_hashchain(self, row, chain):
        """
        row: dict to pull new values from
        chain: chain number (6-8)
        """
        chain_index = self.get_chain_index(chain)
        enabled = str2toggle(row['hashchain{}_enabled'.format(chain)])
        frequency = str2float(
            row['hashchain{}_frequency'.format(chain)], allow_empty=True
        )
        voltage = str2float(row['hashchain{}_voltage'.format(chain)], allow_empty=True)
        if not self._has_chain_voltage() and voltage is not None:
            LOG.warning(
                "Attribute 'voltage' is ignored for {}".format(self._cfg_format_model)
            )
            voltage = None

        if enabled is not None:
            self.cfg.setdefault('hash_chain', {}).setdefault(chain_index, {})[
                'enabled'
            ] = enabled
        if frequency is not None:
            self.cfg.setdefault('hash_chain', {}).setdefault(chain_index, {})[
                'frequency'
            ] = frequency
        if voltage is not None:
            self.cfg.setdefault('hash_chain', {}).setdefault(chain_index, {})[
                'voltage'
            ] = voltage

    # power scaling  --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
    def pull_power_scaling(self, row):
        """
        row: dict to be populated with values pulled from config
        """
        row['power_scaling'] = toggle2str(
            self.cfg.get('power_scaling', {}).get('enabled')
        )
        row['min_power_limit'] = self.cfg.get('power_scaling', {}).get(
            'min_psu_power_limit'
        )
        row['power_step'] = self.cfg.get('power_scaling', {}).get('power_step')
        row['shutdown'] = toggle2str(
            self.cfg.get('power_scaling', {}).get('shutdown_enabled')
        )
        row['shutdown_duration'] = self.cfg.get('power_scaling', {}).get(
            'shutdown_duration'
        )

    def push_power_scaling(self, row):
        enabled = str2toggle(row['power_scaling'])
        min_power = str2int(row['min_power_limit'], allow_empty=True)
        power_step = str2int(row['power_step'], allow_empty=True)
        shutdown = str2toggle(row['shutdown'])
        duration = str2float(row['shutdown_duration'], allow_empty=True)

        if enabled is not None:
            self.cfg.setdefault('power_scaling', {})['enabled'] = enabled
        if min_power is not None:
            self.cfg.setdefault('power_scaling', {})['min_psu_power_limit'] = min_power
        if power_step is not None:
            self.cfg.setdefault('power_scaling', {})['power_step'] = power_step
        if shutdown is not None:
            self.cfg.setdefault('power_scaling', {})['shutdown_enabled'] = shutdown
        if duration is not None:
            self.cfg.setdefault('power_scaling', {})['shutdown_duration'] = duration


class CsvizerError(RuntimeError):
    def __str__(self):
        return '%s: %s' % (self.hint, self.args[0])


class InvalidNumber(CsvizerError):
    hint = 'invalid number'


class InvalidThermalMode(CsvizerError):
    hint = 'invalid thermal mode'


class InvalidToggle(CsvizerError):
    hint = 'invalid toggle'


class InvalidUrl(CsvizerError):
    hint = 'invalid pool url'


class InvalidUser(CsvizerError):
    hint = 'invalid pool user'


class LuciAdminFormParser(html.parser.HTMLParser):
    """
    Helper for getting csrf token in administration web
    """

    def __init__(self):
        super().__init__()
        self.token = None

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if (
            tag == 'input'
            and attrs.get('type') == 'hidden'
            and attrs.get('name') == 'token'
        ):
            self.token = attrs.get('value')


class BosApi:
    """
    Wrapper over bosminer json api. It's main function is to get and hold authentication cookie.
    """

    timeout = 3

    def __init__(self, host, username='root', password=''):
        self.host = host
        self.username = username
        self.password = password
        self.auth()
        self.rpc_id_counter = 0
        self.supported_versions = {
            'am1-s9': ((1, 0), (1, 2)),
            'am2-x17': ((1, 0), (1, 2)),
        }

    @staticmethod
    def _get_target_triple(model: str) -> TargetTriple or None:
        exploded = model.split(maxsplit=1)
        if len(exploded) != 2:
            return None
        vendor, variant = exploded
        if vendor != 'Antminer':
            return None

        from bos_toolbox.platform.descriptor import (
            Target,
            SubTarget,
            S9Variant,
            X17Variant,
        )

        if variant in S9Variant.__members__.values():
            target = Target.am1
            sub_target = SubTarget.s9
            variant = S9Variant(variant)
        elif variant in X17Variant.__members__.values():
            target = Target.am2
            sub_target = SubTarget.x17
            variant = X17Variant(variant)
        else:
            return None

        return TargetTriple(target, sub_target, variant)

    def check_config(self, data) -> TargetTriple:
        """Check if retrieved config is in known format"""
        try:
            format = data['format']
            model = format['model']
            triple = self._get_target_triple(model)
            if not triple:
                raise UnsupportedDevice(model)

            version = format['version']
            supported_versions = self.supported_versions[triple.target_full_name]

            if version.endswith('+'):
                version = version[:-1]
                ver_idx = 1
            else:
                ver_idx = 0

            ver_major, ver_minor = map(lambda v: int(v), version.split('.'))
            ver_range = supported_versions[ver_idx]

            if not (1 <= ver_major <= ver_range[0] and 0 <= ver_minor <= ver_range[1]):
                raise UnsupportedVersion(version)
        except KeyError as ex:
            raise ConfigStructError(str(ex))
        else:
            return triple

    def get_url(self, path):
        """Construct api url from known hostname and given path"""
        url = urllib.parse.urlunsplit(
            ('http', self.host, path, '', '')  # query  # fragment
        )
        return url

    def auth(self):
        """Acquire an authentication token"""
        response = requests.post(
            self.get_url('/cgi-bin/luci/'),
            data={'luci_username': self.username, 'luci_password': self.password},
            allow_redirects=False,
            timeout=self.timeout,
        )
        response.raise_for_status()
        self.authcookie = response.cookies.copy()

    def get_config(self):
        """Load miner config"""
        response = requests.get(
            self.get_url('/cgi-bin/luci/admin/miner/cfg_data'),
            cookies=self.authcookie,
            timeout=self.timeout,
        )
        response.raise_for_status()
        document = response.json()
        if document.get('status', {}).get('code') != 0:
            raise ConfigFetchFailed(document)
        triple = self.check_config(document.get('data'))
        return document['data'], triple

    def set_config(self, data):
        """Write miner config"""
        data['format']['generator'] = 'multiconfiger 0.1'  #
        response = requests.post(
            self.get_url('/cgi-bin/luci/admin/miner/cfg_save'),
            cookies=self.authcookie,
            data=json.dumps({'data': data}),
            timeout=self.timeout,
        )
        response.raise_for_status()
        document = response.json()
        if document.get('status', {}).get('code') != 0:
            raise ConfigStoreFailed(document)
        # print(response.content)

    def apply_config(self):
        """Restart bosminer"""
        response = requests.post(
            self.get_url('/cgi-bin/luci/admin/miner/cfg_apply'),
            cookies=self.authcookie,
            allow_redirects=False,
            timeout=self.timeout,
        )
        response.raise_for_status()
        document = response.json()
        if document.get('status', {}).get('code') != 0:
            raise ConfigApplyFailed(document)

    def rpc(self, endpoint, method, *args):
        """Perform a rpc call"""
        # see https://github.com/openwrt/luci/wiki/JsonRpcHowTo
        rpc = {'method': method, 'params': args, 'id': self.rpc_id()}
        response = requests.post(
            self.get_url('/cgi-bin/luci/rpc/%s' % endpoint),
            cookies=self.authcookie,
            allow_redirects=False,
            data=json.dumps(rpc),
        )
        if response.status_code == 404:
            raise UnsupportedVersion('missing rpc interface')
        response.raise_for_status()
        document = response.json()
        if document.get('id') != rpc['id']:
            raise RpcError('mismatched rpc call (%s)' % document.get('id'))
        if document.get('error'):
            raise RpcError(document.get('error'))
        return document.get('result')

    def rpc_id(self):
        """Provide unique id required by json-rpc spec"""
        self.rpc_id_counter += 1
        return self.rpc_id_counter

    def uci_commit(self, config):
        return self.rpc('uci', 'commit', config)

    def uci_init_section(self, config, section, type):
        """Create section if it does not exist"""
        if self.rpc('uci', 'get', config, section) is None:
            self.rpc('uci', 'set', config, section, type)

    def uci_get(self, config, section, option):
        return self.rpc('uci', 'get', config, section, option)

    def uci_set(self, config, section, option, value):
        return self.rpc('uci', 'set', config, section, option, value)

    def uci_get_bool(self, config, section, option):
        value = self.uci_get(config, section, option)
        # may get nil if option or section is not present
        if value not in ('0', '1', None):
            raise RpcError('Invalid bool value: %s' % value)
        return {'0': False, '1': True}.get(value)

    def uci_get_bool_legacy(self, config, section, option):
        try:
            return self.uci_get_bool(config, section, option)
        except UnsupportedVersion as ex:
            return None

    def uci_set_bool(self, config, section, option, value):
        return self.uci_set(config, section, option, '1' if value else '0')

    def fs_readfile(self, path):
        result = self.rpc('fs', 'readfile', path)
        return result and base64.b64decode(result)

    def fs_writefile(self, path, data: bytes):
        data = base64.b64encode(data).decode()
        return self.rpc('fs', 'writefile', path, data) == 1

    def get_bos_mgmt_id(self):
        result = self.fs_readfile('/etc/bos_mgmt_id')
        return result and result.rstrip().decode()

    def set_bos_mgmt_id(self, value: str) -> bool:
        data = value.encode() + b'\n'
        return self.fs_writefile('/etc/bos_mgmt_id', data)

    def set_password(self, password):

        # we need admin page to obtain csrf token
        response = requests.get(
            self.get_url('/cgi-bin/luci/admin/system/admin'), cookies=self.authcookie
        )
        response.raise_for_status()

        parser = LuciAdminFormParser()
        parser.feed(response.content.decode('utf8'))
        if not parser.token:
            raise MissingToken()

        # save the password. there is more in the form which we skip over
        # using None for filename is undocumented feature of requests lib
        form = {
            'token': (None, parser.token),
            'cbi.submit': (None, '1'),
            'cbid.system._pass.pw1': (None, password),
            'cbid.system._pass.pw2': (None, password),
        }
        response = requests.post(
            self.get_url('/cgi-bin/luci/admin/system/admin'),
            cookies=self.authcookie,
            allow_redirects=False,
            files=form,
        )
        response.raise_for_status()


class BosApiError(RuntimeError):
    pass


class ErrorResponse(BosApiError):
    def __init__(self, document):
        super().__init__(
            '%s (%s)'
            % (
                document.get('status', {}).get('code'),
                document.get('status', {}).get('message'),
            )
        )


class MissingToken(BosApiError):
    pass


class UnsupportedDevice(BosApiError):
    pass


class UnsupportedVersion(BosApiError):
    pass


class ConfigStructError(BosApiError):
    pass


class ConfigFetchFailed(ErrorResponse):
    hint = 'configuration load failed'


class ConfigStoreFailed(ErrorResponse):
    hint = 'configuration save failed'


class ConfigApplyFailed(ErrorResponse):
    hint = 'miner state change failed'


class RpcError(BosApiError):
    pass


# --- helper functions --- --- --- --- --- --- --- --- --- --- --- --- --- --- --- ---
def str2int(s, exception=InvalidNumber, message=None, allow_empty=False):
    """
    Helper function to raise an error if conversion from string to int fails
    """
    if allow_empty and s.strip() == '':
        return None
    try:
        return int(s)
    except ValueError:
        raise exception(message or s)


def str2float(s, exception=InvalidNumber, message=None, allow_empty=False):
    """
    Helper function to raise an error if conversion from string to float fails
    """
    if allow_empty and s.strip() == '':
        return None
    try:
        return float(s)
    except ValueError:
        raise exception(message or s)


def str2toggle(b):
    """
    Turn 'enabled' or 'disabled' strings into bool, or empty string into None
    Exception is thrown for other values.
    """
    b = b.strip().lower()
    if b == '':
        return None
    elif b == 'disabled':
        return False
    elif b == 'enabled':
        return True
    raise InvalidToggle(b)


def toggle2str(b):
    """
    Turn bool config value into enabled/disabled string.
    None is returned as empty string, which should represent value not present.
    (which is why this is not called bool2str)
    """
    return {None: '', False: 'disabled', True: 'enabled'}[b]


def str2url(u):
    """
    Turns string unto url, which is string again so ti does not do anything really.
    But it checks validity, so we can throw more reasonable error then what server does.
    """
    # TODO: this can be dropped once server error messages get better then rust panics
    if u == '':
        return None

    if re.match(
        r'(?:drain|(?:stratum2?\+tcp(?:\+insecure)?)):\/\/[\w\.-]+(?::\d+)?(?:\/[\dA-HJ-NP-Za-km-z]+)?',
        u,
    ):
        return u

    raise InvalidUrl(u)

    if u.strip() == '':
        return None
    parsed = urllib.parse.urlparse(u)
    if parsed.scheme not in ('stratum+tcp', 'stratum2+tcp'):
        raise InvalidUrl(u)
    if parsed.netloc == '':
        raise InvalidUrl(u)


def log_error(data=None):
    import traceback
    import time

    with open('error.log', 'a') as fd:
        print(
            time.strftime('%y-%m-%d %H:%M:%S')
            + (' '.join(sys.argv) if log_error.count == 0 else ''),
            file=fd,
        )
        if data:
            print(data, file=fd)
        traceback.print_exc(file=fd)


log_error.count = 0


def build_arg_parser(parser):
    parser.description = 'Configure mining machines running Braiins OS or Braiins OS+'
    parser.add_argument(
        'action',
        type=lambda x: x.strip().lower(),
        help='load, save, apply or save_apply',
    )
    parser.add_argument('table', help='path to table file in csv format')
    parser.add_argument('-u', '--user', default='root', help='administration username')
    parser.add_argument(
        '-p', '--password', default='', help='administration password or "prompt"'
    )
    parser.add_argument(
        '--change-password',
        default='',
        action='store_true',
        help='allow changing password',
    )
    parser.add_argument(
        '-c', '--check', action='store_true', help='dry run sans writes'
    )
    parser.add_argument('-i', '--ignore', action='store_true', help='no halt on errors')


if __name__ == '__main__':
    try:
        parser = argparse.ArgumentParser()
        build_arg_parser(parser)
        args = parser.parse_args()
        main(parser, args)
    except ErrorResponse as ex:
        # error messages inside responses can be very ugly with rust panic strings
        # handle them with special care until they get better
        log_error()
        sys.exit('error: %s' % ex.hint)
    except Exception as ex:
        log_error()
        sys.exit('error: %s' % ex)
