Company

Companies

Bases: Supergroup

Source code in june/groups/company.py

class Companies(Supergroup):
    """ """
    venue_class = Company

    def __init__(self, companies: List["Companies"]):
        """
        Create companies and provide functionality to allocate workers.

        Parameters
        ----------
        company_size_per_superarea_df: pd.DataFram
            Nr. of companies within a size-range per SuperArea.

        compsec_per_msoa_df: pd.DataFrame
            Nr. of companies per sector sector per SuperArea.
        """
        super().__init__(members=companies)

    @classmethod
    def for_geography(
        cls,
        geography: Geography,
        size_nr_file: str = None,
        sector_nr_per_msoa_file: str = None,
        sct_companies_file: str = None,
        ew_companies_file: str = None,
        ni_companies_file: str = None
    ) -> "Companies":
        """Creates companies for the specified geography, and saves them
        to the super_areas they belong to.

        Args:
            geography (Geography): 
            size_nr_file (str, optional): (Default value = None)
            sector_nr_per_msoa_file (str, optional): (Default value = None)
            sct_companies_file (str, optional): (Default value = None)
            ew_companies_file (str, optional): (Default value = None)
            ni_companies_file (str, optional): (Default value = None)

        """
        if not geography.super_areas:
            raise CompanyError("Empty geography!")
        # After creating the companies
        companies = cls.for_super_areas(
            geography.super_areas,
            size_nr_file,
            sector_nr_per_msoa_file,
            sct_companies_file,
            ew_companies_file,
            ni_companies_file
        )
        logger.info(f"There are {len(companies)} companies in this geography.")


        # Sample 5 companies from each super area for visualization
        sampled_companies = []
        for super_area in geography.super_areas:
            if hasattr(super_area, 'companies') and super_area.companies:
                # Sample 5 companies or fewer if there are less than 5
                sample_companies = random.sample(super_area.companies, min(5, len(super_area.companies)))
                for company in sample_companies:
                    sampled_companies.append({
                        "| Company ID": company.id,
                        "| Super Area": super_area.name,
                        "| Company Sector": company.sector,
                        "| Number of Workers": company.n_workers,
                        "| Coordinates": company.coordinates,
                        "| Max Workers": company.n_workers_max
                    })

        # Convert the sample data to a DataFrame
        df_companies = pd.DataFrame(sampled_companies)
        print("\n===== Sample of Created Companies =====")
        print(df_companies)

        return companies

    @classmethod
    def for_super_areas(
        cls,
        super_areas: List[SuperArea],
        size_nr_per_super_area_file: str = None,
        sector_nr_per_super_area_file: str = None,
        sct_companies_file: str = None,
        ew_companies_file: str = None,
        ni_companies_file: str = None
        ) -> "Companies":
        """Creates companies for the specified super_areas, and saves them
        to the super_areas they belong to.

        Args:
            super_areas (List[SuperArea]): 
            size_nr_per_super_area_file (str, optional): (Default value = None)
            sector_nr_per_super_area_file (str, optional): (Default value = None)
            sct_companies_file (str, optional): (Default value = None)
            ew_companies_file (str, optional): (Default value = None)
            ni_companies_file (str, optional): (Default value = None)

        """
        # Always try to use companies files first (auto-detect), fallback to old CSV method
        try:
            return cls.from_companies_file(super_areas, sct_companies_file, ew_companies_file, ni_companies_file)
        except (FileNotFoundError, KeyError, pd.errors.EmptyDataError):
            # Fallback to old CSV-based method if companies files don't exist or are empty
            logger.info("Companies files not found or empty, falling back to legacy CSV method")

        # Check if legacy CSV files are provided
        if size_nr_per_super_area_file is None or sector_nr_per_super_area_file is None:
            raise CompanyError("No company files available. Please provide either companies files or legacy CSV files.")

        size_per_superarea_df = pd.read_csv(size_nr_per_super_area_file, index_col=0)
        sector_per_superarea_df = pd.read_csv(
            sector_nr_per_super_area_file, index_col=0
        )
        super_area_names = [super_area.name for super_area in super_areas]
        company_sizes_per_super_area = size_per_superarea_df.loc[super_area_names]
        company_sectors_per_super_area = sector_per_superarea_df.loc[super_area_names]
        assert len(company_sectors_per_super_area) == len(company_sizes_per_super_area)
        if len(company_sectors_per_super_area) == 1:
            super_area = super_areas[0]
            companies = cls.create_companies_in_super_area(
                super_area, company_sizes_per_super_area, company_sectors_per_super_area
            )
            super_area.companies = companies
        else:
            companies = []
            for super_area, (_, company_sizes), (_, company_sectors) in zip(
                super_areas,
                company_sizes_per_super_area.iterrows(),
                company_sectors_per_super_area.iterrows(),
            ):
                super_area.companies = cls.create_companies_in_super_area(
                    super_area, company_sizes, company_sectors
                )
                companies += super_area.companies


        return cls(companies)

    @classmethod
    def from_companies_file(
        cls,
        super_areas: List[SuperArea],
        sct_companies_file: str = None,
        ew_companies_file: str = None,
        ni_companies_file: str = None
    ) -> "Companies":
        """Optimized company creation from CSV files with pre-filtering and vectorized operations.
        CSV format: company_id,super_area,industry_code,size_band,employee_count

        Supports mixed geographies by loading from SCT, EW, and NI files as needed:
        - Scotland super areas (S02*): Uses SCT file
        - England/Wales super areas (E02*, W02*): Uses EW file
        - Northern Ireland super areas (N*): Uses NI file

        Args:
            super_areas (List[SuperArea]): 
            sct_companies_file (str, optional): (Default value = None)
            ew_companies_file (str, optional): (Default value = None)
            ni_companies_file (str, optional): (Default value = None)

        """
        super_area_names = [super_area.name for super_area in super_areas]

        # Separate areas by region
        scotland_areas = [name for name in super_area_names if name.startswith('S02')]
        ew_areas = [name for name in super_area_names if name.startswith(('E02', 'W02'))]
        ni_areas = [name for name in super_area_names if name.startswith('N')]

        companies_df_list = []

        # Optimized CSV loading with pre-filtering
        def load_companies_optimized(file_path, target_areas, region_name):
            """

            Args:
                file_path: 
                target_areas: 
                region_name: 

            """
            if not target_areas:
                return None

            logger.info(f"Loading {region_name} companies from: {file_path}")
            try:
                # Read with chunking for memory efficiency on large files
                chunk_size = 50000
                filtered_chunks = []

                for chunk in pd.read_csv(file_path, chunksize=chunk_size):
                    # Pre-filter chunk to only include our target areas (major optimization!)
                    filtered_chunk = chunk[chunk['super_area'].isin(target_areas)]
                    if not filtered_chunk.empty:
                        filtered_chunks.append(filtered_chunk)

                if not filtered_chunks:
                    logger.warning(f"No companies found for {region_name} areas in {file_path}")
                    return None

                result_df = pd.concat(filtered_chunks, ignore_index=True)
                logger.info(f"Found {len(result_df)} {region_name} companies for {len(target_areas)} areas")
                return result_df

            except FileNotFoundError:
                logger.warning(f"{region_name} companies file not found: {file_path}")
                return None

        # Load each region with optimized filtering
        if scotland_areas:
            sct_file = sct_companies_file if sct_companies_file else default_sct_companies_file
            sct_df = load_companies_optimized(sct_file, scotland_areas, "SCT")
            if sct_df is not None:
                companies_df_list.append(sct_df)

        if ew_areas:
            ew_file = ew_companies_file if ew_companies_file else default_ew_companies_file
            ew_df = load_companies_optimized(ew_file, ew_areas, "EW")
            if ew_df is not None:
                companies_df_list.append(ew_df)

        if ni_areas:
            ni_file = ni_companies_file if ni_companies_file else default_ni_companies_file
            ni_df = load_companies_optimized(ni_file, ni_areas, "NI")
            if ni_df is not None:
                companies_df_list.append(ni_df)

        # Combine all company dataframes
        if not companies_df_list:
            raise FileNotFoundError("No company files found or loaded successfully")

        companies_df = pd.concat(companies_df_list, ignore_index=True)
        logger.info(f"Total companies loaded: {len(companies_df)} for {len(super_area_names)} super areas")

        # MAJOR OPTIMIZATION: Use groupby instead of nested loops
        companies = []

        # Group companies by super_area for vectorized processing  
        companies_by_area = companies_df.groupby('super_area')

        for super_area in super_areas:
            try:
                area_companies_df = companies_by_area.get_group(super_area.name)
            except KeyError:
                # No companies for this super area
                super_area.companies = []
                continue

            # Vectorized company creation using apply or list comprehension
            super_area_companies = []

            # Use itertuples for faster iteration (3x faster than iterrows)
            for row in area_companies_df.itertuples(index=False):
                company = cls.venue_class(
                    super_area=super_area,
                    n_workers_max=row.employee_count,
                    sector=str(row.industry_code),
                    registered_members_ids={}
                )
                super_area_companies.append(company)
                companies.append(company)

            super_area.companies = super_area_companies
            logger.debug(f"Created {len(super_area_companies)} companies for super area {super_area.name}")

        return cls(companies)

    @classmethod
    def from_sct_companies_file(
        cls,
        super_areas: List[SuperArea],
        sct_companies_file: str = None
    ) -> "Companies":
        """Legacy method for backward compatibility.
        Creates companies from SCT companies CSV file.

        Args:
            super_areas (List[SuperArea]): 
            sct_companies_file (str, optional): (Default value = None)

        """
        return cls.from_companies_file(super_areas, sct_companies_file=sct_companies_file)

    @classmethod
    def create_companies_in_super_area(
        cls, super_area: SuperArea, company_sizes, company_sectors
    ) -> list:
        """Creates companies in a super area using the sizes and sectors distributions.

        Args:
            super_area (SuperArea): 
            company_sizes: 
            company_sectors: 

        """
        sizes = np.array([])
        for size_bracket, counts in company_sizes.items():
            size_min, size_max = _get_size_brackets(size_bracket)
            sizes = np.concatenate(
                (sizes, np.random.randint(max(size_min, 1), size_max, int(counts)))
            )
        np.random.shuffle(sizes)
        sectors = []
        for sector, counts in company_sectors.items():
            sectors += [sector] * int(counts)
        shuffle(sectors)
        companies = list(
            map(
                lambda company_size, company_sector: cls.create_company(
                    super_area, company_size, company_sector
                ),
                sizes,
                sectors,
            )
        )
        return companies

    @classmethod
    def create_company(cls, super_area, company_size, company_sector):
        """Create a company instance.

        Args:
            super_area (SuperArea): The area the company belongs to.
            company_size (int): Maximum number of workers the company can accommodate.
            company_sector (str): The sector the company belongs to.

        Returns:
            Company: A new company instance.

        """
        company = cls.venue_class(
            super_area=super_area,
            n_workers_max=company_size,
            sector=company_sector,
            registered_members_ids={}  # Initialise as an empty dictionary for subgroup support
        )
        return company

