Add firmware update logic

backend/alembic/versions/20250323_506cc8d086c9-add_signature_fields_to_firmware_update_.py

@@ -1,32 +0,0 @@
-"""Add signature fields to firmware_update table
-
-Revision ID: 506cc8d086c9
-Revises: 00edfb13e611
-Create Date: 2025-03-23 14:49:42.662564+00:00
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '506cc8d086c9'
-down_revision: Union[str, None] = '00edfb13e611'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.add_column('firmware_updates', sa.Column('signing_certificate', sa.Text(), nullable=True))
-    op.add_column('firmware_updates', sa.Column('signature', sa.String(), nullable=True))
-    # ### end Alembic commands ###
-
-
-def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_column('firmware_updates', 'signature')
-    op.drop_column('firmware_updates', 'signing_certificate')
-    # ### end Alembic commands ###

backend/app/models/firmware_update.py

@@ -1,5 +1,5 @@
 import uuid
-from sqlalchemy import Column, DateTime, Enum, ForeignKey, Integer, String, Text, Uuid
+from sqlalchemy import Column, DateTime, Enum, ForeignKey, Integer, String, Uuid
 
 from app.database import Base
 from app.schemas.firmware_update import FirmwareUpdateStatus
@@ -16,7 +16,5 @@ class FirmwareUpdate(Base):
     location = Column(String)
     retrieve_date_time = Column(DateTime)
     install_date_time = Column(DateTime, nullable=True)
-    signing_certificate = Column(Text, nullable=True)
-    signature = Column(String, nullable=True)
 
     chargepoint_id = Column(Uuid, ForeignKey("chargepoints.id"), index=True)

backend/app/routers/chargepoint_v1.py

@@ -356,7 +356,7 @@ async def submit_firmware_update(
     if chargepoint_manager.is_connected(chargepoint_id) == False:
         raise HTTPException(status_code=503, detail="Chargepoint not connected.")
     try:
-        _, status = await firmware_service.submit_firmware_update(firmware_update_id)
-        return FirmwareUpdateSubmissionResponse(status=status)
+        firmware_update, status = await firmware_service.submit_firmware_update(firmware_update_id)
+        return FirmwareUpdateSubmissionResponse(firmware_update, status)
     except TimeoutError:
         raise HTTPException(status_code=503, detail="Chargepoint didn't respond in time.")

backend/app/schemas/firmware_update.py

@@ -28,8 +28,6 @@ class FirmwareUpdateBase(BaseModel):
     location: str
     retrieve_date_time: datetime
     install_date_time: Optional[datetime]
-    signing_certificate: Optional[str]
-    signature: Optional[str]
 
 class FirmwareUpdate(FirmwareUpdateBase):
     id: UUID
@@ -40,4 +38,5 @@ class FirmwareUpdateCreate(FirmwareUpdateBase):
     pass
 
 class FirmwareUpdateSubmissionResponse(BaseModel):
+    firmware_update: FirmwareUpdate
     status: str

backend/app/services/firmware_service.py

@@ -23,9 +23,7 @@ async def create_firmware_update(chargepoint_id: UUID, firmware_update: Firmware
             location=firmware_update.location,
             retrieve_date_time=firmware_update.retrieve_date_time,
             install_date_time=firmware_update.install_date_time,
-            chargepoint_id=db_chargepoint.id,
-            signing_certificate=firmware_update.signing_certificate,
-            signature=firmware_update.signature
+            chargepoint_id=db_chargepoint.id
         )
         db.add(db_firmware_update)
         db.commit()
@@ -44,10 +42,8 @@ async def submit_firmware_update(firmware_update_id: UUID) -> tuple[FirmwareUpda
                     retry_interval=db_firmware_update.retry_interval,
                     firmware=FirmwareType(
                         location=db_firmware_update.location,
-                        retrieve_date_time=db_firmware_update.retrieve_date_time.isoformat(),
-                        install_date_time=db_firmware_update.install_date_time.isoformat(),
-                        signing_certificate=db_firmware_update.signing_certificate,
-                        signature=db_firmware_update.signature
+                        retrieve_date_time=db_firmware_update.retrieve_date_time,
+                        install_date_time=db_firmware_update.install_date_time
                     )
                 ))
             if result.status == "Accepted" or result.status == "AcceptedCanceled":
@@ -62,5 +58,5 @@ async def update_firmware_status(chargepoint_identity: str, request_id: int, sta
     with SessionLocal() as db:
         db_chargepoint = db.query(ChargePoint).filter(ChargePoint.identity == chargepoint_identity).first()
         db_firmware_update = db.query(FirmwareUpdate).filter(FirmwareUpdate.chargepoint_id == db_chargepoint.id).filter(FirmwareUpdate.request_id == request_id).first()
-        db_firmware_update.status = FirmwareUpdateStatus(status)
+        db_firmware_update.status = status
         db.commit()