ruff check fix unsafe

custom_components/ds_air/__init__.py

@@ -1,4 +1,5 @@
 """Platform for DS-AIR of Daikin
+
 https://www.daikin-china.com.cn/newha/products/4/19/DS-AIR/
 """
 
@@ -9,7 +10,7 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.device_registry import DeviceEntry
 
-from .const import CONF_GW, DEFAULT_GW, DEFAULT_HOST, DEFAULT_PORT, DOMAIN
+from .const import CONF_GW, DEFAULT_GW, DOMAIN
 from .ds_air_service import Config, Service
 
 _LOGGER = logging.getLogger(__name__)
@@ -27,7 +28,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     gw = entry.data[CONF_GW]
     scan_interval = entry.data[CONF_SCAN_INTERVAL]
 
-    _LOGGER.debug(f"{host}:{port} {gw} {scan_interval}")
+    _LOGGER.debug("%s:%s %s %s", host, port, gw, scan_interval)
 
     config = Config()
     config.is_c611 = gw == DEFAULT_GW

custom_components/ds_air/climate.py

@@ -74,9 +74,7 @@ async def async_setup_entry(
 ) -> None:
     """Set up the climate devices."""
     service: Service = hass.data[DOMAIN][entry.entry_id]
-    climates = []
-    for aircon in service.get_aircons():
-        climates.append(DsAir(service, aircon))
+    climates = [DsAir(service, aircon) for aircon in service.get_aircons()]
     async_add_entities(climates)
     link = entry.options.get("link")
     sensor_temp_map: dict[str, list[DsAir]] = {}
@@ -249,10 +247,9 @@ def current_temperature(self) -> float | None:
         """Return the current temperature."""
         if self._link_cur_temp:
             return self._attr_current_temperature
-        elif self._device_info.config.is_c611:
+        if self._device_info.config.is_c611:
             return None
-        else:
-            return self._device_info.status.current_temp / 10
+        return self._device_info.status.current_temp / 10
 
     @property
     def target_temperature(self) -> float | None:

custom_components/ds_air/config_flow.py

@@ -105,7 +105,7 @@ async def async_step_init(
     ) -> FlowResult:
         """Manage the options."""
         service = self.hass.data[DOMAIN][self.config_entry.entry_id]
-        self._climates = list(map(lambda state: state.alias, service.get_aircons()))
+        self._climates = [state.alias for state in service.get_aircons()]
         self._len = len(self._climates)
 
         sensors = self.hass.states.async_all("sensor")

custom_components/ds_air/ds_air_service/__init__.py

@@ -3,3 +3,14 @@
 from .dao import UNINITIALIZED_VALUE, AirCon, AirConStatus, Sensor
 from .display import display
 from .service import Service
+
+__all__ = [
+    "Config",
+    "EnumControl",
+    "UNINITIALIZED_VALUE",
+    "AirCon",
+    "AirConStatus",
+    "Sensor",
+    "display",
+    "Service",
+]

custom_components/ds_air/ds_air_service/ctrl_enum.py

@@ -305,11 +305,11 @@ class Voc(IntEnum):
         def __str__(self):
             if self.value == EnumSensor.Voc.STEP_UNUSE:
                 return "不可用"
-            elif self.value == EnumSensor.Voc.STEP_1:
+            if self.value == EnumSensor.Voc.STEP_1:
                 return "优"
-            elif self.value == EnumSensor.Voc.STEP_2:
+            if self.value == EnumSensor.Voc.STEP_2:
                 return "低"
-            elif self.value == EnumSensor.Voc.STEP_3:
+            if self.value == EnumSensor.Voc.STEP_3:
                 return "中"
-            elif self.value == EnumSensor.Voc.STEP_4:
+            if self.value == EnumSensor.Voc.STEP_4:
                 return "高"

custom_components/ds_air/ds_air_service/dao.py

@@ -29,25 +29,25 @@ def unique_id(self):
 class AirConStatus:
     def __init__(
         self,
-        current_temp: int = None,
-        setted_temp: int = None,
-        switch: EnumControl.Switch = None,
-        air_flow: EnumControl.AirFlow = None,
-        breathe: EnumControl.Breathe = None,
-        fan_direction1: EnumControl.FanDirection = None,
-        fan_direction2: EnumControl.FanDirection = None,
-        humidity: EnumControl.Humidity = None,
-        mode: EnumControl.Mode = None,
+        current_temp: int | None = None,
+        setted_temp: int | None = None,
+        switch: EnumControl.Switch | None = None,
+        air_flow: EnumControl.AirFlow | None = None,
+        breathe: EnumControl.Breathe | None = None,
+        fan_direction1: EnumControl.FanDirection | None = None,
+        fan_direction2: EnumControl.FanDirection | None = None,
+        humidity: EnumControl.Humidity | None = None,
+        mode: EnumControl.Mode | None = None,
     ):