__init__(companies)

Create companies and provide functionality to allocate workers.

Parameters

company_size_per_superarea_df: pd.DataFram Nr. of companies within a size-range per SuperArea.

pd.DataFrame

Nr. of companies per sector sector per SuperArea.

Source code in june/groups/company.py

def __init__(self, companies: List["Companies"]):
    """
    Create companies and provide functionality to allocate workers.

    Parameters
    ----------
    company_size_per_superarea_df: pd.DataFram
        Nr. of companies within a size-range per SuperArea.

    compsec_per_msoa_df: pd.DataFrame
        Nr. of companies per sector sector per SuperArea.
    """
    super().__init__(members=companies)

create_companies_in_super_area(super_area, company_sizes, company_sectors) classmethod

Creates companies in a super area using the sizes and sectors distributions.

Parameters:

Name	Type	Description	Default
super_area	SuperArea		required
company_sizes			required
company_sectors			required

Source code in june/groups/company.py

@classmethod
def create_companies_in_super_area(
    cls, super_area: SuperArea, company_sizes, company_sectors
) -> list:
    """Creates companies in a super area using the sizes and sectors distributions.

    Args:
        super_area (SuperArea): 
        company_sizes: 
        company_sectors: 

    """
    sizes = np.array([])
    for size_bracket, counts in company_sizes.items():
        size_min, size_max = _get_size_brackets(size_bracket)
        sizes = np.concatenate(
            (sizes, np.random.randint(max(size_min, 1), size_max, int(counts)))
        )
    np.random.shuffle(sizes)
    sectors = []
    for sector, counts in company_sectors.items():
        sectors += [sector] * int(counts)
    shuffle(sectors)
    companies = list(
        map(
            lambda company_size, company_sector: cls.create_company(
                super_area, company_size, company_sector
            ),
            sizes,
            sectors,
        )
    )
    return companies

