<commit_before>import os
import fnmatch
import serial
import requests
import wakeonlan as wol
from typing import Union

class ChannelHandler:
    def __init__(self, channels):
        self.channels = channels

    def handle_code(self, channel: str, data: Union[str, dict]) -> Union[None, tuple[str, int]]:
        raise NotImplementedError

class WakeOnLanHandler(ChannelHandler):
    def __init__(self):
        super().__init__(["WOL"])

    def handle_code(self, channel: str, data: Union[str, dict]) -> Union[None, tuple[str, int]]:
        wol.send_magic_packet(data)
        return True

class HyperionWebHandler(ChannelHandler):
    REQ_ADDR = "http://localhost:1234"

    def __init__(self):
        super().__init__(["LED"])

    def handle_code(self, _, data: dict) -> Union[None, tuple[str, int]]:
        try:
            requests.post(self.REQ_ADDR + "/" + data["endpoint"], data=data["data"])
        except requests.ConnectionError:
            return "Service Unavailable", 503

class ArduinoHandler(ChannelHandler):
    def __init__(self):
        super().__init__(["MHZ433", "NEXA", "IR"])

        if os.environ.get("DEBUG") is None:
            # Initialize COM-port
            self.ser = self.init_comport()

    def handle_code(self, channel, data) -> Union[None, tuple[str, int]]:
        if channel == "IR":
            self.ser_write(data + ";") # Send IR code to Arduino

        elif channel == "MHZ433" or channel == "NEXA":
            self.ser_write(channel + ": " + data + ";")

    def ser_write(self, data):
        if os.environ.get("DEBUG") is None:
            self.ser.write(data.encode())

    @staticmethod
    def init_comport():
        # Find the right USB port
        matches = []

        for root, _, filenames in os.walk("/dev"):
            for filename in fnmatch.filter(filenames, "ttyUSB*"):
                matches.append(os.path.join(root, filename))

        ser          = serial.Serial()
        ser.port     = matches[-1]
        ser.baudrate = 9600
        ser.timeout  = 0
        ser.xonxoff  = False       # Disable software flow control
        ser.rtscts   = False       # Disable hardware (RTS/CTS) flow control
        ser.dsrdtr   = False       # Disable hardware (DSR/DTR) flow control

        if ser.isOpen():
            print("### Serial conenction already open!")
        else:
            try:
                ser.open()
                print(" * Serial connection open!")
            except Exception as e:
                print(" * Error open serial port: " + str(e))
        return ser
<commit_msg>Make ChannelHandler an abstract class<commit_after>import os
import fnmatch
import serial
import requests
import wakeonlan as wol
from typing import Union
from abc import ABC, abstractmethod

class ChannelHandler(ABC):
    def __init__(self, channels):
        self.channels = channels

    @abstractmethod
    def handle_code(self, channel: str, data: Union[str, dict]) -> Union[None, tuple[str, int]]:
        pass

class WakeOnLanHandler(ChannelHandler):
    def __init__(self):
        super().__init__(["WOL"])

    def handle_code(self, channel: str, data: Union[str, dict]) -> Union[None, tuple[str, int]]:
        wol.send_magic_packet(data)
        return True

class HyperionWebHandler(ChannelHandler):
    REQ_ADDR = "http://localhost:1234"

    def __init__(self):
        super().__init__(["LED"])

    def handle_code(self, _, data: dict) -> Union[None, tuple[str, int]]:
        try:
            requests.post(self.REQ_ADDR + "/" + data["endpoint"], data=data["data"])
        except requests.ConnectionError:
            return "Service Unavailable", 503

class ArduinoHandler(ChannelHandler):
    def __init__(self):
        super().__init__(["MHZ433", "NEXA", "IR"])

        if os.environ.get("DEBUG") is None:
            # Initialize COM-port
            self.ser = self.init_comport()

    def handle_code(self, channel, data) -> Union[None, tuple[str, int]]:
        if channel == "IR":
            self.ser_write(data + ";") # Send IR code to Arduino

        elif channel == "MHZ433" or channel == "NEXA":
            self.ser_write(channel + ": " + data + ";")

    def ser_write(self, data: str):
        if os.environ.get("DEBUG") is None:
            self.ser.write(data.encode())

    @staticmethod
    def init_comport():
        # Find the right USB port
        matches = []

        for root, _, filenames in os.walk("/dev"):
            for filename in fnmatch.filter(filenames, "ttyUSB*"):
                matches.append(os.path.join(root, filename))

        ser          = serial.Serial()
        ser.port     = matches[-1]
        ser.baudrate = 9600
        ser.timeout  = 0
        ser.xonxoff  = False       # Disable software flow control
        ser.rtscts   = False       # Disable hardware (RTS/CTS) flow control
        ser.dsrdtr   = False       # Disable hardware (DSR/DTR) flow control

        if ser.isOpen():
            print("### Serial conenction already open!")
        else:
            try:
                ser.open()
                print(" * Serial connection open!")
            except Exception as e:
                print(" * Error open serial port: " + str(e))
        return ser

{'fraction_non_alphanumeric': 0.12315648343229267, 'fraction_numerical': 0.007278299176402988, 'mean_word_length': 2.269881026925485, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1258288', 'n_tokens_mistral': 1580, 'n_tokens_neox': 1511, 'n_words': 408}