优化baseAnalyst.py的获取业务数据的数据库访问方法loadData

dataAnalysisBehavior/behavior/baseAnalyst.py

@@ -1,3 +1,5 @@
+# baseAnalyst.py
+
 import concurrent.futures
 import logging
 from abc import ABC, abstractmethod
@@ -489,21 +491,39 @@ class BaseAnalyst(ABC):
     def loadData(self, powerFarmID: str, timeGranularity: str, turbineCode: str, select: str, condition: str) -> pd.DataFrame:
         selectStr = ", ".join(f"{field}" for field in select)
         businessDB: DatabaseUtil = GetBusinessDbUtil()
+        
+
+        # query_text = f"""SELECT wind_turbine_number,time_stamp, active_power,wind_velocity,cabin_position,true_wind_direction, rotor_speed, generator_speed,actual_torque,yaw_error1,outside_cabin_temperature,cabin_temperature,main_bearing_temperature,gearboxmedium_speed_shaftbearing_temperature,gearbox_low_speed_shaft_bearing_temperature,gearbox_high_speed_shaft_bearing_temperature,generatordrive_end_bearing_temperature,generatornon_drive_end_bearing_temperature,generator_winding1_temperature,pitch_angle_blade_1,pitch_angle_blade_2,pitch_angle_blade_3,front_back_vibration_of_the_cabin,side_to_side_vibration_of_the_cabin
+        #                    FROM `{dataBatchNum}_{timeGranularity}`
+        #                    WHERE wind_turbine_number IN ({turbineCode}) AND {condition}"""
+        query_text = f"""SELECT {selectStr}
+                            FROM `{powerFarmID}_{timeGranularity}`
+                            WHERE wind_turbine_number IN ({turbineCode}) AND {condition}"""
+              
+        # select = [Field_DeviceCode, Field_Time, Field_ActiverPower, Field_WindSpeed, Field_NacPos, Field_WindDirection, Field_RotorSpeed, Field_GeneratorSpeed, Field_GeneratorTorque, Field_AngleIncluded, Field_EnvTemp, Field_NacTemp, Field_MainBearTemp, Field_GbMsBearTemp, Field_GbLsBearTemp,
+        #       Field_GbHsBearTemp, Field_GeneratorDE, Field_GeneratorNDE, Field_GenWiTemp1, Field_PitchAngel1, Field_PitchAngel2, Field_PitchAngel3, Field_NacFbVib, Field_NacLrVib]
+
+        chunks = []
         with businessDB.session_scope() as session:
-            # query_text = f"""SELECT wind_turbine_number,time_stamp, active_power,wind_velocity,cabin_position,true_wind_direction, rotor_speed, generator_speed,actual_torque,yaw_error1,outside_cabin_temperature,cabin_temperature,main_bearing_temperature,gearboxmedium_speed_shaftbearing_temperature,gearbox_low_speed_shaft_bearing_temperature,gearbox_high_speed_shaft_bearing_temperature,generatordrive_end_bearing_temperature,generatornon_drive_end_bearing_temperature,generator_winding1_temperature,pitch_angle_blade_1,pitch_angle_blade_2,pitch_angle_blade_3,front_back_vibration_of_the_cabin,side_to_side_vibration_of_the_cabin
-            #                    FROM `{dataBatchNum}_{timeGranularity}`
-            #                    WHERE wind_turbine_number IN ({turbineCode}) AND {condition}"""
-            query_text = f"""SELECT {selectStr}
-                               FROM `{powerFarmID}_{timeGranularity}`
-                               WHERE wind_turbine_number IN ({turbineCode}) AND {condition}"""
-            query_result = session.execute(text(query_text)).fetchall()
-
-            # select = [Field_DeviceCode, Field_Time, Field_ActiverPower, Field_WindSpeed, Field_NacPos, Field_WindDirection, Field_RotorSpeed, Field_GeneratorSpeed, Field_GeneratorTorque, Field_AngleIncluded, Field_EnvTemp, Field_NacTemp, Field_MainBearTemp, Field_GbMsBearTemp, Field_GbLsBearTemp,
-            #       Field_GbHsBearTemp, Field_GeneratorDE, Field_GeneratorNDE, Field_GenWiTemp1, Field_PitchAngel1, Field_PitchAngel2, Field_PitchAngel3, Field_NacFbVib, Field_NacLrVib]
-
-            dataFrame = pd.DataFrame(query_result, columns=select)
-
-        return dataFrame
+            # query_result = session.execute(text(query_text)).fetchall()
+            
+            # dataFrame = pd.DataFrame(query_result, columns=select)
+
+        # return dataFrame
+
+            conn=session.connection()
+            result=conn.execution_options(stream_results=True).execute(text(query_text))
+             # 每次取 100000 行，可根据实际情况调整
+            while True:
+                rows = result.fetchmany(100000)
+                if not rows:
+                    break
+                chunks.append(pd.DataFrame(rows, columns=select))
+        
+        if not chunks:
+            return pd.DataFrame()
+        
+        return pd.concat(chunks, ignore_index=True)
 
     def dataProcess(self, powerFarmID: str, dataBatchNum: str, timeGranularity: str, turbineCode: str, select: str, customCondition) -> Tuple[pd.DataFrame, str, str]:
         self.logger.info(
@@ -622,7 +642,7 @@ class BaseAnalyst(ABC):
                 dataFrameOfTurbines = pd.DataFrame()
                 if timeGranularity in ['fault', 'warn']:
                     select_conditions = self.selectAllFaultCondition(conf)
-                maxWorkers = 5
+                maxWorkers = 3
                 with concurrent.futures.ThreadPoolExecutor(max_workers=maxWorkers) as executor:
                     futures = [
                         executor.submit(self.dataProcess, conf.dataContract.dataFilter.powerFarmID,

dataContract/README.MD

@@ -39,6 +39,7 @@
 |机组发电机转速和转矩分析|algorithm.generatorSpeedTorqueAnalyst|GeneratorSpeedTorqueAnalyst|executeAnalysis|minute|speed_torque|
 |静态偏航误差分析|||||
 |机组静态偏航误差分析|algorithm.yawErrorAnalyst|YawErrorAnalyst|executeAnalysis|second|yaw_error|
+|机组动态偏航误差分析|algorithm.yawErrorDensityAnalyst|YawErrorDensityAnalyst|executeAnalysis|second|yaw_error_density|
 |变桨特性分析|||||
 |机组最小桨距角分析|algorithm.minPitchAnalyst|MinPitchAnalyst|executeAnalysis|second|min_pitch|
 |变桨-转速-转矩（功率）协调分析|||||

repositoryZN/utils/rdbmsUtil/databaseUtil.py

@@ -1,3 +1,5 @@
+# databaseUtil.py
+
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, declarative_base
 from sqlalchemy import Column, Integer, String
@@ -10,7 +12,7 @@ from utils.rdbmsUtil.factoryRegistry import FactoryRegistry
 
 
 class DatabaseUtil:
-    def __init__(self, url, pool_size=5, max_overflow=10, pool_recycle=1000, pool_pre_ping=True, connect_timeout=10,read_timeout=120):
+    def __init__(self, url, pool_size=5, max_overflow=10, pool_recycle=100, pool_pre_ping=True, connect_timeout=10,read_timeout=1800):
         self.engine = create_engine(
             url,
             pool_size=pool_size,                # 设置最大连接池大小

repositoryZN/utils/rdbmsUtil/unitTest.py

@@ -6,26 +6,49 @@ from utils.rdbmsUtil.modelUserOperations import UserOperations
 from utils.rdbmsUtil.modelCompany import Company, company_factory
 from utils.rdbmsUtil.modelCompanyOperations import CompanyOperations
 
+from sqlalchemy.sql import text
+import pandas as pd
 
-db = DatabaseUtil('sqlite:///mydatabase.db')
-Base.metadata.create_all(db.engine)
 
-companys = [Company(id=1, name="DS-Tech"), Company(id=2, name="ZN-Tech")]
-users = [User(name="Alice", age=31, companyid=1),
-         User(name="Bob", age=25, companyid=2)]
+def access_sqlite():
+    db = DatabaseUtil('sqlite:///mydatabase.db')
+    Base.metadata.create_all(db.engine)
 
+    companys = [Company(id=1, name="DS-Tech"), Company(id=2, name="ZN-Tech")]
+    users = [User(name="Alice", age=31, companyid=1),
+            User(name="Bob", age=25, companyid=2)]
 
-with db.session_scope() as session:
-    company_ops = CompanyOperations()
 
-    user_ops = UserOperations()
-    
-    company_ops.upsert_companys(session, companys)
-    user_ops.upsert_users(session, users)
-    # session.flush()  # 确保所有的 SQL 操作都被推送到数据库
+    with db.session_scope() as session:
+        company_ops = CompanyOperations()
 
-    user_ops.update(session, name="Alice", age=32)
+        user_ops = UserOperations()
+        
+        company_ops.upsert_companys(session, companys)
+        user_ops.upsert_users(session, users)
+        # session.flush()  # 确保所有的 SQL 操作都被推送到数据库
 
-    result = user_ops.get_user(session, name="Alice", age=32)
-    for user in result:
-        print(user)
+        user_ops.update(session, name="Alice", age=32)
+
+        result = user_ops.get_user(session, name="Alice", age=32)
+        for user in result:
+            print(user)
+
+
+def access_tidb():
+    print('access TiDB')
+    DatabaseUtil(url='mysql+pymysql://root:123456@192.168.50.234:4000/energy',
+                        pool_size=5,
+                        max_overflow=3,
+                        connect_timeout=1800)
+    db = DatabaseUtil('mysql+pymysql://root:123456@192.168.50.234:4000/energy')
+
+    with db.session_scope() as session:
+        query_text = f"""SELECT * FROM analysis_result  """
+        query_result = session.execute(text(query_text)).fetchall()
+
+        dataFrame = pd.DataFrame(query_result)
+
+        print(dataFrame.head(5))
+
+access_tidb()

requirements_windows.txt

@@ -0,0 +1,171 @@
+absl-py==2.3.1
+aiohappyeyeballs==2.6.1
+aiohttp==3.11.14
+aiosignal==1.3.2
+argon2-cffi==23.1.0
+argon2-cffi-bindings==21.2.0
+asgiref==3.8.1
+asttokens==3.0.0
+astunparse==1.6.3
+attrs==25.3.0
+autots==0.6.21
+backcall==0.2.0
+beautifulsoup4==4.13.3
+bleach==6.2.0
+blinker==1.9.0
+build==1.2.2.post1
+certifi==2025.1.31
+cffi==1.17.1
+charset-normalizer==3.4.1
+click==8.1.8
+cobble==0.1.4
+colorama==0.4.6
+comtypes==1.4.10
+contourpy==1.3.1
+cycler==0.12.1
+datasets==3.4.1
+debugpy==1.8.13
+decorator==5.2.1
+defusedxml==0.7.1
+dill==0.3.8
+Django==5.1.7
+djangorestframework==3.15.2
+docopt==0.6.2
+drf-yasg==1.21.10
+et_xmlfile==2.0.0
+executing==2.2.0
+faiss-cpu==1.11.0
+fastjsonschema==2.21.1
+filelock==3.18.0
+Flask==3.1.1
+flatbuffers==25.9.23
+fonttools==4.56.0
+frozenlist==1.5.0
+fsspec==2024.12.0
+gast==0.6.0
+geographiclib==2.0
+geopy==2.4.1
+google-pasta==0.2.0
+greenlet==3.1.1
+grpcio==1.75.0
+h5py==3.14.0
+huggingface-hub==0.31.4
+idna==3.10
+inflection==0.5.1
+ipython==8.12.3
+itsdangerous==2.2.0
+jedi==0.19.2
+Jinja2==3.1.6
+joblib==1.4.2
+jsonschema==4.23.0
+jsonschema-specifications==2024.10.1
+jupyter_client==8.6.3
+jupyter_core==5.7.2
+jupyterlab_pygments==0.3.0
+kaleido==0.1.0.post1
+keras==3.11.3
+kiwisolver==1.4.8
+libclang==18.1.1
+lxml==5.3.2
+mammoth==1.9.0
+Markdown==3.9
+markdown-it-py==4.0.0
+MarkupSafe==3.0.2
+matplotlib==3.10.1
+matplotlib-inline==0.1.7
+mdurl==0.1.2
+minio==7.2.15
+mistune==3.1.2
+ml_dtypes==0.5.3
+modelscope==1.24.0
+mpmath==1.3.0
+multidict==6.2.0
+multiprocess==0.70.16
+namex==0.1.0
+narwhals==1.31.0
+nbclient==0.10.2
+nbconvert==7.16.6
+nbformat==5.10.4
+networkx==3.4.2
+numpy==2.2.4
+openpyxl==3.1.5
+opt_einsum==3.4.0
+optree==0.17.0
+packaging==24.2
+pandas==2.2.3
+pandocfilters==1.5.1
+parso==0.8.4
+patsy==1.0.1
+pickleshare==0.7.5
+pillow==11.1.0
+pip-tools==7.4.1
+pipdeptree==2.25.1
+pipreqs==0.5.0
+platformdirs==4.3.6
+plotly==5.19.0
+plotly-express==0.4.1
+prompt_toolkit==3.0.50
+propcache==0.3.1
+protobuf==6.32.1
+psutil==7.0.0
+pure_eval==0.2.3
+pyarrow==19.0.1
+pycparser==2.22
+pycryptodome==3.22.0
+Pygments==2.19.1
+PyMySQL==1.1.1
+pyparsing==3.2.1
+pyproject_hooks==1.2.0
+python-dateutil==2.9.0.post0
+python-docx==1.1.2
+pytz==2025.1
+pywin32==310
+PyYAML==6.0.2
+pyzmq==26.3.0
+referencing==0.36.2
+regex==2024.11.6
+requests==2.32.3
+rich==14.1.0
+rpds-py==0.23.1
+safetensors==0.5.3
+scikit-learn==1.6.1
+scipy==1.15.2
+seaborn==0.13.2
+sentence-transformers==4.1.0
+six==1.17.0
+soupsieve==2.6
+spark-parser==1.8.9
+SQLAlchemy==2.0.39
+sqlparse==0.5.3
+stack-data==0.6.3
+statsmodels==0.14.4
+sympy==1.14.0
+tenacity==9.0.0
+tensorboard==2.20.0
+tensorboard-data-server==0.7.2
+tensorflow==2.20.0
+termcolor==3.1.0
+threadpoolctl==3.6.0
+tinycss2==1.4.0
+tokenizers==0.21.1
+torch==2.7.0
+torchaudio==2.7.0
+torchvision==0.22.0
+tornado==6.4.2
+tqdm==4.67.1
+traitlets==5.14.3
+transformers==4.52.1
+typing_extensions==4.12.2
+tzdata==2025.1
+uncompyle6==3.9.2
+uritemplate==4.1.1
+urllib3==2.3.0
+wcwidth==0.2.13
+webencodings==0.5.1
+Werkzeug==3.1.3
+windrose==1.9.2
+wrapt==1.17.3
+xdis==6.1.3
+xxhash==3.5.0
+yarg==0.1.9
+yarl==1.18.3