add new endpoint to select OH equipment

main
Cizz22 11 months ago
parent a110836242
commit 95b1928f56

@ -0,0 +1,17 @@
# EditorConfig is awesome: http://EditorConfig.org
root = true
[*]
indent_style = space
indent_size = 4
trim_trailing_whitespace = true
insert_final_newline = true
end_of_line = lf
[*.md]
trim_trailing_whitespace = false
[Makefile]
indent_style = tab
trim_trailing_whitespace = false

6
.gitattributes vendored

@ -0,0 +1,6 @@
# Set the default behavior for all files.
* text=auto eol=lf
# Normalized and converts to native line endings on checkout.
*.py text
*.pyx text

@ -1,44 +1,42 @@
# Quick Start:
#
# pip install pre-commit
# pre-commit install && pre-commit install -t pre-push
# pre-commit run --all-files
#
# To Skip Checks:
#
# git commit --no-verify
fail_fast: false
default_language_version:
python: python3.11.2
python: python3
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
# ruff version.
rev: v0.7.0
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: check-ast
- id: check-builtin-literals
- id: check-merge-conflict
- id: check-yaml
- id: check-toml
- repo: https://github.com/nbQA-dev/nbQA
rev: 1.7.1
hooks:
- id: nbqa-isort
- repo: https://github.com/PyCQA/isort
rev: 5.13.2
hooks:
- id: isort
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.2.1
hooks:
# Run the linter.
#
# When running with --fix, Ruff's lint hook should be placed before Ruff's formatter hook,
# and before Black, isort, and other formatting tools, as Ruff's fix behavior can output code changes that require reformatting.
- id: ruff
args: [--fix]
# Run the formatter.
- id: ruff-format
- id: ruff-format
types_or: [python, pyi, jupyter]
# Typos
- repo: https://github.com/crate-ci/typos
rev: v1.26.1
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: 'v0.2.0'
hooks:
- id: typos
exclude: ^(data/dispatch-sample-data.dump|src/dispatch/static/dispatch/src/|src/dispatch/database/revisions/)
- id: ruff
types_or: [python, pyi, jupyter]
args: [ --fix, --exit-non-zero-on-fix ]
# Pytest
- repo: local
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.8.0
hooks:
- id: tests
name: run tests
entry: pytest -v tests/
- id: mypy
language: system
types: [python]
stages: [push]
pass_filenames: false
args: ['.']

@ -0,0 +1,23 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "PyDebug: Current File",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"args": [],
"justMyCode": true
},
{
"name": "PyDebug: Main File",
"type": "debugpy",
"request": "launch",
"program": "${workspaceFolder}/main.py",
"console": "integratedTerminal",
"args": [],
"justMyCode": true
}
]
}