-        self.current_temp: int = current_temp
-        self.setted_temp: int = setted_temp
-        self.switch: EnumControl.Switch = switch
-        self.air_flow: EnumControl.AirFlow = air_flow
-        self.breathe: EnumControl.Breathe = breathe
-        self.fan_direction1: EnumControl.FanDirection = fan_direction1
-        self.fan_direction2: EnumControl.FanDirection = fan_direction2
-        self.humidity: EnumControl.Humidity = humidity
-        self.mode: EnumControl.Mode = mode
+        self.current_temp: int | None = current_temp
+        self.setted_temp: int | None = setted_temp
+        self.switch: EnumControl.Switch | None = switch
+        self.air_flow: EnumControl.AirFlow | None = air_flow
+        self.breathe: EnumControl.Breathe | None = breathe
+        self.fan_direction1: EnumControl.FanDirection | None = fan_direction1
+        self.fan_direction2: EnumControl.FanDirection | None = fan_direction2
+        self.humidity: EnumControl.Humidity | None = humidity
+        self.mode: EnumControl.Mode | None = mode
 
 
 class AirCon(Device):
@@ -81,10 +81,9 @@ def __init__(self, config: Config):
 def get_device_by_aircon(aircon: AirCon):
     if aircon.new_air_con:
         return EnumDevice.NEWAIRCON
-    elif aircon.bath_room:
+    if aircon.bath_room:
         return EnumDevice.BATHROOM
-    else:
-        return EnumDevice.AIRCON
+    return EnumDevice.AIRCON
 
 
 class Geothermic(Device):

custom_components/ds_air/ds_air_service/decoder.py

@@ -52,8 +52,7 @@ def decoder(b: bytes, config: Config):
     ):
         if length == 0:
             return HeartbeatResult(), None