create_company(super_area, company_size, company_sector) classmethod

Create a company instance.

Parameters:

Name	Type	Description	Default
super_area	SuperArea	The area the company belongs to.	required
company_size	int	Maximum number of workers the company can accommodate.	required
company_sector	str	The sector the company belongs to.	required

Returns:

Name	Type	Description
Company		A new company instance.

Source code in june/groups/company.py

@classmethod
def create_company(cls, super_area, company_size, company_sector):
    """Create a company instance.

    Args:
        super_area (SuperArea): The area the company belongs to.
        company_size (int): Maximum number of workers the company can accommodate.
        company_sector (str): The sector the company belongs to.

    Returns:
        Company: A new company instance.

    """
    company = cls.venue_class(
        super_area=super_area,
        n_workers_max=company_size,
        sector=company_sector,
        registered_members_ids={}  # Initialise as an empty dictionary for subgroup support
    )
    return company

for_geography(geography, size_nr_file=None, sector_nr_per_msoa_file=None, sct_companies_file=None, ew_companies_file=None, ni_companies_file=None) classmethod

Creates companies for the specified geography, and saves them to the super_areas they belong to.

Parameters:

Name	Type	Description	Default
geography	Geography		required
size_nr_file	str	(Default value = None)	None
sector_nr_per_msoa_file	str	(Default value = None)	None
sct_companies_file	str	(Default value = None)	None
ew_companies_file	str	(Default value = None)	None
ni_companies_file	str	(Default value = None)	None