@ -0,0 +1,152 @@
{
"editor.tabSize": 4,
"editor.rulers": [
120
],
"editor.renderWhitespace": "trailing",
"editor.suggestSelection": "first",
"editor.formatOnSave": true,
"editor.defaultFormatter": "esbenp.prettier-vscode",
"editor.stickyScroll.enabled": false,
"editor.bracketPairColorization.enabled": false,
"editor.cursorSmoothCaretAnimation": "on",
"editor.suggest.preview": true,
"terminal.integrated.defaultProfile.windows": "Command Prompt",
"debug.onTaskErrors": "debugAnyway",
"explorer.compactFolders": false,
"explorer.confirmDragAndDrop": false,
"explorer.confirmDelete": false,
"explorer.copyRelativePathSeparator": "/",
"files.autoSave": "onFocusChange",
"files.exclude": {
"node_modules/**/*": true,
"**/.classpath": true,
"**/.project": true,
"**/.settings": true,
"**/.factorypath": true
},
"files.associations": {
"*.pyx": "cython",
".clang*": "yaml",
"*.gpj": "jsonc",
"*.gvw": "jsonc",
"*.hpp.in": "cpp"
},
"files.insertFinalNewline": true,
"files.trimFinalNewlines": true,
"files.trimTrailingWhitespace": true,
"workbench.startupEditor": "none",
"workbench.editorAssociations": {
"*.ipynb": "jupyter-notebook",
"*.md": "vscode.markdown.preview.editor",
"*.svg": "svgPreviewer.customEditor"
},
"workbench.colorTheme": "Dracula Theme Soft",
"git.enableSmartCommit": true,
"git.autofetch": true,
"git.confirmSync": false,
"git.openRepositoryInParentFolders": "always",
"partialDiff.enableTelemetry": false,
"prettier.tabWidth": 4,
"prettier.singleQuote": true,
"prettier.jsxSingleQuote": true,
"prettier.trailingComma": "all",
"prettier.useEditorConfig": true,
"prettier.bracketSpacing": false,
"markdown.validate.enabled": true,
"[markdown]": {
"files.trimTrailingWhitespace": false,
"editor.formatOnSave": false,
"editor.defaultFormatter": "yzhang.markdown-all-in-one",
"editor.wordWrap": "wordWrapColumn",
"editor.wordWrapColumn": 80
},
"[yaml]": {
"editor.formatOnSave": false,
"editor.defaultFormatter": "redhat.vscode-yaml",
"editor.wordWrap": "wordWrapColumn",
"editor.wordWrapColumn": 80
},
"[json]": {
"editor.formatOnSave": false,
"editor.defaultFormatter": "vscode.json-language-features"
},
"[jsonc]": {
"editor.formatOnSave": false
},
"[plaintext]": {
"editor.wordWrap": "wordWrapColumn",
"editor.wordWrapColumn": 120
},
"[toml]": {
"editor.wordWrap": "wordWrapColumn",
"editor.wordWrapColumn": 80,
"editor.defaultFormatter": "tamasfe.even-better-toml",
"editor.formatOnSave": true
},
"better-comments.tags": [
{
"tag": "XXX",
"color": "#F8C471"
},
{
"tag": "WARN",
"color": "#FF6961"
},
{
"tag": "NOTE",
"color": "#3498DB"
},
{
"tag": "TODO",
"color": "#77C3EC"
}
],
"vsintellicode.modify.editor.suggestSelection": "automaticallyOverrodeDefaultValue",
"codesnap.showWindowControls": false,
"codesnap.shutterAction": "copy",
"Workspace_Formatter.excludePattern": [
"**/build",
"**/.*",
"**/.vscode",
"**/html"
],
"svg.preview.autoOpen": true,
"remote.WSL.fileWatcher.polling": true,
"errorLens.delay": 1000,
"errorLens.enabledDiagnosticLevels": [
"error",
"warning"
],
"errorLens.enabled": false,
"[python]": {
"editor.formatOnSave": false,
"editor.defaultFormatter": "charliermarsh.ruff",
"editor.formatOnType": false
},
"python.languageServer": "Jedi",
"python.analysis.addImport.exactMatchOnly": true,
"python.analysis.autoImportCompletions": false,
"python.analysis.completeFunctionParens": false,
"python.analysis.autoFormatStrings": true,
"python.analysis.logLevel": "Error",
"python.createEnvironment.contentButton": "show",
"python.missingPackage.severity": "Error",
"mypy-type-checker.importStrategy": "fromEnvironment",
"black-formatter.importStrategy": "fromEnvironment",
"isort.check": true,
"isort.importStrategy": "fromEnvironment",
"ruff.organizeImports": false,
"ruff.fixAll": false,
"autoDocstring.generateDocstringOnEnter": true,
"autoDocstring.quoteStyle": "'''",
"jupyter.interactiveWindow.creationMode": "perFile",
"jupyter.askForKernelRestart": false,
"jupyter.themeMatplotlibPlots": true,
"jupyter.logging.level": "error",
"notebook.formatOnSave.enabled": false,
"notebook.output.textLineLimit": 20,
"notebook.compactView": false,
"notebook.diff.ignoreMetadata": true,
"notebook.diff.ignoreOutputs": true
}