-        else:
-            return None, None
+        return None, None
 
     return result_factory(
         struct.unpack("<BHBBBBIBIBH" + str(length - 16) + "sB", b[: length + 4]),
@@ -110,11 +109,11 @@ def result_factory(data: tuple, config: Config):
             result = Sensor2InfoResult(cnt, EnumDevice.SYSTEM)
         else:
             result = UnknownResult(cnt, EnumDevice.SYSTEM, cmd_type)
-    elif (
-        dev_id == EnumDevice.NEWAIRCON.value[1]
-        or dev_id == EnumDevice.AIRCON.value[1]
-        or dev_id == EnumDevice.BATHROOM.value[1]
-        or dev_id == EnumDevice.SENSOR.value[1]
+    elif dev_id in (
+        EnumDevice.NEWAIRCON.value[1],
+        EnumDevice.AIRCON.value[1],
+        EnumDevice.BATHROOM.value[1],
+        EnumDevice.SENSOR.value[1],
     ):
         device = EnumDevice((8, dev_id))
         if cmd_type == EnumCmdType.STATUS_CHANGED.value:
@@ -167,20 +166,20 @@ def read4(self):
         self._pos = pos
         return s
 
-    def read(self, l):
+    def read(self, length: int):
         pos = self._pos
-        s = self._b[pos : pos + l]
-        pos += l
+        s = self._b[pos : pos + length]
+        pos += length
         self._pos = pos
         return s
 
-    def read_utf(self, l):
+    def read_utf(self, length: int):
         pos = self._pos
         try:
-            s = self._b[pos : pos + l].decode("utf-8")
+            s = self._b[pos : pos + length].decode("utf-8")
         except UnicodeDecodeError:
             s = None
-        pos += l
+        pos += length
         self._pos = pos
         return s
 
@@ -472,7 +471,7 @@ def load_bytes(self, b: bytes, config: Config) -> None:
         d = Decode(b)
         self._count = d.read2()
         room_count = d.read1()
-        for i in range(room_count):
+        for _i in range(room_count):
             room = Room()
             room.id = d.read2()
             if self.subbody_ver == 1:
@@ -485,14 +484,14 @@ def load_bytes(self, b: bytes, config: Config) -> None:
                 length = d.read1()
                 room.icon = d.read_utf(length)
             unit_count = d.read2()
-            for j in range(unit_count):
+            for _j in range(unit_count):
                 device = EnumDevice((8, d.read4()))
                 device_count = d.read2()
                 for unit_id in range(device_count):
-                    if (
-                        device == EnumDevice.AIRCON
-                        or device == EnumDevice.NEWAIRCON
-                        or device == EnumDevice.BATHROOM
+                    if device in (
+                        EnumDevice.AIRCON,
+                        EnumDevice.NEWAIRCON,
+                        EnumDevice.BATHROOM,
                     ):
                         dev = AirCon(config)
                         room.air_con = dev
@@ -510,10 +509,7 @@ def load_bytes(self, b: bytes, config: Config) -> None:
                         dev = Sensor()
                         self.sensors.append(dev)
                         room.sensor_room = True
-                    elif (
-                        device == EnumDevice.VENTILATION
-                        or device == EnumDevice.SMALL_VAM
-                    ):
+                    elif device in (EnumDevice.VENTILATION, EnumDevice.SMALL_VAM):
                         dev = Ventilation()
                         room.ventilation = dev
                         dev.is_small_vam = device == EnumDevice.SMALL_VAM
@@ -787,10 +783,10 @@ def __init__(self, cmd_id: int, target: EnumDevice):
     def load_bytes(self, b: bytes, config: Config) -> None:
         d = Decode(b)
         room_size = d.read1()
-        for i in range(room_size):
+        for _i in range(room_size):
             room_id = d.read1()
             unit_size = d.read1()
-            for j in range(unit_size):
+            for _j in range(unit_size):
                 aircon = AirCon(config)
                 aircon.unit_id = d.read1()
                 aircon.room_id = room_id

custom_components/ds_air/ds_air_service/display.py

@@ -2,21 +2,23 @@
 
 
 def display(o, d="") -> str:
-    if type(o) == int or type(o) == str or type(o) == bool or type(o) == float:
+    if isinstance(o, (int, str, bool, float)):
         return str(o)
-    elif isinstance(o, Enum):
+
+    if isinstance(o, Enum):
         return o.name
-    elif type(o) == list:
+
+    if isinstance(o, list):
         st = "["
         for i in range(len(o)):
             st += "\n" + d + str(i) + ": " + display(o[i], d + "    ")
         st += "]"
         return st
-    else:
-        li = dir(o)
-        st = ("\033[31m%s:\033[0m" % o.__class__.__name__) + " {"
-        for i in li:
-            if (not i.startswith("_")) and (not callable(o.__getattribute__(i))):
-                st += "\n" + d + i + ": " + display(o.__getattribute__(i), d + "    ")
-        st += "}"
-        return st
+
+    li = dir(o)
+    st = f"\033[31m{o.__class__.__name__}:\033[0m" + " {"
+    for i in li:
+        if (not i.startswith("_")) and (not callable(o.__getattribute__(i))):
+            st += "\n" + d + i + ": " + display(o.__getattribute__(i), d + "    ")
+    st += "}"
+    return st

custom_components/ds_air/ds_air_service/param.py

@@ -184,10 +184,7 @@ def generate_subbody(self, s: Encode, config: Config) -> None:
             if dev.fan_volume != EnumFanVolume.NO:
                 flag = flag | t.AIR_FLOW
             if config.is_new_version:
-                if (
-                    dev.fan_direction1 != EnumFanDirection.FIX
-                    and dev.fan_direction2 != EnumFanDirection.FIX
-                ):
+                if EnumFanDirection.FIX not in (dev.fan_direction1, dev.fan_direction2):
                     flag = flag | t.FAN_DIRECTION
                 if dev.bath_room or dev.three_d_fresh_allow:
                     flag = flag | t.BREATHE

custom_components/ds_air/ds_air_service/service.py

@@ -51,12 +51,13 @@ def do_connect(self):
         self._s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         try:
             self._s.connect((self._host, self._port))
-            _log("connected")
-            return True
         except OSError as exc:
             _log("connected error")
             _log(str(exc))
             return False
+        else:
+            _log("connected")
+            return True
 
     def send(self, p: Param):
         self._locker.acquire()