Source code in june/groups/company.py

@classmethod
def for_geography(
    cls,
    geography: Geography,
    size_nr_file: str = None,
    sector_nr_per_msoa_file: str = None,
    sct_companies_file: str = None,
    ew_companies_file: str = None,
    ni_companies_file: str = None
) -> "Companies":
    """Creates companies for the specified geography, and saves them
    to the super_areas they belong to.

    Args:
        geography (Geography): 
        size_nr_file (str, optional): (Default value = None)
        sector_nr_per_msoa_file (str, optional): (Default value = None)
        sct_companies_file (str, optional): (Default value = None)
        ew_companies_file (str, optional): (Default value = None)
        ni_companies_file (str, optional): (Default value = None)

    """
    if not geography.super_areas:
        raise CompanyError("Empty geography!")
    # After creating the companies
    companies = cls.for_super_areas(
        geography.super_areas,
        size_nr_file,
        sector_nr_per_msoa_file,
        sct_companies_file,
        ew_companies_file,
        ni_companies_file
    )
    logger.info(f"There are {len(companies)} companies in this geography.")


    # Sample 5 companies from each super area for visualization
    sampled_companies = []
    for super_area in geography.super_areas:
        if hasattr(super_area, 'companies') and super_area.companies:
            # Sample 5 companies or fewer if there are less than 5
            sample_companies = random.sample(super_area.companies, min(5, len(super_area.companies)))
            for company in sample_companies:
                sampled_companies.append({
                    "| Company ID": company.id,
                    "| Super Area": super_area.name,
                    "| Company Sector": company.sector,
                    "| Number of Workers": company.n_workers,
                    "| Coordinates": company.coordinates,
                    "| Max Workers": company.n_workers_max
                })

    # Convert the sample data to a DataFrame
    df_companies = pd.DataFrame(sampled_companies)
    print("\n===== Sample of Created Companies =====")
    print(df_companies)

    return companies