19
.vscode/tasks.json vendored

@ -0,0 +1,19 @@
{
"version": "2.0.0",
"tasks": [
{
"label": "Python: Current File",
"type": "shell",
"command": "${command:python.interpreterPath} ${file}",
"args": [],
"group": "build"
},
{
"label": "Python: Main File",
"type": "shell",
"command": "${command:python.interpreterPath} ${workspaceFolder}/main.py",
"args": [],
"group": "build"
}
]
}

@ -1,32 +1,153 @@
[tool.poetry]
name = "optimumohservice"
version = "0.1.0"
description = ""
authors = ["Cizz22 <cisatraa@gmail.com>"]
license = "MIT"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.11"
fastapi = {extras = ["standard"], version = "^0.115.4"}
sqlalchemy = "^2.0.36"
httpx = "^0.27.2"
pytest = "^8.3.3"
faker = "^30.8.2"
factory-boy = "^3.3.1"
sqlalchemy-utils = "^0.41.2"
slowapi = "^0.1.9"
uvicorn = "^0.32.0"
pytz = "^2024.2"
sqlalchemy-filters = "^0.13.0"
asyncpg = "^0.30.0"
requests = "^2.32.3"
pydantic = "^2.10.2"
temporalio = "^1.8.0"
pandas = "^2.2.3"
psycopg2-binary = "^2.9.10"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.black]
target-version = ['py310']
line-length = 120
skip-string-normalization = true
skip-magic-trailing-comma = true
force-exclude = '''
/(
| docs
| setup.py
)/
'''
[tool.isort]
py_version = 310
sections = ["FUTURE", "STDLIB", "THIRDPARTY", "FIRSTPARTY", "LOCALFOLDER"]
default_section = "THIRDPARTY"
known_third_party = []
known_first_party = []
known_local_folder = []
# style: black
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
ensure_newline_before_comments = true
line_length = 120
split_on_trailing_comma = true
lines_after_imports = 2
force_single_line = true
skip_glob = ["docs/*", "setup.py"]
filter_files = true
[tool.ruff]
target-version = "py310"
line-length = 120
indent-width = 4
extend-exclude = ["docs", "test", "tests"]
[tool.ruff.lint]
select = ["F", "E"]
extend-select = ["W", "C90", "I", "N", "B", "A", "C4", "PERF", "RUF"]
ignore = []
fixable = ["ALL"]
unfixable = []
preview = true
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
[tool.ruff.lint.isort]
force-single-line = true
force-sort-within-sections = false
lines-after-imports = 2
known-first-party = []
known-local-folder = []
known-third-party = []
section-order = [
"future",
"standard-library",
"third-party",
"first-party",
"local-folder",
]
[tool.ruff.lint.mccabe]
max-complexity = 24
[tool.ruff.lint.pycodestyle]
ignore-overlong-task-comments = true
[tool.ruff.lint.pydocstyle]
convention = "numpy"
[tool.ruff.lint.flake8-annotations]
allow-star-arg-any = true
ignore-fully-untyped = true
[tool.ruff.lint.pylint]
max-args = 5
max-branches = 12
max-locals = 15
max-statements = 50
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
skip-magic-trailing-comma = false
line-ending = "auto"
preview = false
docstring-code-format = true
[tool.mypy]
# Platform configuration
python_version = "3.10"
# imports related
ignore_missing_imports = true
follow_imports = "silent"
# None and Optional handling
no_implicit_optional = false
strict_optional = false
# Configuring warnings
warn_unused_configs = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
warn_return_any = false
# Untyped definitions and calls
check_untyped_defs = false
disallow_untyped_calls = false
disallow_untyped_defs = false
disallow_incomplete_defs = false
disallow_untyped_decorators = false
# Disallow dynamic typing
disallow_subclassing_any = false
disallow_any_unimported = false
disallow_any_expr = false
disallow_any_decorated = false
disallow_any_explicit = false
disallow_any_generics = false
# Miscellaneous strictness flags
allow_untyped_globals = true
allow_redefinition = true
local_partial_types = false
implicit_reexport = true
strict_equality = true
# Configuring error messages
show_error_context = false
show_column_numbers = false
show_error_codes = true
exclude = ["docs", "test", "tests"]
[tool.pyright]
pythonVersion = "3.10"
typeCheckingMode = "basic"
# enable subset of "strict"
reportDuplicateImport = true
reportInvalidStubStatement = true
reportOverlappingOverload = true
reportPropertyTypeMismatch = true
reportUntypedClassDecorator = false
reportUntypedFunctionDecorator = false
reportUntypedNamedTuple = false
reportUnusedImport = true
# disable subset of "basic"
reportGeneralTypeIssues = false
reportMissingModuleSource = false
reportOptionalCall = false
reportOptionalIterable = false
reportOptionalMemberAccess = false
reportOptionalOperand = false
reportOptionalSubscript = false
reportPrivateImportUsage = false
reportUnboundVariable = false
exclude = ["docs", "setup.py"]

