SRP：将数据科学重构超越课程-DEV365 开发者社区

图片图片是一个大交响乐团：木管乐器，黄铜，弦乐和打击乐，每个部分都与众不同而又和谐地团结在一起。导体挥舞着指挥棒，每个仪器都扮演其角色，导致了一个凝聚力，宏伟的整体。这种音乐和谐是乐器的完美相互作用，是坚实的原则试图在我们的数据科学画布中灌输的内容，单一责任原则（SRP）作为工具。

woohoo！你还在这里。您渴望更多！您不仅走了这么远，而且还征服了"Building the Bedrock: Employing SOLID Principles in Data Science"。我在这里。现在，随着我们腹部的大火，我们都准备好探索者SRP如何革新数据科学代码基础。

固体中的 s ：窗帘后面的大师

srp不仅仅是软件设计原则。这是一种艺术形式，是角色和责任的细致编舞。除了课程的范围之外，SRP还唱了计算机科学交响乐，确保每个“仪器”（无论是函数，班级还是文件）都与其独特的注释共鸣，但有助于该项目的集体旋律。

如果文件，函数或类具有多个责任，则将耦合。对一个责任的改变通常会导致修改另一个责任。它不仅很难维护，而且也不可靠。

给出最终目标是生产不仅强大且可靠而且易于维护的软件，SRP成为我们的软件设计原理的心灵和灵魂。当软件的每个部分都关注一个主要责任时，它变得不那么纠结，因此更容易理解和修改。

让我们代码：创建数据预处理管道

在一个繁华的乐团大厅里，想象一下jupyter笔记本与Pythonic Pandas构图共鸣：数据转换像交响曲一样流动，进行管道数据转换，创建和谐音调。但是随着音乐会的进行，旋律互动，变得复杂且难以遵循。

输入SRP。像精湛的作曲家一样，我们重新布置了音乐作品，将笔记本的复杂部分重构为单独的分数。现在，每个分数都具有独特的旋律，使整个构图更具弹性，并且可以轻而易举。我们曾经纠结的交响曲现在是一个和谐的数据和声芭蕾舞，每个芭蕾舞都有自己的聚光灯。

# Task Prototype. Each Task has to implement *run* method 
# file: prototypes.py
from abc import ABC, abstractmethod

class Task(ABC):
    @property
    def name(self) -> str:
        return self.__class__.__name__

    @abstractmethod
    def run(self, *args, **kwargs):
        pass

现在，做一件事情的乐器

# Tasks that can be divided into their own files
# from prototypes import Task

from typing import Literal

import numpy as np
import pandas as pd
from sklearn.compose import make_column_transformer
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import FunctionTransformer, OneHotEncoder
from sklearn import set_config

from srp import config

set_config(transform_output="pandas")


transformer = make_column_transformer(
    (
        FunctionTransformer(np.log),[
            config.COLUMNS_TO_LOG_TRANSFORM
        ],
    ),
    (
        OneHotEncoder(sparse_output=False),[
            config.COLUMNS_TO_ONEHOTENCODE,
        ],
    ),
    verbose_feature_names_out=False,
    remainder="passthrough",
)

class DropZerosTask(Task):
    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        cleaned_data = data[~data.select_dtypes("number").eq(0).any(axis=1)]
        return cleaned_data

class DropColumnsTask(Task):
    def __init__(self, columns: list[str]):
        self.columns = columns

    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        cleaned_data = data.drop(columns=self.columns)
        return cleaned_data

# TODO: decorator to save and load transformer
class TransformerTask(Task):
    def __init__(self, stage: Literal["train", "predict"] = "train"):
        self.stage = stage

    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        if self.stage == "predict":
            cleaned_data = transformer.transform(data)
        else:
            cleaned_data = transformer.fit_transform(data)

        return cleaned_data

class Floats2IntsTask(Task):
    def __init__(self, columns: list[str]):
        self.columns = columns

    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        cleaned_data = data
        cleaned_data[self.columns] = cleaned_data[self.columns].transform(
            pd.to_numeric,
            errors="coerce",
            downcast="integer",
        )
        return cleaned_data

这种模块化方法具有灵活性。随着需求和技术的变化，调整或替换单个零件而不是大修整个系统要简单得多。

时间到 note 导体，负责添加和呼叫工具：

# DataTasks gather and runner
# file: data/datatasks.py
# from prototypes import Task

from __future__ import annotations
from queue import PriorityQueue
import pandas as pd
from loguru import logger


class DataTasks:
    def __init__(self, tasks: PriorityQueue = None) -> DataTasks:
        if tasks is None:
            self.tasks = PriorityQueue()

    def set_task(self, priority: int, task: Task) -> DataTasks:
        self.tasks.put((priority, task))
        return self

    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        while not self.tasks.empty():
            _, task = self.tasks.get()
            logger.debug(f"priority: {_}, task: {task.name}")
            data = task.run(data)
        return data

由于Tasks仅需要实现返回DataFrame的run方法，因此该软件的这一部分不会随Tasks的更改而变化。因此，制作既适合未来变化又适合其操作弹性的作品。

最后，是一个大交响曲。一条坚固，可维护且适应变化的无缝管道。

# Implementation
# from data.datatasks import DataTasks
# from data.tasks import ...
from srp import config

def process_data(data: pd.DataFrame) -> pd.DataFrame:
    data_chain = DataTasks()
    (
        data_chain.set_task(priority=2, task=DropZerosTask())
        .set_task(
            priority=1,
            task=DropColumnsTask(
                columns=[
                    config.COLUMNS_TO_DROP
                ]
            ),
        )
        .set_task(priority=3, task=TransformerTask())
        .set_task(
            priority=4,
            task=Floats2IntsTask(
                columns=[
                    config.COLUMNS_TO_FLOATS_TO_INTEGER
                ]
            ),
        )
        # Add more or remove tasks to the chain as needed
    )

    # Send data through the chain
    return data_chain.run(data)

# in the main
# import process_data

if __name__ == "__main__":
    import pandas as pd

    URI = "https://raw.githubusercontent.com/Ankit152/Fish-Market/main/Fish.csv"
    dataf = pd.read_csv(URI)
    clean_data = process_data(dataf)

引导星星：避免黑洞

虽然SRP和设计模式提供了一种可能性的宇宙，但要考虑到过度工程和复杂性的黑洞。并非每个问题都需要这个交响曲。有时，简单的旋律会做到。我们仍然需要定制我们对项目叙事的方法。

SRP开始将我们的数据科学项目转变为杰作，具有凝聚力但复杂，简单而精致的杰作。每个组件都像完美的仪器一样发挥作用，为整个交响曲做出了贡献，并确保我们的项目可维护，可扩展且可扩展。

宇宙杰作的旅程并没有结束。它刚刚开始。我们的下一个坚实的指导明星是“ O”。开放原理（OCP）。 OCP指导我们建立开放式扩展名但已关闭的单位。本系列下一篇文章的传奇。

Up Next：“ OCP：重构数据科学项目”

在此之前，请继续编码数据科学固体。

const artist = "Shahnoza Bekbulaeva";
console.log(artist)