for_super_areas(super_areas, size_nr_per_super_area_file=None, sector_nr_per_super_area_file=None, sct_companies_file=None, ew_companies_file=None, ni_companies_file=None) classmethod

Creates companies for the specified super_areas, and saves them to the super_areas they belong to.

Parameters:

Name	Type	Description	Default
super_areas	List[SuperArea]		required
size_nr_per_super_area_file	str	(Default value = None)	None
sector_nr_per_super_area_file	str	(Default value = None)	None
sct_companies_file	str	(Default value = None)	None
ew_companies_file	str	(Default value = None)	None
ni_companies_file	str	(Default value = None)	None

Source code in june/groups/company.py

@classmethod
def for_super_areas(
    cls,
    super_areas: List[SuperArea],
    size_nr_per_super_area_file: str = None,
    sector_nr_per_super_area_file: str = None,
    sct_companies_file: str = None,
    ew_companies_file: str = None,
    ni_companies_file: str = None
    ) -> "Companies":
    """Creates companies for the specified super_areas, and saves them
    to the super_areas they belong to.

    Args:
        super_areas (List[SuperArea]): 
        size_nr_per_super_area_file (str, optional): (Default value = None)
        sector_nr_per_super_area_file (str, optional): (Default value = None)
        sct_companies_file (str, optional): (Default value = None)
        ew_companies_file (str, optional): (Default value = None)
        ni_companies_file (str, optional): (Default value = None)

    """
    # Always try to use companies files first (auto-detect), fallback to old CSV method
    try:
        return cls.from_companies_file(super_areas, sct_companies_file, ew_companies_file, ni_companies_file)
    except (FileNotFoundError, KeyError, pd.errors.EmptyDataError):
        # Fallback to old CSV-based method if companies files don't exist or are empty
        logger.info("Companies files not found or empty, falling back to legacy CSV method")

    # Check if legacy CSV files are provided
    if size_nr_per_super_area_file is None or sector_nr_per_super_area_file is None:
        raise CompanyError("No company files available. Please provide either companies files or legacy CSV files.")

    size_per_superarea_df = pd.read_csv(size_nr_per_super_area_file, index_col=0)
    sector_per_superarea_df = pd.read_csv(
        sector_nr_per_super_area_file, index_col=0
    )
    super_area_names = [super_area.name for super_area in super_areas]
    company_sizes_per_super_area = size_per_superarea_df.loc[super_area_names]
    company_sectors_per_super_area = sector_per_superarea_df.loc[super_area_names]
    assert len(company_sectors_per_super_area) == len(company_sizes_per_super_area)
    if len(company_sectors_per_super_area) == 1:
        super_area = super_areas[0]
        companies = cls.create_companies_in_super_area(
            super_area, company_sizes_per_super_area, company_sectors_per_super_area
        )
        super_area.companies = companies
    else:
        companies = []
        for super_area, (_, company_sizes), (_, company_sectors) in zip(
            super_areas,
            company_sizes_per_super_area.iterrows(),
            company_sectors_per_super_area.iterrows(),
        ):
            super_area.companies = cls.create_companies_in_super_area(
                super_area, company_sizes, company_sectors
            )
            companies += super_area.companies


    return cls(companies)

from_companies_file(super_areas, sct_companies_file=None, ew_companies_file=None, ni_companies_file=None) classmethod

Optimized company creation from CSV files with pre-filtering and vectorized operations. CSV format: company_id,super_area,industry_code,size_band,employee_count

Supports mixed geographies by loading from SCT, EW, and NI files as needed: - Scotland super areas (S02): Uses SCT file - England/Wales super areas (E02, W02): Uses EW file - Northern Ireland super areas (N): Uses NI file

Parameters:

Name	Type	Description	Default
super_areas	List[SuperArea]		required
sct_companies_file	str	(Default value = None)	None
ew_companies_file	str	(Default value = None)	None
ni_companies_file	str	(Default value = None)	None