@ -0,0 +1,36 @@
# Packaging
# python -m pip install -U pip
build>=1.0.3
twine>=5.0.0
setuptools>=69.1.0
# Jupyter
ipython>=8.21.0
ipykernel>=6.29.2
# Linting/Formatting
ruff>=0.2.1
black>=24.2.0
isort>=5.13.2
# Tooling
pre-commit>=3.6.1
# Type Checker
mypy>=1.8.0
mypy-extensions>=1.0.0
pyright>=1.1.350
# Testing
pytest>=8.0.0
pytest-cov>=4.1.0
pytest-benchmark>=4.0.0
codecov>=2.1.13
tox>=4.12.1
# Documentation
mkdocs>=1.5.3
mkdocstrings>=0.24.0
mkdocs-material>=9.5.9
mkdocstrings-python>=1.8.0
Pygments>=2.17.2

@ -0,0 +1 @@
-r requirements-dev.txt

@ -1,18 +1,35 @@
from typing import Optional
from uuid import UUID
import numpy as np
from sqlalchemy import Select, func, select
from fastapi import HTTPException
from fastapi import status
from sqlalchemy import Select
from sqlalchemy import func
from sqlalchemy import select
from sqlalchemy.orm import joinedload
from src.workorder.model import MasterWorkOrder
from src.scope_equipment.model import ScopeEquipment
from src.database.core import DbSession
from src.overhaul_scope.service import get_all
from .schema import CalculationTimeConstrainsParametersRead, CalculationTimeConstrainsParametersRetrive, CalculationTimeConstrainsParametersCreate, CalculationTimeConstrainsRead
from .service import get_calculation_by_reference_and_parameter, get_calculation_result, get_overhaul_cost_by_time_chart, get_corrective_cost_time_chart, create_param_and_data, get_calculation_data_by_id, create_calculation_result_service, get_avg_cost_by_asset
from src.scope_equipment.model import ScopeEquipment
from src.scope_equipment.service import get_by_assetnum
from fastapi import HTTPException, status
from src.workorder.model import MasterWorkOrder
from .schema import CalculationTimeConstrainsParametersCreate
from .schema import CalculationTimeConstrainsParametersRead
from .schema import CalculationTimeConstrainsParametersRetrive
from .schema import CalculationTimeConstrainsRead
from .service import create_calculation_result_service
from .service import create_param_and_data
from .service import get_avg_cost_by_asset
from .service import get_calculation_by_reference_and_parameter
from .service import get_calculation_data_by_id
from .service import get_calculation_result
from .service import get_corrective_cost_time_chart
from .service import get_overhaul_cost_by_time_chart
async def get_create_calculation_parameters(*, db_session: DbSession, calculation_id: str):
async def get_create_calculation_parameters(*, db_session: DbSession, calculation_id: Optional[str] = None):
if calculation_id is not None:
calculation = await get_calculation_data_by_id(calculation_id=calculation_id, db_session=db_session)
@ -63,9 +80,9 @@ async def get_create_calculation_parameters(*, db_session: DbSession, calculatio
async def create_calculation(*, db_session: DbSession, calculation_time_constrains_in: CalculationTimeConstrainsParametersCreate, created_by: str):
calculation_data = await create_param_and_data(
db_session=db_session, calculation_param_in=calculation_time_constrains_in, created_by=created_by)
results = await create_calculation_result_service(db_session=db_session, calculation=calculation_data)
results = await create_calculation_result_service(db_session=db_session, calculation=calculation_data)
return results

@ -1,13 +1,28 @@
from typing import Any, Dict, Optional, Union
from fastapi import APIRouter, HTTPException, status
from .schema import CalculationTimeConstrainsCreate, CalculationTimeConstrainsParametersRead, CalculationTimeConstrainsRead, CalculationTimeConstrainsParametersCreate, CalculationTimeConstrainsParametersRetrive, CalculationResultsRead
from typing import List
from typing import Optional
from typing import Union
from fastapi import APIRouter
from fastapi.params import Query
from src.auth.service import CurrentUser
from src.database.core import DbSession
from src.models import StandardResponse
from src.auth.service import CurrentUser
from .service import create_param_and_data, get_calculation_result, get_calculation_result_by_day
from .flows import get_create_calculation_parameters, create_calculation, get_or_create_scope_equipment_calculation
from fastapi.params import Query
from .flows import create_calculation
from .flows import get_create_calculation_parameters
from .flows import get_or_create_scope_equipment_calculation
from .schema import CalculationResultsRead
from .schema import CalculationSelectedEquipmentUpdate
from .schema import CalculationTimeConstrainsCreate
from .schema import CalculationTimeConstrainsParametersCreate
from .schema import CalculationTimeConstrainsParametersRead
from .schema import CalculationTimeConstrainsParametersRetrive
from .schema import CalculationTimeConstrainsRead
from .service import get_calculation_result
from .service import get_calculation_result_by_day
from .service import bulk_update_equipment
router = APIRouter()
@ -28,7 +43,7 @@ async def create_calculation_time_constrains(db_session: DbSession, current_user
@router.get("/parameters", response_model=StandardResponse[Union[CalculationTimeConstrainsParametersRetrive, CalculationTimeConstrainsParametersRead]])
async def get_calculation_parameters(db_session: DbSession, calculation_id: Optional[str] = Query(None)):
async def get_calculation_parameters(db_session: DbSession, calculation_id: Optional[str] = Query(default=None)):
"""Get all calculation parameter."""
parameters = await get_create_calculation_parameters(db_session=db_session, calculation_id=calculation_id)
@ -54,3 +69,12 @@ async def get_simulation_result(db_session:DbSession, calculation_id, calculatio
simulation_result = await get_calculation_result_by_day(db_session=db_session, calculation_id=calculation_id, simulation_day=calculation_simuation_in.intervalDays)
return StandardResponse(data=simulation_result, message="Data retrieved successfully")
@router.put("/{calculation_id}", response_model=StandardResponse[List[str]])
async def update_selected_equipment(db_session: DbSession, calculation_id, calculation_time_constrains_in: List[CalculationSelectedEquipmentUpdate]):
results = await bulk_update_equipment(db=db_session, selected_equipments=calculation_time_constrains_in, calculation_data_id=calculation_id)
return StandardResponse(
data=results,
message="Data retrieved successfully",
)

@ -91,4 +91,6 @@ class CalculationTimeConstrainsSimulationRead(CalculationTimeConstrainsBase):
simulation: CalculationResultsRead
class CalculationSelectedEquipmentUpdate(CalculationTimeConstrainsBase):
is_included: bool
assetnum: str

@ -1,20 +1,31 @@
from typing import List, Optional, Tuple
from typing import List
from typing import Optional
from typing import Tuple
from uuid import UUID
import numpy as np
from sqlalchemy import and_, func, select
from sqlalchemy.engine import result
from fastapi import HTTPException
from fastapi import status
from sqlalchemy import and_
from sqlalchemy import case
from sqlalchemy import func
from sqlalchemy import select
from sqlalchemy import update
from sqlalchemy.orm import joinedload
from src.database.core import DbSession
from src.overhaul_activity.service import get_all_by_session_id
from src.scope_equipment.model import ScopeEquipment
from src.workorder.model import MasterWorkOrder
from .schema import CalculationTimeConstrainsParametersCreate, CalculationTimeConstrainsRead, EquipmentResult, OptimumResult
from .model import CalculationParam, OverhaulReferenceType, CalculationData, CalculationResult, CalculationEquipmentResult
from fastapi import HTTPException, status
from src.overhaul_scope.service import get_by_scope_name, get
from src.scope_equipment.service import get_by_assetnum
from src.overhaul_scope.service import get as get_scope
from src.workorder.model import MasterWorkOrder
from .model import CalculationData
from .model import CalculationEquipmentResult
from .model import CalculationResult
from .schema import CalculationResultsRead
from .schema import CalculationTimeConstrainsParametersCreate
from .schema import CalculationTimeConstrainsRead
from .schema import OptimumResult
from .schema import CalculationSelectedEquipmentUpdate
def get_overhaul_cost_by_time_chart(overhaul_cost: float, days: int,numEquipments:int ,decay_base: float = 1.01) -> np.ndarray:
@ -22,7 +33,7 @@ def get_overhaul_cost_by_time_chart(overhaul_cost: float, days: int,numEquipment
raise ValueError("Overhaul cost cannot be negative")
if days <= 0:
raise ValueError("Days must be positive")
exponents = np.arange(0, days)
cost_per_equipment = overhaul_cost / numEquipments
# Using a slower decay base to spread the budget depletion over more days
@ -36,14 +47,14 @@ def get_overhaul_cost_by_time_chart(overhaul_cost: float, days: int,numEquipment
# raise ValueError("Overhaul cost cannot be negative")
# if days <= 0:
# raise ValueError("Days must be positive")
# exponents = np.arange(0, days)
# cost_per_equipment = overhaul_cost / numEquipments
# # Introduce randomness by multiplying with a random factor
# random_factors = np.random.normal(1.0, 0.1, numEquipments) # Mean 1.0, Std Dev 0.1
# results = np.array([cost_per_equipment * factor / (decay_base ** exponents) for factor in random_factors])
# results = np.where(np.isfinite(results), results, 0)
# return results
@ -57,7 +68,7 @@ def get_corrective_cost_time_chart(material_cost: float, service_cost: float, da
# Calculate daily failure rate using sigmoid function
daily_failure_rate = base_rate / (1 + np.exp(-acceleration * (day_points - grace_period)/days))
# Calculate cumulative failures
failure_counts = np.cumsum(daily_failure_rate)
@ -121,14 +132,14 @@ async def get_calculation_result(db_session: DbSession, calculation_id: str):
status_code=status.HTTP_404_NOT_FOUND,
detail="A data with this id does not exist.",
)
scope_overhaul = await get_scope(db_session=db_session, overhaul_session_id=scope_calculation.overhaul_session_id)
if not scope_overhaul:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="A data with this id does not exist.",
)
calculation_results = []
for i in range(days):
@ -138,15 +149,15 @@ async def get_calculation_result(db_session: DbSession, calculation_id: str):
"num_failures": 0,
"day": i + 1
}
for eq in scope_calculation.equipment_results:
if not eq.is_included:
continue
result["corrective_cost"] += float(eq.corrective_costs[i])
result["overhaul_cost"] += float(eq.overhaul_costs[i])
result["num_failures"] += int(eq.daily_failures[i])
calculation_results.append(CalculationResultsRead(**result))
@ -167,8 +178,8 @@ async def get_calculation_data_by_id(db_session: DbSession, calculation_id) -> C
).options(
joinedload(CalculationData.equipment_results), joinedload(CalculationData.parameter)
)
result = await db_session.execute(stmt)
return result.unique().scalar()
@ -177,7 +188,7 @@ async def get_calculation_data_by_id(db_session: DbSession, calculation_id) -> C
# days = 360
# calculation = await get_calculation_data_by_id(db_session=db_session, calculation_id=calculation_id)
# # reference = await get_by_assetnum(db_session=db_session, assetnum=calculation.reference_id) if calculation.overhaul_reference_type == OverhaulReferenceType.ASSET else await get(db_session=db_session, scope_id=calculation.reference_id)
# # Multiple Eequipment
# equipments_scope = get_all_by_session_id(db_session=db_session, overhaul_session_id=calculation.overhaul_session_id)
@ -237,15 +248,15 @@ async def create_calculation_result_service(
calculation: CalculationData,
) -> CalculationTimeConstrainsRead:
days = 365 # Changed to 365 days as per requirement
# Get all equipment for this calculation session
equipments = await get_all_by_session_id(db_session=db_session, overhaul_session_id=calculation.overhaul_session_id)
scope = await get_scope(db_session=db_session, overhaul_session_id=calculation.overhaul_session_id)
calculation_data = await get_calculation_data_by_id(db_session=db_session, calculation_id=calculation.id)
# Store results for each equipment
equipment_results: List[CalculationEquipmentResult] = []
@ -259,18 +270,18 @@ async def create_calculation_result_service(
service_cost=eq.service_cost,
days=days
)
overhaul_cost_points = get_overhaul_cost_by_time_chart(
calculation_data.parameter.overhaul_cost,
calculation_data.parameter.overhaul_cost,
days=days,
numEquipments=len(equipments)
)
# Calculate individual equipment optimum points
equipment_total_cost = corrective_costs + overhaul_cost_points
equipment_optimum_index = np.argmin(equipment_total_cost)
equipment_failure_sum = sum(daily_failures[:equipment_optimum_index])
equipment_results.append(CalculationEquipmentResult(
corrective_costs=corrective_costs.tolist(),
overhaul_costs=overhaul_cost_points.tolist(),
@ -281,13 +292,13 @@ async def create_calculation_result_service(
optimum_day=int(equipment_optimum_index + 1),
calculation_data_id=calculation.id
))
# Add to totals
total_corrective_costs += corrective_costs
total_daily_failures += daily_failures
db_session.add_all(equipment_results)
# Calculate optimum points using total costs
total_cost = total_corrective_costs + overhaul_cost_points
@ -316,7 +327,7 @@ async def create_calculation_result_service(
# Update calculation with optimum day
calculation.optimum_oh_day = optimum.days
await db_session.commit()
# Return results including individual equipment data
@ -360,3 +371,38 @@ async def get_avg_cost_by_asset(*, db_session: DbSession, assetnum: str):
result = await db_session.execute(stmt)
return result.scalar_one_or_none()
async def bulk_update_equipment(*, db: DbSession, selected_equipments: List[CalculationSelectedEquipmentUpdate], calculation_data_id:UUID):
# Assuming your model is named Asset
# and updates is a list of dictionaries containing assetnum and updated values
# Create a case statement for each field you want to update
stmt = update(CalculationEquipmentResult)
# Using the case method to match assetnums with their updates
case_mappings = {
asset.assetnum: asset.is_included
for asset in selected_equipments
}
# Get all assetnums that need to be updated
assetnums = list(case_mappings.keys())
# Build the update statement
stmt = (
update(CalculationEquipmentResult)
.where(CalculationEquipmentResult.calculation_data_id == calculation_data_id)
.where(CalculationEquipmentResult.assetnum.in_(assetnums))
.values({
"is_included": case(
(CalculationEquipmentResult.assetnum == assetnum, is_included)
for assetnum, is_included in case_mappings.items()
)
})
)
await db.execute(stmt)
await db.commit()
return assetnums

Loading…
Cancel
Save