feat: Multiple ETL refinements and fixes

.gitignore

@@ -87,5 +87,5 @@ scripts/check_pretreatment_duplicates.py
 # hatch-vcs generated version files
 _version.py
 
-# analysis environment
-analysis
+# analysis environment (only ignore the BioCirv AI submodule workspace)
+analysis/biocirv-ai/

alembic/versions/bd227e99e006_add_fermentation_method_fields_resource_.py

@@ -0,0 +1,79 @@
+"""Add fermentation method fields, resource_image, and county_ag_report_record tables
+
+Revision ID: bd227e99e006
+Revises: 9e8f7a6b5c52
+Create Date: 2026-04-09 14:09:11.091043
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+# revision identifiers, used by Alembic.
+revision: str = 'bd227e99e006'
+down_revision: Union[str, Sequence[str], None] = '9e8f7a6b5c52'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('resource_image',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=True),
+    sa.Column('updated_at', sa.DateTime(), nullable=True),
+    sa.Column('etl_run_id', sa.Integer(), nullable=True),
+    sa.Column('lineage_group_id', sa.Integer(), nullable=True),
+    sa.Column('resource_id', sa.Integer(), nullable=False),
+    sa.Column('resource_name', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('image_url', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('sort_order', sa.Integer(), nullable=True),
+    sa.ForeignKeyConstraint(['etl_run_id'], ['etl_run.id'], ),
+    sa.ForeignKeyConstraint(['resource_id'], ['resource.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('resource_id', 'image_url', name='resource_image_resource_id_image_url_key')
+    )
+    op.create_table('county_ag_report_record',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=True),
+    sa.Column('updated_at', sa.DateTime(), nullable=True),
+    sa.Column('etl_run_id', sa.Integer(), nullable=True),
+    sa.Column('lineage_group_id', sa.Integer(), nullable=True),
+    sa.Column('record_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('geoid', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('primary_ag_product_id', sa.Integer(), nullable=True),
+    sa.Column('description', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('resource_type', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('data_year', sa.Integer(), nullable=True),
+    sa.Column('data_source_id', sa.Integer(), nullable=True),
+    sa.Column('produced_nsjv', sa.Boolean(), nullable=True),
+    sa.Column('processed_nsjv', sa.Boolean(), nullable=True),
+    sa.Column('note', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('prodn_value_note', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.ForeignKeyConstraint(['data_source_id'], ['data_source.id'], ),
+    sa.ForeignKeyConstraint(['etl_run_id'], ['etl_run.id'], ),
+    sa.ForeignKeyConstraint(['geoid'], ['place.geoid'], ),
+    sa.ForeignKeyConstraint(['primary_ag_product_id'], ['primary_ag_product.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('record_id')
+    )
+    op.create_foreign_key('fermentation_record_pretreatment_method_id_fkey', 'fermentation_record', 'method', ['pretreatment_method_id'], ['id'])
+    op.create_foreign_key('fermentation_record_eh_method_id_fkey', 'fermentation_record', 'method', ['eh_method_id'], ['id'])
+    op.create_foreign_key('fermentation_record_strain_id_fkey', 'fermentation_record', 'strain', ['strain_id'], ['id'])
+    op.create_unique_constraint('strain_name_key', 'strain', ['name'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint('strain_name_key', 'strain', type_='unique')
+    op.drop_constraint('fermentation_record_strain_id_fkey', 'fermentation_record', type_='foreignkey')
+    op.drop_constraint('fermentation_record_pretreatment_method_id_fkey', 'fermentation_record', type_='foreignkey')
+    op.drop_constraint('fermentation_record_eh_method_id_fkey', 'fermentation_record', type_='foreignkey')
+    op.drop_table('county_ag_report_record')
+    op.drop_table('resource_image')
+    # ### end Alembic commands ###

resources/docker/docker-compose.yml

@@ -82,6 +82,8 @@ services:
       - ../../alembic.ini:/app/alembic.ini
       - ../../src/ca_biositing/pipeline/ca_biositing:/app/.pixi/envs/etl/lib/python3.12/site-packages/ca_biositing
       - ../../src/ca_biositing/datamodels/ca_biositing/datamodels:/app/.pixi/envs/etl/lib/python3.12/site-packages/ca_biositing/datamodels
+      - ../../src/ca_biositing/pipeline/ca_biositing:/app/.pixi/envs/etl/lib/python3.13/site-packages/ca_biositing
+      - ../../src/ca_biositing/datamodels/ca_biositing/datamodels:/app/.pixi/envs/etl/lib/python3.13/site-packages/ca_biositing/datamodels
     depends_on:
       prefect-server:
         condition: service_healthy

resources/prefect/run_prefect_flow.py

@@ -12,10 +12,11 @@
     "samples": "ca_biositing.pipeline.flows.samples_etl.samples_etl_flow",
     "analysis_records": "ca_biositing.pipeline.flows.analysis_records.analysis_records_flow",
     "aim2_bioconversion": "ca_biositing.pipeline.flows.aim2_bioconversion.aim2_bioconversion_flow",
+    "county_ag_report": "ca_biositing.pipeline.flows.county_ag_report_etl.county_ag_report_flow",
     "usda_etl": "ca_biositing.pipeline.flows.usda_etl.usda_etl_flow",
     "landiq": "ca_biositing.pipeline.flows.landiq_etl.landiq_etl_flow",
     "billion_ton": "ca_biositing.pipeline.flows.billion_ton_etl.billion_ton_etl_flow",
-    #"field_sample": "ca_biositing.pipeline.flows.field_sample_etl.field_sample_etl_flow",
+    "field_sample": "ca_biositing.pipeline.flows.field_sample_etl.field_sample_etl_flow",
     #"prepared_sample": "ca_biositing.pipeline.flows.prepared_sample_etl.prepared_sample_etl_flow",
     "thermochem": "ca_biositing.pipeline.flows.thermochem_etl.thermochem_etl_flow",
 }

src/ca_biositing/datamodels/ca_biositing/datamodels/models/__init__.py

@@ -20,7 +20,7 @@
 from .experiment_equipment import DeconVessel, Equipment, Experiment, ExperimentAnalysis, ExperimentEquipment, ExperimentMethod, ExperimentPreparedSample
 
 # External Data
-from .external_data import BillionTon2023Record, LandiqRecord, LandiqResourceMapping, Polygon, ResourceUsdaCommodityMap, UsdaCensusRecord, UsdaCommodity, UsdaDomain, UsdaMarketRecord, UsdaMarketReport, UsdaStatisticCategory, UsdaSurveyProgram, UsdaSurveyRecord, UsdaTermMap
+from .external_data import BillionTon2023Record, CountyAgReportRecord, LandiqRecord, LandiqResourceMapping, Polygon, ResourceUsdaCommodityMap, UsdaCensusRecord, UsdaCommodity, UsdaDomain, UsdaMarketRecord, UsdaMarketReport, UsdaStatisticCategory, UsdaSurveyProgram, UsdaSurveyRecord, UsdaTermMap
 
 # Field Sampling
 from .field_sampling import AgTreatment, CollectionMethod, FieldSample, FieldSampleCondition, FieldStorageMethod, HarvestMethod, LocationSoilType, PhysicalCharacteristic, ProcessingMethod, SoilType
@@ -41,7 +41,7 @@
 from .places import LocationAddress, Place
 
 # Resource Information
-from .resource_information import PrimaryAgProduct, Resource, ResourceAvailability, ResourceClass, ResourceCounterfactual, ResourceMorphology, ResourceSubclass, ResourcePriceRecord, ResourceTransportRecord, ResourceStorageRecord, ResourceEndUseRecord, ResourceProductionRecord
+from .resource_information import PrimaryAgProduct, Resource, ResourceAvailability, ResourceClass, ResourceCounterfactual, ResourceImage, ResourceMorphology, ResourceSubclass, ResourcePriceRecord, ResourceTransportRecord, ResourceStorageRecord, ResourceEndUseRecord, ResourceProductionRecord
 
 # Sample Preparation
 from .sample_preparation import PreparationMethod, PreparationMethodAbbreviation, PreparedSample

src/ca_biositing/datamodels/ca_biositing/datamodels/models/aim2_records/fermentation_record.py

@@ -8,9 +8,9 @@
 class FermentationRecord(Aim2RecordBase, table=True):
     __tablename__ = "fermentation_record"
 
-    strain_id: Optional[int] = Field(default=None)
-    pretreatment_method_id: Optional[int] = Field(default=None)
-    eh_method_id: Optional[int] = Field(default=None)
+    strain_id: Optional[int] = Field(default=None, foreign_key="strain.id")
+    pretreatment_method_id: Optional[int] = Field(default=None, foreign_key="method.id")
+    eh_method_id: Optional[int] = Field(default=None, foreign_key="method.id")
     well_position: Optional[str] = Field(default=None)
     vessel_id: Optional[int] = Field(default=None, foreign_key="decon_vessel.id")
     analyte_detection_equipment_id: Optional[int] = Field(default=None)

src/ca_biositing/datamodels/ca_biositing/datamodels/models/aim2_records/strain.py

@@ -1,9 +1,10 @@
 from ..base import LookupBase
-from sqlmodel import Field, SQLModel
+from sqlmodel import Field
 from typing import Optional
 
 
 class Strain(LookupBase, table=True):
     __tablename__ = "strain"
 
+    name: Optional[str] = Field(default=None, unique=True)
     parent_strain_id: Optional[int] = Field(default=None)

src/ca_biositing/datamodels/ca_biositing/datamodels/models/external_data/__init__.py

@@ -1,4 +1,5 @@
 from .billion_ton import BillionTon2023Record
+from .county_ag_report_record import CountyAgReportRecord
 from .landiq_record import LandiqRecord
 from .landiq_resource_mapping import LandiqResourceMapping
 from .polygon import Polygon

src/ca_biositing/datamodels/ca_biositing/datamodels/models/external_data/county_ag_report_record.py

@@ -0,0 +1,24 @@
+from ..base import BaseEntity
+from sqlmodel import Field, Relationship
+from typing import Optional
+
+
+class CountyAgReportRecord(BaseEntity, table=True):
+    __tablename__ = "county_ag_report_record"
+
+    record_id: str = Field(nullable=False, unique=True)
+    geoid: Optional[str] = Field(default=None, foreign_key="place.geoid")
+    primary_ag_product_id: Optional[int] = Field(default=None, foreign_key="primary_ag_product.id")
+    description: Optional[str] = Field(default=None)
+    resource_type: Optional[str] = Field(default=None)
+    data_year: Optional[int] = Field(default=None)
+    data_source_id: Optional[int] = Field(default=None, foreign_key="data_source.id")
+    produced_nsjv: Optional[bool] = Field(default=None)
+    processed_nsjv: Optional[bool] = Field(default=None)
+    note: Optional[str] = Field(default=None)
+    prodn_value_note: Optional[str] = Field(default=None)
+
+    # Relationships
+    place: Optional["Place"] = Relationship()
+    primary_ag_product: Optional["PrimaryAgProduct"] = Relationship()
+    data_source: Optional["DataSource"] = Relationship()

src/ca_biositing/datamodels/ca_biositing/datamodels/models/resource_information/__init__.py

@@ -5,6 +5,7 @@
 from .resource_counterfactual import ResourceCounterfactual
 from .resource import ResourceMorphology
 from .resource import ResourceSubclass
+from .resource_image import ResourceImage
 from .resource_price_record import ResourcePriceRecord
 from .resource_transport_record import ResourceTransportRecord
 from .resource_storage_record import ResourceStorageRecord

src/ca_biositing/datamodels/ca_biositing/datamodels/models/resource_information/resource_image.py

@@ -0,0 +1,19 @@
+from ..base import BaseEntity
+from sqlmodel import Field, Relationship
+from typing import Optional
+from sqlalchemy import UniqueConstraint
+
+
+class ResourceImage(BaseEntity, table=True):
+    __tablename__ = "resource_image"
+    __table_args__ = (
+        UniqueConstraint('resource_id', 'image_url', name='resource_image_resource_id_image_url_key'),
+    )
+
+    resource_id: int = Field(foreign_key="resource.id")
+    resource_name: Optional[str] = Field(default=None)
+    image_url: Optional[str] = Field(default=None)
+    sort_order: Optional[int] = Field(default=None)
+
+    # Relationships
+    resource: Optional["Resource"] = Relationship()

src/ca_biositing/pipeline/ca_biositing/pipeline/etl/extract/county_ag_report.py

@@ -0,0 +1,11 @@
+"""
+ETL Extract: County Ag Reports
+"""
+
+from .factory import create_extractor
+
+GSHEET_NAME = "Aim 1-Feedstock Collection and Processing Data-BioCirV"
+
+primary_products = create_extractor(GSHEET_NAME, "07.7-Primary_products")
+pp_production_value = create_extractor(GSHEET_NAME, "07.7a-PP_Prodn_Value")
+pp_data_sources = create_extractor(GSHEET_NAME, "07.7b-PP_Data_sources")

src/ca_biositing/pipeline/ca_biositing/pipeline/etl/extract/producers.py

@@ -0,0 +1,28 @@
+"""
+Factory extractor for 04_Producers worksheet from SampleMetadata_v03-BioCirV.
+
+This worksheet contains producer/origin information and extended sample metadata:
+- Sample_name: Unique sample identifier (join key)
+- Resource, ProviderCode, FV_Date_Time: Redundant copies from 01_Sample_IDs
+- Producer: Producer name (identifies the source organization)
+- Prod_Location: Producer location name (maps to field_sample_storage_location_id)
+- Prod_Street, Prod_City, Prod_Zip: Producer address components
+- Prod_Date: Production date
+- Harvest_Method: Method used for harvesting
+- Treatment: Treatment applied to the sample
+- Soil_Type: Type of soil at production location
+- Crop_Variety, Crop_Cultivar: Variety and cultivar information
+- Production_Notes: Notes about the production process
+- Other metadata: Additional extended fields for sample context
+
+This extractor provides producer/origin context and addresses for
+field_sample_storage_location_id creation via LocationAddress.
+"""
+
+from .factory import create_extractor
+
+GSHEET_NAME = "SampleMetadata_v03-BioCirV"
+WORKSHEET_NAME = "04_Producers"
+
+# Create the extract task using the factory pattern
+extract = create_extractor(GSHEET_NAME, WORKSHEET_NAME, task_name="extract_producers")

src/ca_biositing/pipeline/ca_biositing/pipeline/etl/extract/qty_field_storage.py

@@ -0,0 +1,28 @@
+"""
+Factory extractor for 03_Qty_FieldStorage worksheet from SampleMetadata_v03-BioCirV.
+
+This worksheet contains sample quantity and field storage information:
+- Sample_name: Unique sample identifier (join key)
+- Resource, ProviderCode, FV_Date_Time: Redundant copies from 01_Sample_IDs
+- Sample_Container: Container type and size (e.g., "Bucket (5 gal.)", "Core", "Bale")
+  * Used for amount_collected_unit_id extraction (unit is embedded in this field)
+- Qty: Amount collected (maps to amount_collected)
+- Qty_Unit: Explicit unit column (if present; otherwise extract from Sample_Container)
+- Primary_Collector: Collector identifier (maps to collector_id via Contact lookup)
+- Collection_Team: Team members involved in collection
+- Destination_Lab: Lab where sample was sent
+- FieldStorage_Location: Storage location name (maps to field_storage_location_id)
+- FieldStorage_Conditions: Storage conditions (temperature, humidity, etc.)
+- FieldStorage_Duration: Duration stored in field
+- Other metadata: Comments, dates, etc.
+
+This extractor provides quantity, unit, and field storage context for collected samples.
+"""
+
+from .factory import create_extractor
+
+GSHEET_NAME = "SampleMetadata_v03-BioCirV"
+WORKSHEET_NAME = "03_Qty_FieldStorage"
+
+# Create the extract task using the factory pattern
+extract = create_extractor(GSHEET_NAME, WORKSHEET_NAME, task_name="extract_qty_field_storage")

src/ca_biositing/pipeline/ca_biositing/pipeline/etl/extract/resource_images.py

@@ -0,0 +1,10 @@
+"""
+ETL Extract: Resource Images
+"""
+
+from .factory import create_extractor
+
+GSHEET_NAME = "Aim 1-Feedstock Collection and Processing Data-BioCirV"
+WORKSHEET_NAME = "08.0_Resource_images"
+
+extract = create_extractor(GSHEET_NAME, WORKSHEET_NAME)