Source code in june/groups/company.py

@classmethod
def from_companies_file(
    cls,
    super_areas: List[SuperArea],
    sct_companies_file: str = None,
    ew_companies_file: str = None,
    ni_companies_file: str = None
) -> "Companies":
    """Optimized company creation from CSV files with pre-filtering and vectorized operations.
    CSV format: company_id,super_area,industry_code,size_band,employee_count

    Supports mixed geographies by loading from SCT, EW, and NI files as needed:
    - Scotland super areas (S02*): Uses SCT file
    - England/Wales super areas (E02*, W02*): Uses EW file
    - Northern Ireland super areas (N*): Uses NI file

    Args:
        super_areas (List[SuperArea]): 
        sct_companies_file (str, optional): (Default value = None)
        ew_companies_file (str, optional): (Default value = None)
        ni_companies_file (str, optional): (Default value = None)

    """
    super_area_names = [super_area.name for super_area in super_areas]

    # Separate areas by region
    scotland_areas = [name for name in super_area_names if name.startswith('S02')]
    ew_areas = [name for name in super_area_names if name.startswith(('E02', 'W02'))]
    ni_areas = [name for name in super_area_names if name.startswith('N')]

    companies_df_list = []

    # Optimized CSV loading with pre-filtering
    def load_companies_optimized(file_path, target_areas, region_name):
        """

        Args:
            file_path: 
            target_areas: 
            region_name: 

        """
        if not target_areas:
            return None

        logger.info(f"Loading {region_name} companies from: {file_path}")
        try:
            # Read with chunking for memory efficiency on large files
            chunk_size = 50000
            filtered_chunks = []

            for chunk in pd.read_csv(file_path, chunksize=chunk_size):
                # Pre-filter chunk to only include our target areas (major optimization!)
                filtered_chunk = chunk[chunk['super_area'].isin(target_areas)]
                if not filtered_chunk.empty:
                    filtered_chunks.append(filtered_chunk)

            if not filtered_chunks:
                logger.warning(f"No companies found for {region_name} areas in {file_path}")
                return None

            result_df = pd.concat(filtered_chunks, ignore_index=True)
            logger.info(f"Found {len(result_df)} {region_name} companies for {len(target_areas)} areas")
            return result_df

        except FileNotFoundError:
            logger.warning(f"{region_name} companies file not found: {file_path}")
            return None

    # Load each region with optimized filtering
    if scotland_areas:
        sct_file = sct_companies_file if sct_companies_file else default_sct_companies_file
        sct_df = load_companies_optimized(sct_file, scotland_areas, "SCT")
        if sct_df is not None:
            companies_df_list.append(sct_df)

    if ew_areas:
        ew_file = ew_companies_file if ew_companies_file else default_ew_companies_file
        ew_df = load_companies_optimized(ew_file, ew_areas, "EW")
        if ew_df is not None:
            companies_df_list.append(ew_df)

    if ni_areas:
        ni_file = ni_companies_file if ni_companies_file else default_ni_companies_file
        ni_df = load_companies_optimized(ni_file, ni_areas, "NI")
        if ni_df is not None:
            companies_df_list.append(ni_df)

    # Combine all company dataframes
    if not companies_df_list:
        raise FileNotFoundError("No company files found or loaded successfully")

    companies_df = pd.concat(companies_df_list, ignore_index=True)
    logger.info(f"Total companies loaded: {len(companies_df)} for {len(super_area_names)} super areas")

    # MAJOR OPTIMIZATION: Use groupby instead of nested loops
    companies = []

    # Group companies by super_area for vectorized processing  
    companies_by_area = companies_df.groupby('super_area')

    for super_area in super_areas:
        try:
            area_companies_df = companies_by_area.get_group(super_area.name)
        except KeyError:
            # No companies for this super area
            super_area.companies = []
            continue

        # Vectorized company creation using apply or list comprehension
        super_area_companies = []

        # Use itertuples for faster iteration (3x faster than iterrows)
        for row in area_companies_df.itertuples(index=False):
            company = cls.venue_class(
                super_area=super_area,
                n_workers_max=row.employee_count,
                sector=str(row.industry_code),
                registered_members_ids={}
            )
            super_area_companies.append(company)
            companies.append(company)

        super_area.companies = super_area_companies
        logger.debug(f"Created {len(super_area_companies)} companies for super area {super_area.name}")

    return cls(companies)

from_sct_companies_file(super_areas, sct_companies_file=None) classmethod

Legacy method for backward compatibility. Creates companies from SCT companies CSV file.

Parameters:

Name	Type	Description	Default
super_areas	List[SuperArea]		required
sct_companies_file	str	(Default value = None)	None

Source code in june/groups/company.py

@classmethod
def from_sct_companies_file(
    cls,
    super_areas: List[SuperArea],
    sct_companies_file: str = None
) -> "Companies":
    """Legacy method for backward compatibility.
    Creates companies from SCT companies CSV file.

    Args:
        super_areas (List[SuperArea]): 
        sct_companies_file (str, optional): (Default value = None)

    """
    return cls.from_companies_file(super_areas, sct_companies_file=sct_companies_file)

Company

Bases: Group

The Company class represents a company that contains information about its workers which are not yet distributed to key company sectors (e.g. as schools and hospitals).

Currently we treat the workforce of a company as one single sub-group and therefore we invoke the base class group with the default Ngroups = 1. We made this explicit here, although it is not necessary.

Source code in june/groups/company.py

class Company(Group):
    """The Company class represents a company that contains information about
    its workers which are not yet distributed to key company sectors
    (e.g. as schools and hospitals).

    Currently we treat the workforce of a company as one single sub-group
    and therefore we invoke the base class group with the default Ngroups = 1.
    We made this explicit here, although it is not necessary.

    """

    __slots__ = ("super_area", "sector", "n_workers_max")

    def __init__(self, super_area=None, n_workers_max=np.inf, sector=None, registered_members_ids=None):
        """
        Initialise a Company instance.

        Parameters
        ----------
        disease_config : DiseaseConfig
            Configuration object for the disease.
        super_area : str, optional
            The area the company belongs to.
        n_workers_max : int, optional
            Maximum number of workers the company can accommodate (default is np.inf).
        sector : str, optional
            The sector the company belongs to.
        registered_members_ids : dict, optional
            A dict mapping subgroup IDs to lists of member IDs.
        """
        # Initialise the base Group class with disease_config
        super().__init__()

        # Assign attributes specific to Company
        self.super_area = super_area
        self.sector = sector
        self.n_workers_max = n_workers_max

        # Initialise registered_members_ids as a dictionary
        self.registered_members_ids = registered_members_ids if registered_members_ids is not None else {}

    def add(self, person):
        """

        Args:
            person: 

        """
        super().add(
            person,
            subgroup_type=self.get_index_subgroup(person),
            activity="primary_activity",
        )

    def add_to_registered_members(self, person_id, subgroup_type=0):
        """Add a person to the registered members list for a specific subgroup.

        Args:
            person_id (int): The ID of the person to add
            subgroup_type (int, optional, optional): The subgroup to add the person to (default: 0)

        """
        # Create the subgroup if it doesn't exist
        if subgroup_type not in self.registered_members_ids:
            self.registered_members_ids[subgroup_type] = []

        # Add the person if not already in the list
        if person_id not in self.registered_members_ids[subgroup_type]:
            self.registered_members_ids[subgroup_type].append(person_id)

    @property
    def n_workers(self):
        """ """
        return len(self.people)

    # @property
    # def workers(self):
    #     return self.subgroups[self.SubgroupType.workers]

    @property
    def coordinates(self):
        """ """
        return self.super_area.coordinates

    @property
    def area(self):
        """ """
        return self.super_area.areas[0]

    def get_interactive_group(self, people_from_abroad=None):
        """

        Args:
            people_from_abroad: (Default value = None)

        """
        return InteractiveCompany(self, people_from_abroad=people_from_abroad)

area property

coordinates property

n_workers property

__init__(super_area=None, n_workers_max=np.inf, sector=None, registered_members_ids=None)

Initialise a Company instance.

Parameters

disease_config : DiseaseConfig Configuration object for the disease. super_area : str, optional The area the company belongs to. n_workers_max : int, optional Maximum number of workers the company can accommodate (default is np.inf). sector : str, optional The sector the company belongs to. registered_members_ids : dict, optional A dict mapping subgroup IDs to lists of member IDs.

Source code in june/groups/company.py

def __init__(self, super_area=None, n_workers_max=np.inf, sector=None, registered_members_ids=None):
    """
    Initialise a Company instance.

    Parameters
    ----------
    disease_config : DiseaseConfig
        Configuration object for the disease.
    super_area : str, optional
        The area the company belongs to.
    n_workers_max : int, optional
        Maximum number of workers the company can accommodate (default is np.inf).
    sector : str, optional
        The sector the company belongs to.
    registered_members_ids : dict, optional
        A dict mapping subgroup IDs to lists of member IDs.
    """
    # Initialise the base Group class with disease_config
    super().__init__()

    # Assign attributes specific to Company
    self.super_area = super_area
    self.sector = sector
    self.n_workers_max = n_workers_max

    # Initialise registered_members_ids as a dictionary
    self.registered_members_ids = registered_members_ids if registered_members_ids is not None else {}

add(person)

Parameters:

Name	Type	Description	Default
person			required

Source code in june/groups/company.py

def add(self, person):
    """

    Args:
        person: 

    """
    super().add(
        person,
        subgroup_type=self.get_index_subgroup(person),
        activity="primary_activity",
    )

add_to_registered_members(person_id, subgroup_type=0)

Add a person to the registered members list for a specific subgroup.

Parameters:

Name	Type	Description	Default
person_id	int	The ID of the person to add	required
subgroup_type	(int, optional)	The subgroup to add the person to (default: 0)	0

Source code in june/groups/company.py

def add_to_registered_members(self, person_id, subgroup_type=0):
    """Add a person to the registered members list for a specific subgroup.

    Args:
        person_id (int): The ID of the person to add
        subgroup_type (int, optional, optional): The subgroup to add the person to (default: 0)

    """
    # Create the subgroup if it doesn't exist
    if subgroup_type not in self.registered_members_ids:
        self.registered_members_ids[subgroup_type] = []

    # Add the person if not already in the list
    if person_id not in self.registered_members_ids[subgroup_type]:
        self.registered_members_ids[subgroup_type].append(person_id)

get_interactive_group(people_from_abroad=None)

Parameters:

Name	Type	Description	Default
people_from_abroad		(Default value = None)	None

Source code in june/groups/company.py

def get_interactive_group(self, people_from_abroad=None):
    """

    Args:
        people_from_abroad: (Default value = None)

    """
    return InteractiveCompany(self, people_from_abroad=people_from_abroad)

CompanyError

Bases: BaseException

Source code in june/groups/company.py

class CompanyError(BaseException):
    """ """
    pass

InteractiveCompany

Bases: InteractiveGroup

Source code in june/groups/company.py

class InteractiveCompany(InteractiveGroup):
    """ """
    sector_betas = _read_sector_betas()

    def __init__(self, group: "Group", people_from_abroad=None):
        super().__init__(group=group, people_from_abroad=people_from_abroad)
        self.sector = group.sector


    def get_processed_beta(self, betas, beta_reductions):
        """

        Args:
            betas: 
            beta_reductions: 

        """
        beta_processed = super().get_processed_beta(
            betas=betas, beta_reductions=beta_reductions
        )
        return beta_processed * self.sector_betas.get(self.sector, 1.0)

get_processed_beta(betas, beta_reductions)

Parameters:

Name	Type	Description	Default
betas			required
beta_reductions			required

Source code in june/groups/company.py

def get_processed_beta(self, betas, beta_reductions):
    """

    Args:
        betas: 
        beta_reductions: 

    """
    beta_processed = super().get_processed_beta(
        betas=betas, beta_reductions=beta_reductions
    )
    return beta_processed * self.sector_betas.get(self.sector, 1.0)